RabbitMQ快速使用代码手册

曹旭辉  金牌会员 | 2023-6-16 17:42:33 | 来自手机 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 835|帖子 835|积分 2505

本篇博客的内容为RabbitMQ在开发过程中的快速上手使用，侧重于代码部分，几乎没有相关概念的介绍，相关概念请参考以下csdn博客，两篇都是我找的精华帖，供大家学习。本篇博客也持续更新~~~
内容代码部分由于word转md格式有些问题，可以直接查看我的有道云笔记，链接：https://note.youdao.com/s/Ab7Cjiu
参考文档

csdn博客：
基础部分：https://blog.csdn.net/qq_35387940/article/details/100514134
高级部分：https://blog.csdn.net/weixin_49076273/article/details/124991012
application.yml
  1. server:
  3. port: 8021
  5. spring:
  7. #给项目来个名字
  9. application:
  11. name: rabbitmq-provider
  13. #配置rabbitMq 服务器
  15. rabbitmq:
  17. host: 127.0.0.1
  19. port: 5672
  21. username: root
  23. password: root
  25. #虚拟host 可以不设置,使用server默认host
  27. virtual-host: JCcccHost
  29. #确认消息已发送到交换机(Exchange)
  31. #publisher-confirms: true
  33. publisher-confirm-type: correlated
  35. #确认消息已发送到队列(Queue)
  37. publisher-returns: true
复制代码
完善更多信息
  1. spring:
  3. rabbitmq:
  5. host: localhost
  7. port: 5672
  9. virtual-host: /
  11. username: guest
  13. password: guest
  15. publisher-confirm-type: correlated
  17. publisher-returns: true
  19. template:
  21. mandatory: true
  23. retry:
  25. #发布重试，默认false
  27. enabled: true
  29. #重试时间 默认1000ms
  31. initial-interval: 1000
  33. #重试最大次数 最大3
  35. max-attempts: 3
  37. #重试最大间隔时间
  39. max-interval: 10000
  41. #重试的时间隔乘数，比如配2，0
  42. 第一次等于10s，第二次等于20s，第三次等于40s
  44. multiplier: 1
  46. listener:
  48. \# 默认配置是simple
  50. type: simple
  52. simple:
  54. \# 手动ack Acknowledge mode of container. auto none
  56. acknowledge-mode: manual
  58. #消费者调用程序线程的最小数量
  60. concurrency: 10
  62. #消费者最大数量
  64. max-concurrency: 10
  66. #限制消费者每次只处理一条信息，处理完在继续下一条
  68. prefetch: 1
  70. #启动时是否默认启动容器
  72. auto-startup: true
  74. #被拒绝时重新进入队列
  76. default-requeue-rejected: true
复制代码
相关注解说明

@RabbitListener 注解是指定某方法作为消息消费的方法，例如监听某 Queue
里面的消息。
@RabbitListener标注在方法上，直接监听指定的队列，此时接收的参数需要与发送市类型一致。
  1. \@Component
  3. public class PointConsumer {
  5. //监听的队列名
  7. \@RabbitListener(queues = "point.to.point")
  9. public void processOne(String name) {
  11. System.out.println("point.to.point：" + name);
  13. }
  15. }
复制代码
@RabbitListener 可以标注在类上面，需配合 @RabbitHandler 注解一起使用
@RabbitListener 标注在类上面表示当有收到消息的时候，就交给
@RabbitHandler 的方法处理，根据接受的参数类型进入具体的方法中。
  1. \@Component
  3. \@RabbitListener(queues = "consumer_queue")
  5. public class Receiver {
  7. \@RabbitHandler
  9. public void processMessage1(String message) {
  11. System.out.println(message);
  13. }
  15. \@RabbitHandler
  17. public void processMessage2(byte\[\] message) {
  19. System.out.println(new String(message));
  21. }
  23. }
复制代码
@Payload
可以获取消息中的 body 信息
  1. \@RabbitListener(queues = "debug")
  3. public void processMessage1(@Payload String body) {
  5. System.out.println("body："+body);
  7. }
