SpringBoot整合RabbitMQ
一、SpringBoot整合rabbitmq
在Spring项目中，可以使用Spring-Rabbit去操作RabbitMQ，其是在spring boot项目中只需要引入对应的amqp启动器依赖即可，方便的使用RabbitTemplate发送消息，使用注解接收消息。
生产者工程：
-
application.yml文件配置RabbitMQ相关信息；
-
在生产者工程中编写配置类，用于创建交换机和队列，并进行绑定
-
注入RabbitTemplate对象，通过RabbitTemplate对象发送消息到交换机
消费者工程：
-
application.yml文件配置RabbitMQ相关信息
-
创建消息处理类，用于接收队列中的消息并进行处理
1、生产方搭建
创建生产方工程springboot-rabbit-production
-
添加依赖
<?xml version="1.0" encoding="UTF-8"?> <project> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <exclusions> <exclusion> <groupId>org.junit.vintage</groupId> <artifactId>junit-vintage-engine</artifactId> </exclusion> </exclusions> </dependency> </dependencies> </project> -
修改配置文件
server: port: 8080 spring: application: name: springboot-rabbit-production rabbitmq: #rabbitmq配置 host: 127.0.0.1 virtual-host: / username: guest password: guest port: 5672 -
创建RabbitMQ队列与交换机绑定的配置类
/** * RabbitMQ队列与交换机绑定的配置类 */ @Configuration public class RabbitConfig { //交换机名称 public static final String ITEM_TOPIC_EXCHANGE = "item_topic_exchange"; //队列名称 public static final String ITEM_QUEUE = "item_queue"; /** * 实例化一个交换机 * @return */ @Bean public TopicExchange itemTopicExchange(){ return ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build(); //也可以直接new //return new TopicExchange("itemTopicExchange",true,false); } /** * 实例化一个对列 * @return */ @Bean public Queue itemQueue(){ // return QueueBuilder.durable(queueName).exclusive().autoDelete().build(); return QueueBuilder.durable(ITEM_QUEUE).build(); //return new Queue("itemQueue",true,false,false,null); } /** * 绑定队列和交换机 * @param queue * @param exchange * @return */ @Bean public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue, @Qualifier("itemTopicExchange") Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs(); } }创建队列参数说明：
参数 说明 name 字符串值，exchange 的名称。 durable 布尔值，表示该 queue 是否持久化。 持久化意味着当 RabbitMQ 重启后，该 queue 是否会恢复/仍存在。 另外，需要注意的是，queue 的持久化不等于其中的消息也会被持久化。 exclusive 布尔值，表示该 queue 是否排它式使用。排它式使用意味着仅声明他的连接可见/可用，其它连接不可见/不可用。 autoDelete 布尔值，表示当该 queue 没“人”（connection）用时，是否会被自动删除。 不指定 durable、exclusive 和 autoDelete 时，默认为 true 、 false 和 false 。表示持久化、非排它、不用自动删除
创建交换机参数说明
参数 说明 name 字符串值，exchange 的名称。 durable 布尔值，表示该 exchage 是否持久化。 持久化意味着当 RabbitMQ 重启后，该 exchange 是否会恢复/仍存在。 autoDelete 布尔值，表示当该 exchange 没“人”（queue）用时，是否会被自动删除。 不指定 durable 和 autoDelete 时，默认为
true和false。表示持久化、不用自动删除 -
创建测试类，发送消息到消息队列
@SpringBootTest class SpringbootRabbitProductionApplicationTests { @Autowired private RabbitTemplate rabbitTemplate; @Test public void test1(){ rabbitTemplate.convertAndSend(RabbitConfig.ITEM_TOPIC_EXCHANGE, "item.add", "SpringBoot真滴好"); //发送的消息也可以是对象，只不过对象需要序列化 } @Test public void test(){ rabbitTemplate.convertAndSend(RabbitConfig.ITEM_TOPIC_EXCHANGE, "item.insert", "商品新增id..."); rabbitTemplate.convertAndSend(RabbitConfig.ITEM_TOPIC_EXCHANGE, "item.update", "商品修改...."); rabbitTemplate.convertAndSend(RabbitConfig.ITEM_TOPIC_EXCHANGE, "item.delete", "商品删除...."); } }
2、消费方搭建
创建消费者工程springboot-rabbit-consumer
-
添加依赖
同上
-
修改配置文件
同上
-
创建消息监听处理类
方式一： @Component public class MyListener { @RabbitListener(queues = "item_queue") public void myMessage(String message){ System.out.println("消费者接收到的消息为：" + message); } } 方式二： @Component @RabbitListener(queues = "item_queue") public class HelloReceiver { @RabbitHandler public void process(String message) { System.out.println("消费者接收到的消息为："+message); } }
二、消息的应答确认
消费者消息确认分为以下两种模式 ：
-
自动确认：acknowledge="none"（默认，不推荐使用）
自动ACK: 消费者配置中如果是自动ack机制，MQ将消息发送给消费者后直接就将消息给删除了，这个的前提条件是消费者程序没有出现异常，如果消费者接收消息后处理时出现异常，那么MQ将会尝试重发消息给消费者直至达到了消费者服务中配置的最大重试次数后将会直接抛出异常不再重试。
-
手动确认：acknowledge="manual" ，也称手动ack （推荐使用）
手动ACK:消费者设置了手动ACK机制后，可以显式的提交/拒绝消息(这一步骤叫做发送ACK)，如果消息被消费后正常被提交了ack，那么此消息可以说是流程走完了，然后MQ将此消息从队列中删除。而如果消息被消费后被拒绝了，消费者可选择让MQ重发此消息或者让MQ直接移除此消息。后面可以使用死信队列来进行接收这些被消费者拒绝的消息，再进行后续的业务处理。
案例：手动确认消息
-
修改yml配置
server: port: 8081 spring: application: name: springboot-rabbit-consumer rabbitmq: #rabbitmq配置 host: 127.0.0.1 virtual-host: / username: guest password: guest port: 5672 listener: simple: acknowledge-mode: manual direct: acknowledge-mode: manual #消息的确认机制 none:自动模式（默认开启） manual:手动模式 auto:自动模式 （根据侦听器检测是正常返回、还是抛出异常来发出 ack/nack） -
创建监听器
@Component @RabbitListener(queues = "item_queue") public class MyListener { /** * 方式一 * @param str 消息内容 * @param channel * @param tag 消息编号 * @param message 消息对象 */ @SneakyThrows @RabbitHandler public void process(String str, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag, Message message){ System.out.println(str); System.out.println(tag); System.out.println(message); channel.basicReject(tag,true); //basicReject()用来拒绝消息，如果被拒绝的消息应该被排队而不是被丢弃 } /** * 方式二 * @param channel 队列通道 * @param message 消息对象 */ @SneakyThrows @RabbitHandler public void process(String str,Channel channel, Message message){ try{ System.out.println("CorrelationId:"+message.getMessageProperties().getCorrelationId()); System.out.println(str);//获得队列中的消息 System.out.println(message.getMessageProperties().getDeliveryTag()); channel.basicAck(message.getMessageProperties().getDeliveryTag(),true); //消费消息 }catch (Exception e){ if (message.getMessageProperties().getRedelivered()) { System.out.println("消息已重复处理失败,拒绝再次接收..."); channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息 } else { System.out.println("消息即将再次返回队列处理..."); channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); } } } } -
basicAck的批量应答问题说明：
-
channel.basicAck(100，true) 如果前面还有4，6，7的deliveryTag未被确认，则会一起确认，减少网络流量，当然当前deliveryTag=8这条消息也会确认，如果没有前面没有未被确认的消息，则只会确认当前消息，也就是说可以一次性确认某个队列小于等于delivery_tag值的所有消息
-
basicNack的参数说明：
-
第一个参数为deliveryTag，也就是每个消息标记index，消息标记值从1开始，依次递增
-
第二个参数为multiple，表示是否批量，如果为true，那么小于或者等于该消息标记的消息（如果还没有签收）都会拒绝签收
-
第三个参数为requeue，表示被拒绝的消息是否重回队列，如果设置为true，则消息重新回到queue，那么broker会重新推送该消息给消费端，如果设置为false，则消息在队列中被删除，即消息会被直接丢失（当然如果为false，还有一种情况就是放到死信队列）
三、消息过期时间设置
TTL 全称 Time To Live（存活时间/过期时间）。当消息到达存活时间后，还没有被消费，会被自动清除。RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。当消息超过过期时间还没有被消费，则丢弃
由于ttl表示消息在队列的存活时间，所以在生产者工程操作
-
修改rabbit队列配置
/** * RabbitMQ队列与交换机绑定的配置类 */ @Configuration public class RabbitConfig { //交换机名称 public static final String ITEM_TOPIC_EXCHANGE = "item_topic_exchange"; //队列名称 public static final String ITEM_QUEUE = "item_queue"; /** * 实例化一个交换机 * @return */ @Bean public TopicExchange itemTopicExchange(){ return ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build(); //也可以直接new //return new TopicExchange("itemTopicExchange",true,false); } /** * 实例化一个对列 * @return */ @Bean public Queue itemQueue(){ return QueueBuilder.durable(ITEM_QUEUE) .withArgument("x-message-ttl",20000) //设置队列过期时间 .build(); } /** * 绑定队列和交换机 * @param queue * @param exchange * @return */ @Bean public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue, @Qualifier("itemTopicExchange") Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs(); } } -
测试类
-
情况1：发给test_queue_ttl 队列的消息统一设置过期时间，交换机发给 test_queue_ttl 队列后，10秒后，10条消息消失
@SpringBootTest public class TtlTest { @Autowired private RabbitTemplate rabbitTemplate; /** * 10秒后队列中的数据消失 */ @Test public void test1(){ for(int i=1;i<=10;i++){ rabbitTemplate.convertAndSend("test_exchange", "ttl.h"+1, "第"+"条数据"); } } } -
情况2：某条消息单独设置过期时间
/** * 单独设置某条消息的过期时间 */ @Test public void test2(){ MessagePostProcessor messagePostProcessor=new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { //刚才我们在配置文件设置的队列的消息是10秒，这里是5秒，注意:以时间短的为准 message.getMessageProperties().setExpiration("5000"); //消息的过期时间 return message;//消息一定要返回 } }; rabbitTemplate.convertAndSend("test_exchange", "ttl.haha", "哈哈哈哈",messagePostProcessor); } -
情况3：发送给队列的n条信息中，单独给某个消息设置过期
/** * 发送给队列的n条信息中，单独给某个消息设置过期 */ @Test public void test3(){ MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { //刚才我们在配置文件设置的队列的消息是10秒，这里是5秒，注意:以时间短的为准 message.getMessageProperties().setExpiration("5000"); //消息的过期时间 return message;//消息一定要返回 } }; for(int i=1;i<=10;i++){ if(i==5) { rabbitTemplate.convertAndSend("test_exchange", "ttl.h" + 1, "第" + "条数据",messagePostProcessor); }else{ rabbitTemplate.convertAndSend("test_exchange", "ttl.h" + 1, "第" + "条数据"); } } }
-
情况3：当i == 5时，也就是给第五条消息设置过期时间是5秒，其它的还是10秒，发现失效，这里要注意一点，由于这条消息发送给队列的时候不是在队列的头部，故不会单独判断，而是和其它队列一样，10秒钟就消失，可以改成i==0，则第一条消息是5秒过期，或者i<3，即队列的头三条都是5秒的时间。
另外配置文件和代码都设置了过期时间，以时间短的为准
四、 死信队列和延迟队列
1、死信队列
死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息在队列成为Dead message后，通过该队列把这条死信消息发给另一个交换机，这个交换机就是DLX。
消息成为死信的三种情况（面试常问）：
-
队列消息长度到达限制；
-
消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；
-
原队列存在消息过期设置，消息到达超时时间未被消费；
队列绑定死信交换机： 给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key
a、生产方配置死信队列
-
修改生产者Rabbit配置类
/** * RabbitMQ队列与交换机绑定的配置类 */ @Configuration public class RabbitConfig { /** * 创建正常交换机 * @return */ @Bean("orderExchanger") public TopicExchange getOrderExchanger(){ return ExchangeBuilder.topicExchange("order_exchanger").durable(true).build(); } /** * 创建正常队列 * @return */ @Bean("orderQueue") public Queue getOrderQueue(){ return QueueBuilder.durable("order_queue") .withArgument("x-dead-letter-exchange","order_exchanger_dlx") //正常队列数据路由给死信交换机 .withArgument("x-dead-letter-routing-key","dlx.#") //正常队列与死信队列之间的路由键 .withArgument("x-message-ttl",20000) //队列过期时间 .build(); } /** * 正常交换机与正常队列绑定 * @param orderExchanger * @param orderQueue * @return */ @Bean public Binding orderBinding(@Qualifier("orderExchanger") Exchange orderExchanger, @Qualifier("orderQueue") Queue orderQueue){ return BindingBuilder.bind(orderQueue).to(orderExchanger).with("order.#").noargs(); } /** * 创建死信交换机 * @return */ @Bean("orderExchangeDlx") public TopicExchange getOrderExchangeDlx(){ return ExchangeBuilder.topicExchange("order_exchanger_dlx").durable(true).build(); } /** * 创建死信队列 * @return */ @Bean("orderQueueDlx") public Queue getOrderQueueDlx(){ return QueueBuilder.durable("order_queue_dlx").build(); } /** * 死信交换机与死信队列绑定 * @return */ @Bean public Binding orderBindingDlx(@Qualifier("orderExchangeDlx") Exchange orderExchangeDlx, @Qualifier("orderQueueDlx") Queue orderQueueDlx){ return BindingBuilder.bind(orderQueueDlx).to(orderExchangeDlx).with("dlx.#").noargs(); } } -
测试
@SpringBootTest class SpringbootRabbitProductionApplicationTests { @Autowired private RabbitTemplate rabbitTemplate; /** * 测试死信了队列 */ @Test public void testDlx(){ //1、测试过期时间，死信消息 // rabbitTemplate.convertAndSend("test_exchange","order.haha","我是一条消息，我会死吗"); //2、测试队列长度限制，消息死信 for (int i = 1; i <=20 ; i++) { rabbitTemplate.convertAndSend("test_exchange","order.haha","我是一条消息，我会死吗"+i); } } }在消费方中监听正常队列，然后拒绝消费并不重新放回原队列，该消息会进入死信队列
上面的测试结果：死信队列会有21条记录 1（过期） + 10（限制）+10（正常队列过期后的10条）
2、延迟队列
延迟队列，即消息进入队列后不会立即被消费者调用，只有到达指定时间后，才会被调用者调用消费。
-
需求一：下单后，30分钟未支付，取消订单，回滚库存。
当用户提交订单后，数据库保存订单信息，同时库存表相应的库存减少，然后消息队列保存订单的信息（如订单Id），此时库存系统监听队列，队列不会把消息立刻发送给库存，而是过30分钟再把信息发送给库存系统，库存系统去查询订单数据库，根据订单id查询，如果该订单还没有支付，则取消订单，回滚库存，如果支付过了，则库存表什么都不用做。也就是给用户30分钟的机会，一个订单在30分钟后还没有支付，则该订单的库存信息直接回滚。
-
需求二：新用户注册成功7天后，发送短信问候。
实现方式：
-
定时器：我们可以写一段代码，在某个时间段查询订单表的支付情况。把提交订单的时间查出来和当前系统时间比较，30分钟之类如果订单状态为未支付，则取消该订单，大家思考一下有什么问题？
-
延迟队列：很可惜，在RabbitMQ中并未提供延迟队列功能。但是可以使用：TTL+死信队列 组合实现延迟队列的效果。
-
延迟队列实现过程
-
修改生产方rabbit配置
/** * RabbitMQ队列与交换机绑定的配置类 */ @Configuration public class RabbitConfig { /** * 创建正常交换机 * @return */ @Bean("orderExchanger") public TopicExchange getOrderExchanger(){ return ExchangeBuilder.topicExchange("order_exchanger").durable(true).build(); } /** * 创建正常队列 * @return */ @Bean("orderQueue") public Queue getOrderQueue(){ return QueueBuilder.durable("order_queue") .withArgument("x-dead-letter-exchange","order_exchanger_dlx") //正常队列数据路由给死信交换机 .withArgument("x-dead-letter-routing-key","dlx.#") //正常队列与死信队列之间的路由键 .withArgument("x-message-ttl",20000) //队列过期时间 .build(); } /** * 正常交换机与正常队列绑定 * @param orderExchanger * @param orderQueue * @return */ @Bean public Binding orderBinding(@Qualifier("orderExchanger") Exchange orderExchanger, @Qualifier("orderQueue") Queue orderQueue){ return BindingBuilder.bind(orderQueue).to(orderExchanger).with("order.#").noargs(); } /** * 创建死信交换机 * @return */ @Bean("orderExchangeDlx") public TopicExchange getOrderExchangeDlx(){ return ExchangeBuilder.topicExchange("order_exchanger_dlx").durable(true).build(); } /** * 创建死信队列 * @return */ @Bean("orderQueueDlx") public Queue getOrderQueueDlx(){ return QueueBuilder.durable("order_queue_dlx").build(); } /** * 死信交换机与死信队列绑定 * @return */ @Bean public Binding orderBindingDlx(@Qualifier("orderExchangeDlx") Exchange orderExchangeDlx, @Qualifier("orderQueueDlx") Queue orderQueueDlx){ return BindingBuilder.bind(orderQueueDlx).to(orderExchangeDlx).with("dlx.#").noargs(); } } -
生产者测试类
@RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = {"classpath:rabbitmq-config.xml"}) public class RabbitTest { @Autowired private RabbitTemplate rabbitTemplate; /** * 延迟队列测试 */ @Test public void testDelay() throws InterruptedException { //1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息 rabbitTemplate.convertAndSend("order_exchange","order", "订单信息：id=1,time=2021年1月1日16:41:47"); //2.打印倒计时10秒 10秒后 消息发送到死信队列，而监听器OrderListener是监听死信队列的 for (int i = 10; i > 0 ; i--) { System.out.println(i); Thread.sleep(1000); } } } -
消费方监听类
@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) { System.out.println("出现异常，拒绝接受"); //4.拒绝签收，不重回队列 requeue=false channel.basicNack(deliveryTag,true,false); } } }启动生产方和消费方后可以看到10秒后消费方才获得队列中的数据
原文地址：https://blog.csdn.net/zhuge_long/article/details/137542164
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！