• RabbitMQ高级-高级特性


    1.消息可靠性传递

    在使用RabbitMQ的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ为我们提供了两种方式来控制消息的投递可靠性模式

    1.confirm 确认模式        确认模式是由exchange决定的

    2.return 退回模式        回退模式是由routingKey决定的

    rabbitmq整个消息投递的路径为:

    producer -> rabbitmq broker -> exchange -> queue ->consumer

    消息从producer到exchange则会返回一个confirmCallback

    消息从exchange->queue投递失败则会返回一个returnCallback

    我们将利用这两个callback控制消息的可靠性投递

    创建rabbitmq-producer-spring项目 并配置

    配置文件:开启模式

    1. "1.0" encoding="UTF-8"?>
    2. <beans xmlns="http://www.springframework.org/schema/beans"
    3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4. xmlns:context="http://www.springframework.org/schema/context"
    5. xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    6. xsi:schemaLocation="http://www.springframework.org/schema/beans
    7. http://www.springframework.org/schema/beans/spring-beans.xsd
    8. http://www.springframework.org/schema/context
    9. https://www.springframework.org/schema/context/spring-context.xsd
    10. http://www.springframework.org/schema/rabbit
    11. http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    12. <context:property-placeholder location="classpath:rabbitmq.properties"/>
    13. <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
    14. port="${rabbitmq.port}"
    15. username="${rabbitmq.username}"
    16. password="${rabbitmq.password}"
    17. virtual-host="${rabbitmq.virtual-host}"
    18. publisher-confirms="true"
    19. publisher-returns="true"
    20. />
    21. <rabbit:admin connection-factory="connectionFactory"/>
    22. <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
    23. <rabbit:queue id="test_queue_confirm" name="test_queue_confirm">rabbit:queue>
    24. <rabbit:direct-exchange name="test_exchange_confirm">
    25. <rabbit:bindings>
    26. <rabbit:binding queue="test_queue_confirm" key="confirm">rabbit:binding>
    27. rabbit:bindings>
    28. rabbit:direct-exchange>
    29. beans>

    确认模式 (publisher-confirms="true") 

    1. @RunWith(SpringRunner.class)
    2. @ContextConfiguration(locations = "classpath:/spring-rabbitmq-producer.xml")
    3. public class ProducerTest {
    4. //@Autowired 在配置文件配好了
    5. //RabbitAdmin rabbitAdmin;//用于操作交换机、队列
    6. @Autowired
    7. private RabbitTemplate rabbitTemplate;
    8. //确认模式,收到消息与否都会被执行
    9. @Test
    10. public void testConfirm() {
    11. //1. 开启去确认回调函数 publisher-confirms="true"
    12. //2. 注册回调函数,当交换机收到消息,confirm方法会被调用
    13. rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
    14. @Override
    15. public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    16. System.out.println("correlationData = " + correlationData); //关联数据,在发送消息时指定的
    17. //确认机制:ack true表示收到了 false表示没收到
    18. if (ack){
    19. System.out.println("接收成功消息" + ack);
    20. }else {
    21. System.out.println("接收失败消息" + cause);
    22. //做一些处理,让消息再次发送。
    23. }
    24. }
    25. });
    26. //3. 发送消息
    27. rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "confirm info...");//成功
    28. //测试不正常情况
    29. //rabbitTemplate.convertAndSend("test_exchange_confirm000", "confirm", "message confirm....");//失败
    30. }
    31. }

    回退模式 (publisher-returns="true")

    1. @RunWith(SpringRunner.class)
    2. @ContextConfiguration(locations = "classpath:/spring-rabbitmq-producer.xml")
    3. public class ProducerTest {
    4. //@Autowired 在配置文件配好了
    5. //RabbitAdmin rabbitAdmin;//用于操作交换机、队列
    6. @Autowired
    7. private RabbitTemplate rabbitTemplate;
    8. //回退模式
    9. @Test
    10. public void testReturn() {
    11. //1.开启回退模式 publisher-returns="true"
    12. //2.注册回退函数(ReturnCallBack),当没有任何队列与交换机绑定,或即使绑定也没有任何匹配时,将执行回退方法
    13. // 成功接收,不执行return方法
    14. //3. 必须设置强制回退 setMandatory(true) 设置交换机处理失败消息的模式
    15. rabbitTemplate.setMandatory(true);
    16. rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
    17. @Override
    18. public void returnedMessage(Message message, int replyCode, String replyText,
    19. String exchange, String routingKey) {
    20. System.out.println("消息对象message = " + message);
    21. System.out.println("错误码replyCode = " + replyCode);
    22. System.out.println("错误信息replyText = " + replyText);
    23. System.out.println("交换机exchange = " + exchange);
    24. System.out.println("路由键routingKey = " + routingKey);
    25. //处理
    26. }
    27. });
    28. //3. 发送消息
    29. //rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "confirm info...");//成功
    30. //测试不正常情况
    31. rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm123", "confirm info...");//成功
    32. }
    33. }

    小结:

    设置 ConnectionFactory的publisher-confirms="true" 开启 确认模式

    使用 rabbitTemplate.setConfirmCallback 设置回调函数。当消息发送到 exchange 后回调 confirm 方法。在方法中判断 ack,如果为true,则发送成功,如果为false,则发送失败,需要处理。

    设置 ConnectionFactory 的 publisher-returns="true" 开启 退回模式

    使用 rabbitTemplate.setReturnCallback 设置退回函数,当消息从exchange 路由到 queue 失败后,如果设置了 rabbitTemplate.setMandatory(true) 参数,则会将消息退回给 producer并执行回调函数returnedMessage

    2.Consumer Ack (消费端确认方式)

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

    有三种确认方式:

    自动确认:acknowledge=“none”   默认

    手动确认:acknowledge=“manual”

    根据异常情况确认:acknowledge=“auto”,(这种方式使用麻烦)

    其中自动确认是指,当消息一旦被Consumer接收到,则自动确认收到,并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中,很可能消息接收到,业务处理出现异常,那么该消息就会丢失。

    如果设置了手动确认方式,则需要在业务处理成功后,调用channel.basicAck(),手动签收,如果出现异常,则调用channel.basicNack()方法,让其自动重新发送消息

    创建rabbitmq-consumer-spring项目并配置

    1. "1.0" encoding="UTF-8"?>
    2. "http://www.springframework.org/schema/beans"
    3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4. xmlns:context="http://www.springframework.org/schema/context"
    5. xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    6. xsi:schemaLocation="http://www.springframework.org/schema/beans
    7. http://www.springframework.org/schema/beans/spring-beans.xsd
    8. http://www.springframework.org/schema/context
    9. https://www.springframework.org/schema/context/spring-context.xsd
    10. http://www.springframework.org/schema/rabbit
    11. http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    12. "classpath:rabbitmq.properties"/>
    13. "connectionFactory" host="${rabbitmq.host}"
    14. port="${rabbitmq.port}"
    15. username="${rabbitmq.username}"
    16. password="${rabbitmq.password}"
    17. virtual-host="${rabbitmq.virtual-host}"/>
    18. package="com.listener" />
    19. "connectionFactory" acknowledge="manual">
    20. "ackListener" queue-names="test_queue_confirm">
    1. package com;
    2. import org.junit.Test;
    3. import org.junit.runner.RunWith;
    4. import org.springframework.test.context.ContextConfiguration;
    5. import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    6. @RunWith(SpringJUnit4ClassRunner.class)
    7. @ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
    8. public class ConsumerTest {
    9. @Test
    10. public void test(){
    11. while (true){
    12. }
    13. }
    14. }

    1.自动确认 (默认) 

    1. //自动确认
    2. @Component
    3. //implements MessageListener 自动确认
    4. //ChannelAwareMessageListener 手动确认
    5. public class AckListener implements MessageListener {
    6. //自动确认
    7. @Override
    8. public void onMessage(Message message) {
    9. try {
    10. System.out.println(new String(message.getBody(),"UTF-8"));
    11. } catch (UnsupportedEncodingException e) {
    12. e.printStackTrace();
    13. }
    14. }
    15. }

    2.手动确认 (acknowledge="manual":手动签收) 

    1. package com.listener;
    2. import com.rabbitmq.client.Channel;
    3. import org.springframework.amqp.core.Message;
    4. import org.springframework.amqp.core.MessageListener;
    5. import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
    6. import org.springframework.stereotype.Component;
    7. import javax.sound.midi.Soundbank;
    8. import java.io.UnsupportedEncodingException;
    9. /*
    10. * 1.设置手动签收:acknowledge=”manual“
    11. * 2.让监听器类实现ChannelAwareMessageListener接口
    12. * 3.如果消息成功处理,则调用channel的basicAck()签收
    13. * 4.如果消息处理失败,则调用channel的basicNack()拒绝签收,broker重新发送给consumer
    14. * */
    15. @Component
    16. //implements MessageListener 自动确认
    17. //ChannelAwareMessageListener 手动确认
    18. public class AckListener implements ChannelAwareMessageListener {
    19. //手动确认
    20. @Override
    21. public void onMessage(Message message, Channel channel) throws Exception {
    22. //消息唯一标识
    23. long deliveryTag = message.getMessageProperties().getDeliveryTag();
    24. try {
    25. //1.接收转换消息
    26. System.out.println(new String(message.getBody(),"UTF-8"));
    27. //消息得到后需要进行业务逻辑处理,如存储数据库,发送邮件、短信、保存到Redis等等
    28. //2.处理业务逻辑
    29. System.out.println("处理业务逻辑...");
    30. int i = 3/0;//模拟业务异常
    31. //3.手动签收
    32. //业务逻辑成功,才会确认,否则不予确认
    33. channel.basicAck(deliveryTag,true);//消息唯一标识和是否确认多个
    34. } catch (Exception e) {
    35. e.printStackTrace();
    36. //4.拒绝签收
    37. //NoAck 不予确认
    38. /*
    39. * 三个参数
    40. * deliveryTag 消息标签,唯一标识,类似数据库主键
    41. * multiple true 拒绝接收的所有消息 false,拒绝刚刚的消息
    42. * requeue true 表示是否重回原有队列,false表示丢弃或成为死信 (被死信队列接收)
    43. */
    44. channel.basicNack(deliveryTag,true,false);
    45. }
    46. }
    47. }

    3.消费端限流 (prefetch="2"限流为2) 

     

    就是在消费这些消息的时候做了限流

    修改配置

    1. <context:component-scan base-package="com.listener" />
    2. <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="2">
    3. <rabbit:listener ref="qosListener" queue-names="test_queue_confirm">rabbit:listener>
    4. rabbit:listener-container>

    新建 com.listener.QosListener 

    1. package com.listener;
    2. import com.rabbitmq.client.Channel;
    3. import org.springframework.amqp.core.Message;
    4. import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
    5. import org.springframework.stereotype.Component;
    6. /**
    7. * Consumer 限流机制
    8. * 1. 确保消息被确认。不确认是不继续处理其他消息的
    9. * 2. listener-container配置属性
    10. * prefetch=1,表示消费端每次从mq拉去一条消息来消费,直到手动确认消费完毕后,才会继续拉取下一条消息。
    11. */
    12. @Component
    13. public class QosListener implements ChannelAwareMessageListener {
    14. @Override
    15. public void onMessage(Message message, Channel channel) throws Exception {
    16. long deliveryTag = message.getMessageProperties().getDeliveryTag();
    17. try {
    18. //1.获取消息
    19. System.out.println(new String(message.getBody()));
    20. //2.确认消息
    21. channel.basicAck(deliveryTag,true);
    22. }catch (Exception e){
    23. e.printStackTrace();
    24. channel.basicNack(deliveryTag,true,false);
    25. }
    26. }
    27. }

    小结:

    中配置prefetch属性设置消费端一次拉去多少消息

    消费端的确认模式一定为手动确认。acknowledge=“manual”

    4.TTL (消息存活时间)

    TTL 全称 Time To Live(存活时间/过期时间)

    当消息到达存活时间后,还没有被消费,会被自动清除

    RabbitMQ可以对消息设置过期时间,也可以对整个队列(Queue)设置过期时间

    控制后台演示消息过期 

    修改管理后台界面,增加队列

    参数:表示过期时间,单位毫秒 ,10000表示10秒

    增加交换机

    绑定队列

    发送消息

    Delivery mode:2-Persistent表示需要进行持久化

    查看消息,可以看到消息,但十秒之后,消息自动消失,因为我们设置了十秒消息过期

    代码实现 

    修改配置文件 spring-rabbitmq-producer.xml

    1. <rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
    2. <rabbit:queue-arguments>
    3. <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer">entry>
    4. <entry key="x-queue-type" value="classic" value-type="java.lang.String">entry>
    5. rabbit:queue-arguments>
    6. rabbit:queue>
    7. <rabbit:topic-exchange name="test_exchange_ttl">
    8. <rabbit:bindings>
    9. <rabbit:binding pattern="ttl.#" queue="test_queue_ttl">rabbit:binding>
    10. rabbit:bindings>
    11. rabbit:topic-exchange>

    在测试类 ProducerTest 中,添加测试方法,发送消息 

    1. /**
    2. * TTL:过期时间
    3. * 1. 队列统一过期
    4. * 2. 消息单独过期
    5. * 如果设置了消息的过期时间,也设置了队列的过期时间,它以时间短的为准。
    6. */
    7. @Test
    8. public void testTTL() {
    9. //队列统一过期
    10. for (int i = 0; i < 10; i++) {
    11. rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl");
    12. }
    13. }
    14. @Test
    15. public void testMessageTtl() {
    16. // 消息后处理对象,设置一些消息的参数信息
    17. MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
    18. @Override
    19. public Message postProcessMessage(Message message) throws AmqpException {
    20. //1.设置message的信息
    21. // 第二个方法:消息的过期时间 ,5秒之后过期
    22. message.getMessageProperties().setExpiration("5000");
    23. //2.返回该消息
    24. return message;
    25. }
    26. };

    小结

    设置消息队列过期时间使用参数:x-message-ttl,单位:ms(毫秒),会对整个队列消息统一过期

    设置消息过期时间使用参数:expiration。单位:ms(毫秒),当该消息在队列头部时(消费时),会单独判断这一消息是否过期

    如果两者都进行了设置,以时间段的为准

    5.DLX(死信队列)

    死信队列,英文缩写:DLX 。DeadLetter Exchange(死信交换机),无法被消费的消息,当消息成为Dead message后,可以被重新发送到另一个交换机,这个交换机就是DLX

    面试题: 消息成为死信的三种情况:

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

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

    消息超时,原队列存在消息过期设置,消息到达超时时间未被消费

    队列绑定死信交换机:

    给队列设置参数:x-dead-letter-exchange和x-dead-letter-routing-key

    修改配置文件 spring-rabbitmq-producer.xml

    1. <rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
    2. <rabbit:queue-arguments>
    3. <entry key="x-dead-letter-exchange" value="exchange_dlx"/>
    4. <entry key="x-dead-letter-routing-key" value="dlx.hehe">entry>
    5. <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
    6. <entry key="x-max-length" value="10" value-type="java.lang.Integer"/>
    7. rabbit:queue-arguments>
    8. rabbit:queue>
    9. <rabbit:topic-exchange name="test_exchange_dlx">
    10. <rabbit:bindings>
    11. <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx">rabbit:binding>
    12. rabbit:bindings>
    13. rabbit:topic-exchange>
    14. <rabbit:queue name="queue_dlx" id="queue_dlx">rabbit:queue>
    15. <rabbit:topic-exchange name="exchange_dlx">
    16. <rabbit:bindings>
    17. <rabbit:binding pattern="dlx.#" queue="queue_dlx">rabbit:binding>
    18. rabbit:bindings>
    19. rabbit:topic-exchange>

    在测试类 ProducerTest 中,添加测试方法

    三种成为死信情况 

    1. /*
    2. 发送测试死信消息:
    3. 1、过期时间
    4. 2、长度限制
    5. 3、消息拒收
    6. */
    7. @Test
    8. public void testDlx(){
    9. //1. 测试过期时间,死信消息
    10. //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?");
    11. //2. 测试长度限制后,消息死信
    12. /*for (int i = 0; i < 11; i++) {
    13. rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?i =" + i);
    14. }*/
    15. //3.发送消息:拒收情况
    16. rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?");
    17. }

    消费端需创建DlxListener并在配置添加对应监听容器 

    小结

    1.死信交换机和死信队列和普通的没有区别

    2.当消息成为死信后,如果该队列绑定了死信交换机,则消息会被死信交换机重新路由到死信队列

    6.延迟队列

    延迟队列,即消息进入队列后不会立即被消费,只有到达指定时间后,才会被消费

    延迟队列存储的对象肯定是对应的延时消息,所谓”延时消息”是指当消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费

    需求场景:

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

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

    实现方式

    1.定时器

    2.延迟队列

    但是RabbitMQ没有提供延迟队列功能

    但是可以使用:TTL+死信队列 组合实现延迟队列的效果

    代码实现 

    修改配置文件 spring-rabbitmq-producer.xml

    1. <rabbit:queue name="order_queue" id="order_queue">
    2. <rabbit:queue-arguments>
    3. <entry key="x-dead-letter-exchange" value="order_exchange_dlx">entry>
    4. <entry key="x-dead-letter-routing-key" value="dlx.order.cancel">entry>
    5. <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
    6. rabbit:queue-arguments>
    7. rabbit:queue>
    8. <rabbit:topic-exchange name="order_exchange">
    9. <rabbit:bindings>
    10. <rabbit:binding pattern="order.#" queue="order_queue">rabbit:binding>
    11. rabbit:bindings>
    12. rabbit:topic-exchange>
    13. <rabbit:queue name="order_queue_dlx" id="order_queue_dlx">rabbit:queue>
    14. <rabbit:topic-exchange name="order_exchange_dlx">
    15. <rabbit:bindings>
    16. <rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx">rabbit:binding>
    17. rabbit:bindings>
    18. rabbit:topic-exchange>

    生产端:ProducerTest

    1. package com;
    2. import org.junit.Test;
    3. import org.junit.runner.RunWith;
    4. import org.springframework.amqp.AmqpException;
    5. import org.springframework.amqp.core.Message;
    6. import org.springframework.amqp.core.MessagePostProcessor;
    7. import org.springframework.amqp.rabbit.connection.CorrelationData;
    8. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    9. import org.springframework.beans.factory.annotation.Autowired;
    10. import org.springframework.test.context.ContextConfiguration;
    11. import org.springframework.test.context.junit4.SpringRunner;
    12. @RunWith(SpringRunner.class)
    13. @ContextConfiguration(locations = "classpath:/spring-rabbitmq-producer.xml")
    14. public class ProducerTest {
    15. //@Autowired 在配置文件配好了
    16. //RabbitAdmin rabbitAdmin;//用于操作交换机、队列
    17. @Autowired
    18. private RabbitTemplate rabbitTemplate;
    19. //确认模式,收到消息与否都会被执行
    20. @Test
    21. public void testConfirm() {
    22. //1. 开启去确认回调函数 publisher-confirms="true"
    23. //2. 注册回调函数,当交换机收到消息,confirm方法会被调用
    24. rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
    25. @Override
    26. public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    27. System.out.println("correlationData = " + correlationData); //关联数据,在发送消息时指定的
    28. //确认机制:ack true表示收到了 false表示没收到
    29. if (ack){
    30. System.out.println("接收成功消息" + ack);
    31. }else {
    32. System.out.println("接收失败消息" + cause);
    33. //做一些处理,让消息再次发送。
    34. }
    35. }
    36. });
    37. //3. 发送消息
    38. rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "confirm info...");//成功
    39. //测试不正常情况
    40. //rabbitTemplate.convertAndSend("test_exchange_confirm000", "confirm", "message confirm....");//失败
    41. }
    42. @Test
    43. public void testSend(){
    44. for (int i = 0; i < 10; i++) {
    45. //发送消息
    46. rabbitTemplate.convertAndSend("test_exchange_confirm","confirm","confirm info...");
    47. }
    48. }
    49. /**
    50. * TTL:过期时间
    51. * 1. 队列统一过期
    52. * 2. 消息单独过期
    53. * 如果设置了消息的过期时间,也设置了队列的过期时间,它以时间短的为准。
    54. */
    55. @Test
    56. public void testTTL() {
    57. //队列统一过期
    58. for (int i = 0; i < 10; i++) {
    59. rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl");
    60. }
    61. }
    62. @Test
    63. public void testMessageTtl() {
    64. // 消息后处理对象,设置一些消息的参数信息
    65. MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
    66. @Override
    67. public Message postProcessMessage(Message message) throws AmqpException {
    68. //1.设置message的信息
    69. // 第二个方法:消息的过期时间 ,5秒之后过期
    70. message.getMessageProperties().setExpiration("5000");
    71. //2.返回该消息
    72. return message;
    73. }
    74. };
    75. }
    76. /*
    77. 发送测试死信消息:
    78. 1、过期时间
    79. 2、长度限制
    80. 3、消息拒收
    81. */
    82. @Test
    83. public void testDlx(){
    84. //1. 测试过期时间,死信消息
    85. //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?");
    86. //2. 测试长度限制后,消息死信
    87. /*for (int i = 0; i < 11; i++) {
    88. rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?i =" + i);
    89. }*/
    90. //3.发送消息:拒收情况
    91. rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?");
    92. }
    93. @Test
    94. public void testDelay() throws InterruptedException {
    95. //1、发送订单消息。将来是在订单系统中,下单成功后,发送消息
    96. rabbitTemplate.convertAndSend("order_exchange","order.msg","订单信息:id=1,time=333333");
    97. //2、打印倒计时
    98. for (int i = 10; i > 0; i--) {
    99. System.out.println(i+"...");
    100. Thread.sleep(1000);
    101. }
    102. }
    103. }

    消费端:OrderListener

    1. package com.listener;
    2. import com.rabbitmq.client.Channel;
    3. import org.springframework.amqp.core.Message;
    4. import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
    5. import org.springframework.stereotype.Component;
    6. @Component
    7. public class OrderListener implements ChannelAwareMessageListener {
    8. @Override
    9. public void onMessage(Message message, Channel channel) throws Exception {
    10. Thread.sleep(1000);
    11. long deliveryTag = message.getMessageProperties().getDeliveryTag();
    12. try{
    13. //1.接收转换消息
    14. System.out.println(new String(message.getBody()));
    15. //2.处理业务逻辑
    16. System.out.println("处理业务逻辑...");
    17. System.out.println("根据订单id查询其状态...");
    18. System.out.println("判断状态是否为支付成功?");
    19. System.out.println("取消订单,回滚库存!");
    20. //3.手动签收
    21. channel.basicAck(deliveryTag,true);
    22. }catch (Exception e){
    23. //4.拒绝签收
    24. /*
    25. * 第三个参数:requeue:重回队列。如果设置为true,则消息重新回到queue,broker会重新发送该消息给消费端
    26. *
    27. * */
    28. System.out.println("出现异常,拒绝接收!");
    29. //拒绝接收,不重回队列requeue=false
    30. channel.basicNack(deliveryTag,true,false);
    31. //channel.basicReject(deliveryTag,true);只允许单挑确认
    32. }
    33. }
    34. }

    修改配置:spring-rabbitmq-consumer.xml 

    1. "1.0" encoding="UTF-8"?>
    2. <beans xmlns="http://www.springframework.org/schema/beans"
    3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4. xmlns:context="http://www.springframework.org/schema/context"
    5. xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    6. xsi:schemaLocation="http://www.springframework.org/schema/beans
    7. http://www.springframework.org/schema/beans/spring-beans.xsd
    8. http://www.springframework.org/schema/context
    9. https://www.springframework.org/schema/context/spring-context.xsd
    10. http://www.springframework.org/schema/rabbit
    11. http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    12. <context:property-placeholder location="classpath:rabbitmq.properties"/>
    13. <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
    14. port="${rabbitmq.port}"
    15. username="${rabbitmq.username}"
    16. password="${rabbitmq.password}"
    17. virtual-host="${rabbitmq.virtual-host}"/>
    18. <context:component-scan base-package="com.listener" />
    19. <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="2">
    20. <rabbit:listener ref="orderListener" queue-names="order_queue_dlx">rabbit:listener>
    21. rabbit:listener-container>
    22. beans>

    消费端:ConsumerTest

    1. package com;
    2. import org.junit.Test;
    3. import org.junit.runner.RunWith;
    4. import org.springframework.test.context.ContextConfiguration;
    5. import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    6. @RunWith(SpringJUnit4ClassRunner.class)
    7. @ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
    8. public class ConsumerTest {
    9. @Test
    10. public void test(){
    11. while (true){
    12. }
    13. }
    14. }

    7.日志与监控

    1.RabbitMQ日志

    RabbitMQ默认日志存放路径: /var/log/rabbitmq/rabbit@xxx.log

    日志包含了RabbitMQ的版本号,Erlang的版本号,RabbitMQ服务结点名称,cookie的hash值,RabbitMQ配置文件地址,内存限制,磁盘限制,默认账户guest的创建以及权限配置等等

    2.web管控台监控 

    8.消息追踪-Firehose 

    在使用任何消息中间件的过程中,难免会出现某条消息异常丢失的情况。对于RabbitMQ而言,可能是因为生产者或消费者与RabbitMQ断开了连接,而它们与RabbitMQ又采用了不同的确认机制;也有可能是因为交换器与队列之间不同的转发策略;甚至是交换器并没有与任何队列进行绑定,生产者又不感知或者没有采取相应的措施;另外RabbitMQ本身的集群策略也可能导致消息的丢失。这个时候就需要有一个较好的机制跟踪记录消息的投递过程,以此协助开发和运维人员进行问题的定位

    在RabbitMQ中可以使用 Firehose 和 rabbitmq_tracing插件 功能来实现消息追踪

    firehose的机制是将生产者投递给rabbitmq的消息,rabbitmq投递给消费者的消息按照指定的格式发送到默认的exchange上。这个默认的exchange的名称为amq.rabbitmq.trace,它是一个topic类型的exchange。发送到这个exchange上的消息的routing key为 publish.exchangename 和 deliver.queuename。其中exchangename和queuename为实际exchange和queue的名称,分别对应生产者投递到exchange的消息,和消费者从queue上获取的消息

    注意: 打开 trace 会影响消息写入功能,适当打开后请关闭

    开启Firehose命令:

    rabbitmqctl trace_on

    关闭Firehose命令:

    rabbitmqctl trace_off

  • 相关阅读:
    LQ0142 技能升级【二分】
    CommunityToolkit.Mvvm8.1 viewmodel使用-旧式写法(2)
    centos 8.2 指南
    vue介绍
    力扣每日一题66:加一
    使用VNC过程中遇到的问题总结
    Java:单例模式探究
    数据结构栈的使用——马踏棋盘
    workflow一次完成多个模型评价和比较
    Minianaconda安装jupyter notebook遇到的问题及解决
  • 原文地址:https://blog.csdn.net/weixin_61611746/article/details/136306326