复制代码
@Header，@Headers
可以获得消息中的 headers 信息
  1. \@RabbitListener(queues = "debug")
  3. public void processMessage1(@Payload String body, \@Header String token)
  4. {
  6. System.out.println("body："+body);
  8. System.out.println("token："+token);
  10. }
  12. \@RabbitListener(queues = "debug")
  14. public void processMessage1(@Payload String body, \@Headers
  15. Map\<String,Object\> headers) {
  17. System.out.println("body："+body);
  19. System.out.println("Headers："+headers);
  21. }
复制代码
快速使用

配置xml文件
  1. <dependency\>
  3. \<groupId\>org.springframework.boot\</groupId\>
  5. \<artifactId\>spring-boot-starter-amqp\</artifactId\>
  7. \</dependency\>
复制代码
配置exchange、queue

注解快速创建版本
  1. \@Configuration
  3. public class RabbitmqConfig {
  5. //创建交换机
  7. //通过ExchangeBuilder能创建direct、topic、Fanout类型的交换机
  9. \@Bean("bootExchange")
  11. public Exchange bootExchange() {
  13. return
  14. ExchangeBuilder.topicExchange("zx_topic_exchange").durable(true).build();
  16. }
  18. //创建队列
  20. \@Bean("bootQueue")
  22. public Queue bootQueue() {
  24. return QueueBuilder.durable("zx_queue").build();
  26. }
  28. /\*\*
  30. \* 将队列与交换机绑定
  32. \*
  34. \* \@param queue
  36. \* \@param exchange
  38. \* \@return
  40. \*/
  42. \@Bean
  44. public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue,
  45. \@Qualifier("bootExchange") Exchange exchange) {
  47. return
  48. BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
  50. }
  52. }
复制代码
Direct
  1. import org.springframework.amqp.core.Binding;
  3. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.DirectExchange;
  7. import org.springframework.amqp.core.Queue;
  9. import org.springframework.context.annotation.Bean;
  11. import org.springframework.context.annotation.Configuration;
  13. /\*\*
  15. \* \@Author : JCccc
  17. \* \@CreateTime : 2019/9/3
  19. \* \@Description :
  21. \*\*/
  23. \@Configuration
  25. public class DirectRabbitConfig {
  27. //队列 起名：TestDirectQueue
  29. \@Bean
  31. public Queue TestDirectQueue() {
  33. //
  34. durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
  36. //
  37. exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
  39. //
  40. autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
  42. // return new Queue("TestDirectQueue",true,true,false);
  44. //一般设置一下队列的持久化就好,其余两个就是默认false
  46. return new Queue("TestDirectQueue",true);
  48. }
  50. //Direct交换机 起名：TestDirectExchange
  52. \@Bean
  54. DirectExchange TestDirectExchange() {
  56. // return new DirectExchange("TestDirectExchange",true,true);
  58. return new DirectExchange("TestDirectExchange",true,false);
  60. }
  62. //绑定 将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
  64. \@Bean
  66. Binding bindingDirect() {
  68. return
  69. BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
  71. }
  73. \@Bean
  75. DirectExchange lonelyDirectExchange() {
  77. return new DirectExchange("lonelyDirectExchange");
  79. }
  81. }
复制代码
Fanout
  1. import org.springframework.amqp.core.Binding;
  3. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.FanoutExchange;
  7. import org.springframework.amqp.core.Queue;
  9. import org.springframework.context.annotation.Bean;
  11. import org.springframework.context.annotation.Configuration;
  13. /\*\*
  15. \* \@Author : JCccc
  17. \* \@CreateTime : 2019/9/3
  19. \* \@Description :
  21. \*\*/
  23. \@Configuration
  25. public class FanoutRabbitConfig {
  27. /\*\*
  29. \* 创建三个队列 ：fanout.A fanout.B fanout.C
  31. \* 将三个队列都绑定在交换机 fanoutExchange 上
  33. \* 因为是扇型交换机, 路由键无需配置,配置也不起作用
  35. \*/
  37. \@Bean
  39. public Queue queueA() {
  41. return new Queue("fanout.A");
  43. }
  45. \@Bean
  47. public Queue queueB() {
  49. return new Queue("fanout.B");
  51. }
  53. \@Bean
  55. public Queue queueC() {
  57. return new Queue("fanout.C");
  59. }
  61. \@Bean
  63. FanoutExchange fanoutExchange() {
  65. return new FanoutExchange("fanoutExchange");
  67. }
  69. \@Bean
  71. Binding bindingExchangeA() {
  73. return BindingBuilder.bind(queueA()).to(fanoutExchange());
  75. }
  77. \@Bean
  79. Binding bindingExchangeB() {
  81. return BindingBuilder.bind(queueB()).to(fanoutExchange());
  83. }
  85. \@Bean
  87. Binding bindingExchangeC() {
  89. return BindingBuilder.bind(queueC()).to(fanoutExchange());
  91. }
  93. }
