1.首先咱们简单了解一下消息中间件的应用场景javascript
场景说明:用户注册后,须要发注册邮件和注册短信,传统的作法有两种1.串行的方式;2.并行的方式
(1)串行方式:将注册信息写入数据库后,发送注册邮件,再发送注册短信,以上三个任务所有完成后才返回给客户端。 这有一个问题是,邮件,短信并非必须的,它只是一个通知,而这种作法让客户端等待没有必要等待的东西.
(2)并行方式:将注册信息写入数据库后,发送邮件的同时,发送短信,以上三个任务完成后,返回给客户端,并行的方式能提升处理的时间。
假设三个业务节点分别使用50ms,串行方式使用时间150ms,并行使用时间100ms。虽然并性已经提升的处理时间,可是,前面说过,邮件和短信对我正常的使用网站没有任何影响,客户端没有必要等着其发送完成才显示注册成功,英爱是写入数据库后就返回.
(3)消息队列
引入消息队列后,把发送邮件,短信不是必须的业务逻辑异步处理
由此能够看出,引入消息队列后,用户的响应时间就等于写入数据库的时间+写入消息队列的时间(能够忽略不计),引入消息队列后处理后,响应时间是串行的3倍,是并行的2倍。前端
场景:双11是购物狂节,用户下单后,订单系统须要通知库存系统,传统的作法就是订单系统调用库存系统的接口.
这种作法有一个缺点:java
订单系统和库存系统高耦合.
引入消息队列
spring
订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功。数据库
流量削峰通常在秒杀活动中应用普遍
场景:秒杀活动,通常会由于流量过大,致使应用挂掉,为了解决这个问题,通常在应用前端加入消息队列。
做用:
1.能够控制活动人数,超过此必定阀值的订单直接丢弃(我为何秒杀一次都没有成功过呢^^)
2.能够缓解短期的高流量压垮应用(应用程序按本身的最大处理能力获取订单)
1.用户的请求,服务器收到以后,首先写入消息队列,加入消息队列长度超过最大值,则直接抛弃用户请求或跳转到错误页面.
springboot
2.秒杀业务根据消息队列中的请求信息,再作后续处理.服务器
以上内容的来源是:https://blog.csdn.net/whoamiyang/article/details/54954780,在此感谢并发
2.各类消息中间件性能的比较:app
TPS比较 一ZeroMq 最好,RabbitMq 次之, ActiveMq 最差。
负载均衡
持久化消息比较—zeroMq不支持,activeMq和rabbitMq都支持。持久化消息主要是指:MQ down或者MQ所在的服务器down了,消息不会丢失的机制。
可靠性、灵活的路由、集群、事务、高可用的队列、消息排序、问题追踪、可视化管理工具、插件系统、社区—RabbitMq最好,ActiveMq次之,ZeroMq最差。
高并发—从实现语言来看,RabbitMQ最高,缘由是它的实现语言是天生具有高并发高可用的erlang语言。
综上所述:RabbitMQ的性能相对来讲更好更全面,是消息中间件的首选。
3.接下来咱们在springboot当中整合使用RabbitMQ
第一步:导入maven依赖
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> <version>1.5.2.RELEASE</version> </dependency>
第二步:在application.properties文件当中引入RabbitMQ基本的配置信息
#对于rabbitMQ的支持 spring.rabbitmq.host=127.0.0.1 spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest
第三步:编写RabbitConfig类,类里面设置不少个EXCHANGE,QUEUE,ROUTINGKEY,是为了接下来的不一样使用场景。
/** Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线,保证数据能按照指定的方式进行传输, Exchange:消息交换机,它指定消息按什么规则,路由到哪一个队列。 Queue:消息的载体,每一个消息都会被投到一个或多个队列。 Binding:绑定,它的做用就是把exchange和queue按照路由规则绑定起来. Routing Key:路由关键字,exchange根据这个关键字进行消息投递。 vhost:虚拟主机,一个broker里能够有多个vhost,用做不一样用户的权限分离。 Producer:消息生产者,就是投递消息的程序. Consumer:消息消费者,就是接受消息的程序. Channel:消息通道,在客户端的每一个链接里,可创建多个channel. */
@Configuration public class RabbitConfig { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @Value("${spring.rabbitmq.host}") private String host; @Value("${spring.rabbitmq.port}") private int port; @Value("${spring.rabbitmq.username}") private String username; @Value("${spring.rabbitmq.password}") private String password; public static final String EXCHANGE_A = "my-mq-exchange_A"; public static final String EXCHANGE_B = "my-mq-exchange_B"; public static final String EXCHANGE_C = "my-mq-exchange_C"; public static final String QUEUE_A = "QUEUE_A"; public static final String QUEUE_B = "QUEUE_B"; public static final String QUEUE_C = "QUEUE_C"; public static final String ROUTINGKEY_A = "spring-boot-routingKey_A"; public static final String ROUTINGKEY_B = "spring-boot-routingKey_B"; public static final String ROUTINGKEY_C = "spring-boot-routingKey_C"; @Bean public ConnectionFactory connectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); connectionFactory.setVirtualHost("/"); connectionFactory.setPublisherConfirms(true); return connectionFactory; } @Bean @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) //必须是prototype类型 public RabbitTemplate rabbitTemplate() { RabbitTemplate template = new RabbitTemplate(connectionFactory()); return template; } }
第四步:编写消息的生产者
@Component public class MsgProducer implements RabbitTemplate.ConfirmCallback { private final Logger logger = LoggerFactory.getLogger(this.getClass()); //因为rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE,因此不能自动注入 private RabbitTemplate rabbitTemplate; /** * 构造方法注入rabbitTemplate */ @Autowired public MsgProducer(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; rabbitTemplate.setConfirmCallback(this); //rabbitTemplate若是为单例的话,那回调就是最后设置的内容 } public void sendMsg(String content) { CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString()); //把消息放入ROUTINGKEY_A对应的队列当中去,对应的是队列A rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A, RabbitConfig.ROUTINGKEY_A, content, correlationId); } /** * 回调 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { logger.info(" 回调id:" + correlationData); if (ack) { logger.info("消息成功消费"); } else { logger.info("消息消费失败:" + cause); } } }
第五步:把交换机,队列,经过路由关键字进行绑定,写在RabbitConfig类当中
/** * 针对消费者配置 * 1. 设置交换机类型 * 2. 将队列绑定到交换机 FanoutExchange: 将消息分发到全部的绑定队列,无routingkey的概念 HeadersExchange :经过添加属性key-value匹配 DirectExchange:按照routingkey分发到指定队列 TopicExchange:多关键字匹配 */ @Bean public DirectExchange defaultExchange() { return new DirectExchange(EXCHANGE_A); }
/** * 获取队列A * @return */ @Bean public Queue queueA() { return new Queue(QUEUE_A, true); //队列持久 }
@Bean public Binding binding() { return BindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A); }
一个交换机能够绑定多个消息队列,也就是消息经过一个交换机,能够分发到不一样的队列当中去。
@Bean public Binding binding() { return BindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A); } @Bean public Binding bindingB(){ return BindingBuilder.bind(queueB()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_B); }
第六步:编写消息的消费者,这一步也是最复杂的,由于能够编写出不少不一样的需求出来,写法也有不少的不一样。
好比一个生产者,一个消费者
@Component @RabbitListener(queues = RabbitConfig.QUEUE_A) public class MsgReceiver { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @RabbitHandler public void process(String content) { logger.info("接收处理队列A当中的消息: " + content); } }
好比一个生产者,多个消费者,能够写多个消费者,而且他们的分发是负载均衡的。
@Component @RabbitListener(queues = RabbitConfig.QUEUE_A) public class MsgReceiverC_one { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @RabbitHandler public void process(String content) { logger.info("处理器one接收处理队列A当中的消息: " + content); } }
@Component @RabbitListener(queues = RabbitConfig.QUEUE_A) public class MsgReceiverC_two { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @RabbitHandler public void process(String content) { logger.info("处理器two接收处理队列A当中的消息: " + content); } }
另一种消息处理机制的写法以下,在RabbitMQConfig类里面增长bean:
@Bean public SimpleMessageListenerContainer messageContainer() { //加载处理消息A的队列 SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory()); //设置接收多个队列里面的消息,这里设置接收队列A //假如想一个消费者处理多个队列里面的信息能够以下设置: //container.setQueues(queueA(),queueB(),queueC()); container.setQueues(queueA()); container.setExposeListenerChannel(true); //设置最大的并发的消费者数量 container.setMaxConcurrentConsumers(10); //最小的并发消费者的数量 container.setConcurrentConsumers(1); //设置确认模式手工确认 container.setAcknowledgeMode(AcknowledgeMode.MANUAL); container.setMessageListener(new ChannelAwareMessageListener() { @Override public void onMessage(Message message, Channel channel) throws Exception { /**经过basic.qos方法设置prefetch_count=1,这样RabbitMQ就会使得每一个Consumer在同一个时间点最多处理一个Message, 换句话说,在接收到该Consumer的ack前,它不会将新的Message分发给它 */ channel.basicQos(1); byte[] body = message.getBody(); logger.info("接收处理队列A当中的消息:" + new String(body)); /**为了保证永远不会丢失消息,RabbitMQ支持消息应答机制。 当消费者接收到消息并完成任务后会往RabbitMQ服务器发送一条确认的命令,而后RabbitMQ才会将消息删除。*/ channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } }); return container; }
下面是当一个消费者,处理多个队列里面的信息打印的log
Fanout Exchange
Fanout 就是咱们熟悉的广播模式,给Fanout交换机发送消息,绑定了这个交换机的全部队列都收到这个消息。
//配置fanout_exchange @Bean FanoutExchange fanoutExchange() { return new FanoutExchange(RabbitConfig.FANOUT_EXCHANGE); } //把全部的队列都绑定到这个交换机上去 @Bean Binding bindingExchangeA(Queue queueA,FanoutExchange fanoutExchange) { return BindingBuilder.bind(queueA).to(fanoutExchange); } @Bean Binding bindingExchangeB(Queue queueB, FanoutExchange fanoutExchange) { return BindingBuilder.bind(queueB).to(fanoutExchange); } @Bean Binding bindingExchangeC(Queue queueC, FanoutExchange fanoutExchange) { return BindingBuilder.bind(queueC).to(fanoutExchange); }
消息发送,这里不设置routing_key,由于设置了也无效,发送端的routing_key写任何字符都会被忽略。
public void sendAll(String content) { rabbitTemplate.convertAndSend("fanoutExchange","", content); }
消息处理的结果以下所示: