RabbitMQ

<!--2. rabbitmq--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>

spring: rabbitmq: host: 192.168.88.131 port: 5672 username: guest password: guest virtual-host: /

@Configuration public class RabbitMQConfig { public static final String EXCHANGE_NAME = "boot_topic_exchange"; public static final String QUEUE_NAME = "boot_queue"; // 1 交换机 @Bean("bootExchange") public Exchange bootExchange(){ return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build(); } //2.Queue 队列 @Bean("bootQueue") public Queue bootQueue(){ return QueueBuilder.durable(QUEUE_NAME).build(); } //3. 队列和交互机绑定关系 Binding /* 1. 知道哪个队列 2. 知道哪个交换机 3. routing key noargs()：表示不指定参数 */ @Bean public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs(); } }

<!--RabbitMQ 启动依赖--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>

spring: rabbitmq: host: 192.168.88.131 port: 5672 username: guest password:guest virtual-host: /

@Component public class RabbimtMQListener { @RabbitListener(queues = "boot_queue") public void listenerQueue(Message message){ System.out.println(new String(message.getBody())); } }

传统模式的缺点

中间件模式的的优点

应用场景

应用场景

传统模式

中间件模式

削峰填谷

每秒的响应请求数，也即是最大吞吐能力。

QPS = 并发量 / 平均响应时间

并发量 = QPS * 平均响应时间

原理：每天80%的访问集中在20%的时间里，这20%时间叫做峰值时间。

公式：( 总PV数 * 80% ) / ( 每天秒数 * 20% ) = 峰值时间每秒请求数(QPS) 。

机器：峰值时间每秒QPS / 单台机器的QPS = 需要的机器 。

AMQP是一种高级消息队列协议。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。

JMS即Java消息服务应用程序接口，是一个Java平台中关于面向消息中间件的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

JMS是定义了统一的接口，来对消息操作进行统一； AMQP是通过规定协议来统一数据交互的格式

JMS限定了必须使用Java语言； AMQP只是协议，不规定实现方式，因此是跨语言的。

JMS规定了两种消息模式； 而AMQP的消息模式更加丰富

① 创建工程（生产者、消费者）

② 分别添加依赖

③ 编写生产者发送消息

④ 编写消费者接收消息

分支主题

多个消费端共同消费同一个队列中的消息。

应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

生产者

消费者1

消费者2

小结

交换机

简介

生产者

消费者1

消费者2

小结：交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。

发布订阅模式与工作队列模式的区别

路由模式特点

分支主题

生产者

消费者1

消费者2

小结：Routing模式要求队列在绑定交换机时要指定routing key，消息会转发到符合routing key的队列。

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候**使用通配符**！

通配符规则

分支主题

生产者

消费者1

消费者2

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活。

<dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.1.8.RELEASE</version> </dependency>

rabbitmq.host=192.168.88.131 rabbitmq.port=5672 rabbitmq.username=guest rabbitmq.password=guest rabbitmq.virtual-host=/

<!--加载配置文件--> <context:property-placeholder location="classpath:rabbitmq.properties"/> <!-- 定义rabbitmq connectionFactory --> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.virtual-host}"/> <!—定义RabbitAdmin对象，用于管理交换机、队列--> <rabbit:admin connection-factory="connectionFactory"/> <!--定义持久化队列,不存在则自动创建；不绑定到交换机则绑定到默认交换机 默认交换机类型为direct,名字为："",路由键为队列的名称 --> <rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/> <!-- ~~~~~~~~~~~~~~广播；所有队列都能收到消息~~~~~~~~~~~~~ --> <!--定义广播交换机中的持久化队列,不存在则自动创建--> <rabbit:queue id="spring_fanout_queue_11" name="spring_fanout_queue_1" auto-declare="true"/> <!--定义广播交换机中的持久化队列,不存在则自动创建--> <rabbit:queue id="spring_fanout_queue_22" name="spring_fanout_queue_2" auto-declare="true"/> <!--定义广播类型交换机；并绑定上述两个队列--> <rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true"> <rabbit:bindings> <rabbit:binding queue="spring_fanout_queue_11"/> <rabbit:binding queue="spring_fanout_queue_22"/> </rabbit:bindings> </rabbit:fanout-exchange> <!-- ~~~~~~~~~~~~~~~通配符；*匹配一个单词,#匹配多个单词 ~~~~~~~~~~~~~ --> <!--定义主题交换机中的持久化队列,不存在则自动创建--> <rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/> <!--定义主题交换机中的持久化队列,不存在则自动创建--> <rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/> <!--定义主题交换机中的持久化队列,不存在则自动创建--> <rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/> <rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true"> <rabbit:bindings> <rabbit:binding pattern="atguigu.*" queue="spring_topic_queue_star"/> <rabbit:binding pattern="atguigu.#" queue="spring_topic_queue_well"/> <rabbit:binding pattern="guigu.#" queue="spring_topic_queue_well2"/> </rabbit:bindings> </rabbit:topic-exchange> <!--定义rabbitTemplate对象操作可以在代码中方便发送消息--> <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>

@RunWith(SpringRunner.class) @ContextConfiguration(locations = "classpath:spring-rabbitmq.xml") public class ProducerTest { @Autowired private RabbitTemplate rabbitTemplate; /** * 只发队列消息 * 默认交换机类型为 direct * 交换机的名称为空,路由键为队列的名称 */ @Test public void queueTest(){ //路由键与队列同名 rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消息。"); } /** * 发送广播 * 交换机类型为 fanout * 绑定到该交换机的所有队列都能够收到消息 */ @Test public void fanoutTest(){ /** * 参数1：交换机名称 * 参数2：路由键名（广播设置为空） * 参数3：发送的消息内容 */ rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到spring_fanout_exchange交换机的广播消息"); } /** * 通配符 * 交换机类型为 topic * 匹配路由键的通配符,*表示一个单词,#表示多个单词 * 绑定到该交换机的匹配队列能够收到对应消息 */ @Test public void topicTest(){ /** * 参数1：交换机名称 * 参数2：路由键名 * 参数3：发送的消息内容 */ rabbitTemplate.convertAndSend("spring_topic_exchange", "atguigu.bj", "发送到spring_topic_exchange交换机atguigu.bj的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "atguigu.bj.1", "发送到spring_topic_exchange交换机atguigu.bj.1的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "atguigu.bj.2", "发送到spring_topic_exchange交换机atguigu.bj.2的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "guigu.cn", "发送到spring_topic_exchange交换机guigu.cn的消息"); } }

<dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.1.8.RELEASE</version> </dependency>

rabbitmq.host=192.168.6.100 rabbitmq.port=5672 rabbitmq.username=admin rabbitmq.password=123456 rabbitmq.virtual-host=/

<!--加载配置文件--> <context:property-placeholder location="classpath:rabbitmq.properties"/> <!-- 定义rabbitmq connectionFactory --> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.virtual-host}"/> <bean id="springQueueListener" class="com.atguigu.rabbitmq.listener.SpringQueueListener"/> <bean id="fanoutListener1" class="com.atguigu.rabbitmq.listener.FanoutListener1"/> <bean id="fanoutListener2" class="com.atguigu.rabbitmq.listener.FanoutListener2"/> <bean id="topicListenerStar" class="com.atguigu.rabbitmq.listener.TopicListenerStar"/> <bean id="topicListenerWell" class="com.atguigu.rabbitmq.listener.TopicListenerWell"/> <bean id="topicListenerWell2" class="com.atguigu.rabbitmq.listener.TopicListenerWell2"/> . <rabbit:listener ref="springQueueListener" queue-names="spring_queue"/> <rabbit:listener ref="fanoutListener1" queue-names="spring_fanout_queue_1"/> <rabbit:listener ref="fanoutListener2" queue-names="spring_fanout_queue_2"/> <rabbit:listener ref="topicListenerStar" queue-names="spring_topic_queue_star"/> <rabbit:listener ref="topicListenerWell" queue-names="spring_topic_queue_well"/> <rabbit:listener ref="topicListenerWell2" queue-names="spring_topic_queue_well2"/> </rabbit:listener-container>

队列监听器

广播监听器1

广播监听器2

星号通配符监听器

井号通配符监听器

井号通配符监听器2

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。

confirm 确认模式

return 退回模式

<dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.1.8.RELEASE</version> </dependency>

rabbitmq.host=192.168.6.100 rabbitmq.port=5672 rabbitmq.username=admin rabbitmq.password=123456 rabbitmq.virtual-host=/

!--加载配置文件--> <context:property-placeholder location="classpath:rabbitmq.properties"/> <!-- 定义rabbitmq connectionFactory 确认模式开启：publisher-confirms="true" --> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.virtual-host}" publisher-confirms="true" publisher-returns="true" /> <!--定义管理交换机、队列--> <rabbit:admin connection-factory="connectionFactory"/> <!--定义rabbitTemplate对象操作可以在代码中方便发送消息--> <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/> <!--消息可靠性投递（生产端）--> <rabbit:queue id="test_queue_confirm" name="test_queue_confirm"></rabbit:queue> <rabbit:direct-exchange name="test_exchange_confirm"> <rabbit:bindings> <rabbit:binding queue="test_queue_confirm" key="confirm"></rabbit:binding> </rabbit:bindings> </rabbit:direct-exchange>

@RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml") public class ProducerTest { @Autowired private RabbitTemplate rabbitTemplate; /** * 确认模式： * 步骤： * 1. 确认模式开启：ConnectionFactory中开启publisher-confirms="true" * 2. 在rabbitTemplate定义ConfirmCallBack回调函数 */ @Test public void testConfirm() { //2. 定义回调 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { if (ack){ //接收成功 System.out.println("接收成功消息" + cause); }else { //接收失败 System.out.println("接收失败消息" + cause); //做一些处理,让消息再次发送。 } } }); //3. 发送消息 rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message confirm....");//成功 //rabbitTemplate.convertAndSend("test_exchange_confirm000", "confirm", "message confirm....");//失败 } }

@RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml") public class ProducerTest { @Autowired private RabbitTemplate rabbitTemplate; /** * 回退模式： 当消息发送给Exchange后,Exchange路由到Queue失败时 才会执行 ReturnCallBack * 步骤： * 1. 开启回退模式：publisher-returns="true" * 2. 设置ReturnCallBack * 3. 设置Exchange处理消息的模式： * 1). 如果消息没有路由到Queue,则丢弃消息（默认） * 2). 如果消息没有路由到Queue,返回给消息发送方ReturnCallBack * rabbitTemplate.setMandatory(true); */ @Test public void testReturn() { //设置交换机处理失败消息的模式 rabbitTemplate.setMandatory(true); //2.设置ReturnCallBack rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { /** * @param message 消息对象 * @param replyCode 错误码 * @param replyText 错误信息 * @param exchange 交换机 * @param routingKey 路由键 */ @Override public void returnedMessage(Message message, int replyCode,String replyText,String exchange,String routingKey) { System.out.println("return 执行了...."); System.out.println(message); System.out.println(replyCode); System.out.println(replyText); System.out.println(exchange); System.out.println(routingKey); //处理 } }); //3. 发送消息 /** * 参数1：交换机名称 * 参数2：路由键名（广播设置为空） * 参数3：发送的消息内容 */ rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message confirm...."); } }

ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。

二种确认方式

<!--加载配置文件--> <context:property-placeholder location="classpath:rabbitmq.properties"/> <!-- 定义rabbitmq connectionFactory --> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.virtual-host}"/> <context:component-scan base-package="com.atguigu.listener" /> <!--定义监听器容器 acknowledge="manual"：手动签收 --> <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual"> <rabbit:listener ref="ackListener" queue-names="test_queue_confirm"></rabbit:listener> </rabbit:listener-container>

自动确认

手动确认

在rabbit:listener-container标签中设置acknowledge属性，设置ack方式 none：自动确认，manual：手动确认

如果在消费端没有出现异常，则调用channel.basicAck(deliveryTag，true);方法确认签收消息

如果出现异常，则在catch中调用 basicNack，拒绝消息，让MQ重新发送消息。

/** * Consumer 限流机制 * 1. 确保消息被确认。不确认是不继续处理其他消息的 * 2. listener-container配置属性 * prefetch = 1,表示消费端每次从mq拉去一条消息来消费,直到手动确认消费完毕后,才会继续拉取下一条消息。 */ @Component public class QosListener implements ChannelAwareMessageListener { @Override public void onMessage(Message message, Channel channel) throws Exception { //1.获取消息 System.out.println(new String(message.getBody())); } }

分支主题

有 9 条消息待消费 ，有 1 条消息未确认

分支主题

在 <rabbit:listener-container>中配置 prefetch 属性设置消费端一次拉取多少条消息

消费端的必须确认才会继续处理其他消息。

spring-rabbitmq-producer.xml

/** * TTL：过期时间 * 1. 队列统一过期 * 2. 消息单独过期 * 如果设置了消息的过期时间,也设置了队列的过期时间,它以时间短的为准。 */ @Test public void testMessageTtl() { // 消息后处理对象,设置一些消息的参数信息 MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { //1.设置message的信息 // 第二个方法：消息的过期时间 ,5秒之后过期 message.getMessageProperties().setExpiration("5000"); //2.返回该消息 return message; } }; //消息单独过期 rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl....",messagePostProcessor); }

1. 队列消息数量到达限制；比如队列最大只能存储10条消息，而发了11条消息，根据先进先出，最先发的消息会进入死信队列。

2. 消费者拒接消费消息，basicNack/basicReject，并且不把消息重新放入原目标队列，requeue=false；

3. 原队列存在消息过期设置，消息到达超时时间未被消费；

① 丢弃，如果不是很重要，可以选择丢弃

② 记录死信入库，然后做后续的业务分析或处理

③ 通过死信队列，由负责监听死信的应用程序进行处理

死信队列小结

在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将进行取消处理。这时就可以使用延时队列将订单信息发送到延时队列。

1. 下单后，30分钟未支付，取消订单，回滚库存。

2. 新用户注册成功30分钟后，发送短信问候。

分支主题

@Component public class OrderListener implements ChannelAwareMessageListener { @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //1.接收转换消息 System.out.println(new String(message.getBody())); //2. 处理业务逻辑 System.out.println("处理业务逻辑..."); System.out.println("根据订单id查询其状态..."); System.out.println("判断状态是否为支付成功"); System.out.println("取消订单,回滚库存...."); //3. 手动签收 channel.basicAck(deliveryTag,true); } catch (Exception e) { //e.printStackTrace(); System.out.println("出现异常,拒绝接受"); //4.拒绝签收,不重回队列 requeue=false channel.basicNack(deliveryTag,true,false); } } }

spring-rabbitmq-consumer.xml