复制代码
Topic
  1. import org.springframework.amqp.core.Binding;
  3. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.Queue;
  7. import org.springframework.amqp.core.TopicExchange;
  9. import org.springframework.context.annotation.Bean;
  11. import org.springframework.context.annotation.Configuration;
  13. /\*\*
  15. \* \@Author : JCccc
  17. \* \@CreateTime : 2019/9/3
  19. \* \@Description :
  21. \*\*/
  23. \@Configuration
  25. public class TopicRabbitConfig {
  27. //绑定键
  29. public final static String man = "topic.man";
  31. public final static String woman = "topic.woman";
  33. \@Bean
  35. public Queue firstQueue() {
  37. return new Queue(TopicRabbitConfig.man);
  39. }
  41. \@Bean
  43. public Queue secondQueue() {
  45. return new Queue(TopicRabbitConfig.woman);
  47. }
  49. \@Bean
  51. TopicExchange exchange() {
  53. return new TopicExchange("topicExchange");
  55. }
  57. //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
  59. //这样只要是消息携带的路由键是topic.man,才会分发到该队列
  61. \@Bean
  63. Binding bindingExchangeMessage() {
  65. return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
  67. }
  69. //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
  71. // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
  73. \@Bean
  75. Binding bindingExchangeMessage2() {
  77. return
  78. BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
  80. }
  82. }
复制代码
生产者发送消息

直接发送给队列
  1. //指定消息队列的名字，直接发送消息到消息队列中
  3. \@Test
  5. public void testSimpleQueue() {
  7. // 队列名称
  9. String queueName = "simple.queue";
  11. // 消息
  13. String message = "hello, spring amqp!";
  15. // 发送消息
  17. rabbitTemplate.convertAndSend(queueName, message);
  19. }
复制代码
发送给交换机，然后走不同的模式
  1. ////指定交换机的名字，将消息发送给交换机，然后不同模式下，消息队列根据key得到消息
  3. \@Test
  5. public void testSendDirectExchange() {
  7. // 交换机名称，有三种类型
  9. String exchangeName = "itcast.direct";
  11. // 消息
  13. String message =
  14. "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
  16. // 发送消息，red为队列的key，因此此队列会得到消息
  18. rabbitTemplate.convertAndSend(exchangeName, "red", message);
  20. }
复制代码
也可以将发送的消息封装到HashMap中然后发送给交换机
  1. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  3. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.web.bind.annotation.GetMapping;
  7. import org.springframework.web.bind.annotation.RestController;
  9. import java.time.LocalDateTime;
  11. import java.time.format.DateTimeFormatter;
  13. import java.util.HashMap;
  15. import java.util.Map;
  17. import java.util.UUID;
  19. /\*\*
  21. \* \@Author : JCccc
  23. \* \@CreateTime : 2019/9/3
  25. \* \@Description :
  27. \*\*/
  29. \@RestController
  31. public class SendMessageController {
  33. \@Autowired
  35. RabbitTemplate rabbitTemplate;
  36. //使用RabbitTemplate,这提供了接收/发送等等方法
  38. \@GetMapping("/sendDirectMessage")
  40. public String sendDirectMessage() {
  42. String messageId = String.valueOf(UUID.randomUUID());
  44. String messageData = "test message, hello!";
  46. String createTime =
  47. LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd
  48. HH:mm:ss"));
  50. Map\<String,Object\> map=new HashMap\<\>();
  52. map.put("messageId",messageId);
  54. map.put("messageData",messageData);
  56. map.put("createTime",createTime);
  58. //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
  60. rabbitTemplate.convertAndSend("TestDirectExchange",
  61. "TestDirectRouting", map);
  63. return "ok";
  65. }
  67. }
复制代码
消费者接收消息
  1. //使用注解@RabbitListener定义当前方法监听RabbitMQ中指定名称的消息队列。
  3. \@Component
  5. public class MessageListener {
  7. \@RabbitListener(queues = "direct_queue")
  9. public void receive(String id){
  11. System.out.println("已完成短信发送业务(rabbitmq direct)，id："+id);
  13. }
  15. }
  17. 参数用Map接收也可以
  19. \@Component
  21. \@RabbitListener(queues = "TestDirectQueue")//监听的队列名称
  22. TestDirectQueue
  24. public class DirectReceiver {
  26. \@RabbitHandler
  28. public void process(Map testMessage) {
  30. System.out.println("DirectReceiver消费者收到消息 : " +
  31. testMessage.toString());
  33. }
  35. }
复制代码
高级特性

消息可靠性传递

有confirm和return两种
在application.yml中添加以下配置项：
  1. server:
  3. port: 8021
  5. spring:
  7. #给项目来个名字
  9. application:
  11. name: rabbitmq-provider
  13. #配置rabbitMq 服务器
  15. rabbitmq:
  17. host: 127.0.0.1
  19. port: 5672
  21. username: root
  23. password: root
  25. #虚拟host 可以不设置,使用server默认host
  27. virtual-host: JCcccHost
  29. #确认消息已发送到交换机(Exchange)
  31. #publisher-confirms: true
  33. publisher-confirm-type: correlated
  35. #确认消息已发送到队列(Queue)
  37. publisher-returns: true
复制代码
有两种配置方法：
写到配置类中
写到工具类或者普通类中，但是这个类得实现那两个接口
写法一

编写消息确认回调函数
  1. import org.springframework.amqp.core.Message;
  3. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  5. import org.springframework.amqp.rabbit.connection.CorrelationData;
  7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  9. import org.springframework.context.annotation.Bean;
  11. import org.springframework.context.annotation.Configuration;
  13. \@Configuration
  15. public class RabbitConfig {
  17. \@Bean
  19. public RabbitTemplate createRabbitTemplate(ConnectionFactory
  20. connectionFactory){
  22. RabbitTemplate rabbitTemplate = new RabbitTemplate();
  24. rabbitTemplate.setConnectionFactory(connectionFactory);
  26. //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
  28. rabbitTemplate.setMandatory(true);
  30. rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
  32. \@Override
  34. public void confirm(CorrelationData correlationData, boolean ack, String
  35. cause) {
  37. System.out.println("ConfirmCallback:
  38. "+"相关数据："+correlationData);
  40. System.out.println("ConfirmCallback: "+"确认情况："+ack);
  42. System.out.println("ConfirmCallback: "+"原因："+cause);
  44. }
  46. });
  48. rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
  50. \@Override
  52. public void returnedMessage(Message message, int replyCode, String
  53. replyText, String exchange, String routingKey) {
  55. System.out.println("ReturnCallback: "+"消息："+message);
  57. System.out.println("ReturnCallback: "+"回应码："+replyCode);
  59. System.out.println("ReturnCallback: "+"回应信息："+replyText);
  61. System.out.println("ReturnCallback: "+"交换机："+exchange);
  63. System.out.println("ReturnCallback: "+"路由键："+routingKey);
  65. }
  67. });
  69. return rabbitTemplate;
  71. }
  73. }
复制代码
写法二
  1. \@Component
  3. \@Slf4j
  5. public class SmsRabbitMqUtils implements RabbitTemplate.ConfirmCallback,
  6. RabbitTemplate.ReturnsCallback {
  8. \@Resource
  10. private RedisTemplate\<String, String\> redisTemplate;
  12. \@Resource
  14. private RabbitTemplate rabbitTemplate;
  16. private String finalId = null;
  18. private SmsDTO smsDTO = null;
  20. /\*\*
  22. \* 发布者确认的回调
  24. \*
  26. \* \@param correlationData 回调的相关数据。
  28. \* \@param b ack为真，nack为假
  30. \* \@param s 一个可选的原因，用于nack，如果可用，否则为空。
  32. \*/
  34. \@Override
  36. public void confirm(CorrelationData correlationData, boolean b, String
  37. s) {
  39. // 消息发送成功，将redis中消息的状态（status）修改为1
  41. if (b) {
  43. redisTemplate.opsForHash().put(RedisConstant.SMS_MESSAGE_PREFIX +
  44. finalId, "status", 1);
  46. } else {
  48. // 发送失败，放入redis失败集合中，并删除集合数据
  50. log.error("短信消息投送失败：{}\--\>{}", correlationData, s);
  52. redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + finalId);
  54. redisTemplate.opsForHash().put(RedisConstant.MQ_PRODUCER, finalId,
  55. this.smsDTO);
  57. }
  59. }
  61. /\*\*
  63. \* 发生异常时的消息返回提醒
  65. \*
  67. \* \@param returnedMessage
  69. \*/
  71. \@Override
  73. public void returnedMessage(ReturnedMessage returnedMessage) {
  75. log.error("发生异常，返回消息回调:{}", returnedMessage);
  77. // 发送失败，放入redis失败集合中，并删除集合数据
  79. redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + finalId);
  81. redisTemplate.opsForHash().put(RedisConstant.MQ_PRODUCER, finalId,
  82. this.smsDTO);
  84. }
  86. \@PostConstruct
  88. public void init() {
  90. rabbitTemplate.setConfirmCallback(this);
  92. rabbitTemplate.setReturnsCallback(this);
  94. }
  96. }
复制代码
消息确认机制

手动确认
  1. yml配置
  3. #手动确认 manual
  5. listener:
  7. simple:
  9. acknowledge-mode: manual
复制代码
写法一

首先在消费者项目中创建MessageListenerConfig
  1. import com.elegant.rabbitmqconsumer.receiver.MyAckReceiver;
  3. import org.springframework.amqp.core.AcknowledgeMode;
  5. import org.springframework.amqp.core.Queue;
  7. import
  8. org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
  10. import
  11. org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
  13. import org.springframework.beans.factory.annotation.Autowired;
  15. import org.springframework.context.annotation.Bean;
  17. import org.springframework.context.annotation.Configuration;
  19. \@Configuration
  21. public class MessageListenerConfig {
  23. \@Autowired
  25. private CachingConnectionFactory connectionFactory;
  27. \@Autowired
  29. private MyAckReceiver myAckReceiver;//消息接收处理类
  31. \@Bean
  33. public SimpleMessageListenerContainer simpleMessageListenerContainer() {
  35. SimpleMessageListenerContainer container = new
  36. SimpleMessageListenerContainer(connectionFactory);
  38. container.setConcurrentConsumers(1);
  40. container.setMaxConcurrentConsumers(1);
  42. container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //
  43. RabbitMQ默认是自动确认，这里改为手动确认消息
  45. //设置一个队列
  47. container.setQueueNames("TestDirectQueue");
  49. //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
  51. //
  52. container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
  54. //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
  56. //container.setQueues(new Queue("TestDirectQueue",true));
  58. //container.addQueues(new Queue("TestDirectQueue2",true));
  60. //container.addQueues(new Queue("TestDirectQueue3",true));
  62. container.setMessageListener(myAckReceiver);
  64. return container;
  66. }
  68. }
复制代码
然后创建手动确认监听类MyAckReceiver(手动确认模式需要实现ChannelAwareMessageListener)
  1. import com.rabbitmq.client.Channel;
  3. import org.springframework.amqp.core.Message;
  5. import
  6. org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
  8. import org.springframework.stereotype.Component;
  10. import java.io.ByteArrayInputStream;
  12. import java.io.ObjectInputStream;
  14. import java.util.Map;
  16. \@Component
  18. public class MyAckReceiver implements ChannelAwareMessageListener {
  20. \@Override
  22. public void onMessage(Message message, Channel channel) throws Exception
  23. {
  25. long deliveryTag = message.getMessageProperties().getDeliveryTag();
  27. try {
  29. byte\[\] body = message.getBody();
  31. ObjectInputStream ois = new ObjectInputStream(new
  32. ByteArrayInputStream(body));
  34. Map\<String,String\> msgMap = (Map\<String,String\>) ois.readObject();
  36. String messageId = msgMap.get("messageId");
  38. String messageData = msgMap.get("messageData");
  40. String createTime = msgMap.get("createTime");
  42. ois.close();
  44. System.out.println(" MyAckReceiver messageId:"+messageId+"
  45. messageData:"+messageData+" createTime:"+createTime);
  47. System.out.println("消费的主题消息来自："+message.getMessageProperties().getConsumerQueue());
  49. channel.basicAck(deliveryTag, true);
  50. //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认
  51. delivery_tag 小于等于传入值的所有消息
  53. //channel.basicReject(deliveryTag,
  54. true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
  56. } catch (Exception e) {
  58. channel.basicReject(deliveryTag, false);
  60. e.printStackTrace();
  62. }
  64. }
  66. }
复制代码
如果想实现不同的队列，有不同的监听确认处理机制，做不同的业务处理，那么这样做：
首先需要在配置类中绑定队列，然后只需要根据消息来自不同的队列名进行区分处理即可
  1. import com.rabbitmq.client.Channel;
  3. import org.springframework.amqp.core.Message;
  5. import
  6. org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
  8. import org.springframework.stereotype.Component;
  10. import java.io.ByteArrayInputStream;
  12. import java.io.ObjectInputStream;
  14. import java.util.Map;
  16. \@Component
  18. public class MyAckReceiver implements ChannelAwareMessageListener {
  20. \@Override
  22. public void onMessage(Message message, Channel channel) throws Exception
  23. {
  25. long deliveryTag = message.getMessageProperties().getDeliveryTag();
  27. try {
  29. byte\[\] body = message.getBody();
  31. ObjectInputStream ois = new ObjectInputStream(new
  32. ByteArrayInputStream(body));
  34. Map\<String,String\> msgMap = (Map\<String,String\>) ois.readObject();
  36. String messageId = msgMap.get("messageId");
  38. String messageData = msgMap.get("messageData");
  40. String createTime = msgMap.get("createTime");
  42. ois.close();
  44. if
  45. ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
  47. System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
  49. System.out.println("消息成功消费到 messageId:"+messageId+"
  50. messageData:"+messageData+" createTime:"+createTime);
  52. System.out.println("执行TestDirectQueue中的消息的业务处理流程\...\...");
  54. }
  56. if
  57. ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
  59. System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
  61. System.out.println("消息成功消费到 messageId:"+messageId+"
  62. messageData:"+messageData+" createTime:"+createTime);
  64. System.out.println("执行fanout.A中的消息的业务处理流程\...\...");
  66. }
  68. channel.basicAck(deliveryTag, true);
  70. //channel.basicReject(deliveryTag, true);//为true会重新放回队列
  72. } catch (Exception e) {
  74. channel.basicReject(deliveryTag, false);
  76. e.printStackTrace();
  78. }
  80. }
  82. }
复制代码
写法二
  1. \@Component
  3. \@Slf4j
  5. public class SendSmsListener {
  7. \@Resource
  9. private RedisTemplate\<String, String\> redisTemplate;
  11. \@Resource
  13. private SendSmsUtils sendSmsUtils;
  15. /\*\*
  17. \* 监听发送短信普通队列
  19. \* \@param smsDTO
  21. \* \@param message
  23. \* \@param channel
  25. \* \@throws IOException
  27. \*/
  29. \@RabbitListener(queues = SMS_QUEUE_NAME)
  31. public void sendSmsListener(SmsDTO smsDTO, Message message, Channel
  32. channel) throws IOException {
  34. String messageId = message.getMessageProperties().getMessageId();
  36. int retryCount = (int)
  37. redisTemplate.opsForHash().get(RedisConstant.SMS_MESSAGE_PREFIX +
  38. messageId, "retryCount");
  40. if (retryCount \> 3) {
  42. //重试次数大于3，直接放到死信队列
  44. log.error("短信消息重试超过3次：{}", messageId);
  46. //basicReject方法拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列。
  48. //该方法reject后，该消费者还是会消费到该条被reject的消息。
  50. channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
  52. redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + messageId);
  54. return;
  56. }
  58. try {
  60. String phoneNum = smsDTO.getPhoneNum();
  62. String code = smsDTO.getCode();
  64. if(StringUtils.isAnyBlank(phoneNum,code)){
  66. throw new RuntimeException("sendSmsListener参数为空");
  68. }
  70. // 发送消息
  72. SendSmsResponse sendSmsResponse = sendSmsUtils.sendSmsResponse(phoneNum,
  73. code);
  75. SendStatus\[\] sendStatusSet = sendSmsResponse.getSendStatusSet();
  77. SendStatus sendStatus = sendStatusSet\[0\];
  79. if(!"Ok".equals(sendStatus.getCode()) \|\|!"send
  80. success".equals(sendStatus.getMessage())){
  82. throw new RuntimeException("发送验证码失败");
  84. }
  86. //手动确认消息
  88. channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  90. log.info("短信发送成功：{}",smsDTO);
  92. redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + messageId);
  94. } catch (Exception e) {
  96. redisTemplate.opsForHash().put(RedisConstant.SMS_MESSAGE_PREFIX+messageId,"retryCount",retryCount+1);
  98. channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
  100. }
  102. }
  104. /\*\*
  106. \* 监听到发送短信死信队列
  108. \* \@param sms
  110. \* \@param message
  112. \* \@param channel
  114. \* \@throws IOException
  116. \*/
  118. \@RabbitListener(queues = SMS_DELAY_QUEUE_NAME)
  120. public void smsDelayQueueListener(SmsDTO sms, Message message, Channel
  121. channel) throws IOException {
  123. try{
  125. log.error("监听到死信队列消息==\>{}",sms);
  127. channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  129. }catch (Exception e){
  131. channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
  133. }
  135. }
  137. }
复制代码
消费端限流
  1. #配置RabbitMQ
  3. spring:
  5. rabbitmq:
  7. host: 192.168.126.3
  9. port: 5672
  11. username: guest
  13. password: guest
  15. virtual-host: /
  17. #开启自动确认 none 手动确认 manual
  19. listener:
  21. simple:
  23. #消费端限流机制必须开启手动确认
  25. acknowledge-mode: manual
  27. #消费端最多拉取的消息条数，签收后不满该条数才会继续拉取
  29. prefetch: 5
复制代码
消息存活时间TTL

可以设置队列的存活时间，也可以设置具体消息的存活时间
设置队列中所有消息的存活时间
return QueueBuilder
.durable(QUEUE_NAME)//队列持久化
.ttl(10000)//设置队列的所有消息存活10s
.build();
即在创建队列时，设置存活时间
设置某条消息的存活时间
//发送消息，并设置该消息的存活时间
  1. \@Test
  3. public void testSendMessage()
  5. {
  7. //1.创建消息属性
  9. MessageProperties messageProperties = new MessageProperties();
  11. //2.设置存活时间
  13. messageProperties.setExpiration("10000");
  15. //3.创建消息对象
  17. Message message = new
  18. Message("sendMessage\...".getBytes(),messageProperties);
  20. //4.发送消息
  22. rabbitTemplate.convertAndSend("my_topic_exchange1","my_routing",message);
  24. }
复制代码
若设置中间的消息的存活时间，当过期时，该消息不会被移除，但是该消息已经不会被消费了，需要等到该消息到队里顶端才会被移除。因为队列是头出，尾进，故而要移除它需要等到它在顶端时才可以。
在队列设置存活时间，也在单条消息设置存活时间，则以时间短的为准
死信队列

死信队列和普通队列没有任何区别，只需要将普通队列需要绑定死信交换机和死信队列就能够实现功能
  1. import org.springframework.amqp.core.\*;
  3. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  9. \@Configuration//Rabbit配置类
  11. public class RabbitConfig4 {
  13. private final String DEAD_EXCHANGE = "dead_exchange";
  15. private final String DEAD_QUEUE = "dead_queue";
  17. private final String NORMAL_EXCHANGE = "normal_exchange";
  19. private final String NORMAL_QUEUE = "normal_queue";
  21. //创建死信交换机
  23. \@Bean(DEAD_EXCHANGE)
  25. public Exchange deadExchange()
  27. {
  29. return ExchangeBuilder
  31. .topicExchange(DEAD_EXCHANGE)//交换机类型 ;参数为名字
  32. topic为通配符模式的交换机
  34. .durable(true)//是否持久化，true即存到磁盘,false只在内存上
  36. .build();
  38. }
  40. //创建死信队列
  42. \@Bean(DEAD_QUEUE)
  44. public Queue deadQueue()
  46. {
  48. return QueueBuilder
  50. .durable(DEAD_QUEUE)//队列持久化
  52. //.maxPriority(10)//设置队列的最大优先级，最大可以设置255，但官网推荐不超过10，太高比较浪费资源
  54. .build();
  56. }
  58. //死信交换机绑定死信队列
  60. \@Bean
  62. //@Qualifier注解,使用名称装配进行使用
  64. public Binding bindDeadQueue(@Qualifier(DEAD_EXCHANGE) Exchange
  65. exchange, \@Qualifier(DEAD_QUEUE) Queue queue)
  67. {
  69. return BindingBuilder
  71. .bind(queue)
  73. .to(exchange)
  75. .with("dead_routing")
  77. .noargs();
  79. }
  81. //创建普通交换机
  83. \@Bean(NORMAL_EXCHANGE)
  85. public Exchange normalExchange()
  87. {
  89. return ExchangeBuilder
  91. .topicExchange(NORMAL_EXCHANGE)//交换机类型 ;参数为名字
  92. topic为通配符模式的交换机
  94. .durable(true)//是否持久化，true即存到磁盘,false只在内存上
  96. .build();
  98. }
  100. //创建普通队列
  102. \@Bean(NORMAL_QUEUE)
  104. public Queue normalQueue()
  106. {
  108. return QueueBuilder
  110. .durable(NORMAL_QUEUE)//队列持久化
  112. //.maxPriority(10)//设置队列的最大优先级，最大可以设置255，但官网推荐不超过10，太高比较浪费资源
  114. .deadLetterExchange(DEAD_EXCHANGE)//绑定死信交换机
  116. .deadLetterRoutingKey("dead_routing")//死信队列路由关键字
  118. .ttl(10000)//消息存活10s
  120. .maxLength(10)//队列最大长度为10
  122. .build();
  124. }
  126. //普通交换机绑定普通队列
  128. \@Bean
  130. //@Qualifier注解,使用名称装配进行使用
  132. public Binding bindNormalQueue(@Qualifier(NORMAL_EXCHANGE) Exchange
  133. exchange, \@Qualifier(NORMAL_QUEUE) Queue queue)
  135. {
  137. return BindingBuilder
  139. .bind(queue)
  141. .to(exchange)
  143. .with("my_routing")
  145. .noargs();
  147. }
  149. }
复制代码
延迟队列

RabbitMQ并未实现延迟队列功能，所以可以通过死信队列实现延迟队列的功能
即给普通队列设置存活时间30分钟，过期后发送至死信队列，在死信消费者监听死信队列消息，查看订单状态，是否支付，未支付则取消订单，回退库存即可。
消费者监听延迟队列
  1. \@Component
  3. public class ExpireOrderConsumer {
  5. //监听过期订单队列
  7. \@RabbitListener(queues = "expire_queue")
  9. public void listenMessage(String orderId)
  11. {
  13. //模拟处理数据库等业务
  15. System.out.println("查询"+orderId+"号订单的状态，如果已支付无需处理，如果未支付则回退库存");
  17. }
  19. }
  21. 控制层代码
  23. \@RestController
  25. public class OrderController {
  27. \@Autowired
  29. private RabbitTemplate rabbitTemplate;
  31. \@RequestMapping(value = "/place/{orderId}",method =
  32. RequestMethod.GET)
  34. public String placeOrder(@PathVariable String orderId)
  36. {
  38. //模拟service层处理
  40. System.out.println("处理订单数据\...");
  42. //将订单id发送到订单队列
  44. rabbitTemplate.convertAndSend("order_exchange","order_routing",orderId);
  46. return "下单成功，修改库存";
  48. }
  50. }
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

曹旭辉

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表