spring boot Rabbitmq集成,延时消息队列实现

本篇主要记录Spring boot 集成Rabbitmq,分为两部分, 第一部分为创建普通消息队列, 第二部分为延时消息队列实现:

spring boot提供对mq消息队列支持amqp相关包,引入即可:

[html] view plain copy
 
  1. <!-- rabbit mq -->
  2. <dependency>
  3. <groupId>org.springframework.boot</groupId>
  4. <artifactId>spring-boot-starter-amqp</artifactId>
  5. </dependency>

属性配置文件application.properties:

[plain] view plain copy
 
  1. #rabbitmq
  2. spring.rabbitmq.host=127.0.0.1
  3. spring.rabbitmq.port=5672
  4. spring.rabbitmq.username=root
  5. spring.rabbitmq.password=root

RabbitMq配置类,配置连接工厂以及操作对象:

[java] view plain copy
 
  1. @Configuration
  2. @ConfigurationProperties(prefix = "spring.rabbitmq")
  3. public class RabbitMQConfiguration {
  4. private static Logger logger = Logger.getLogger(RabbitMQConfiguration.class);
  5. private String host;
  6. private int port;
  7. private String username;
  8. private String password;
  9. // 链接信息
  10. @Bean
  11. public ConnectionFactory connectionFactory() {
  12. CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
  13. connectionFactory.setUsername(username);
  14. connectionFactory.setPassword(password);
  15. connectionFactory.setVirtualHost("/");
  16. connectionFactory.setPublisherConfirms(true);
  17. logger.info("Create ConnectionFactory bean ..");
  18. return connectionFactory;
  19. }
  20. @Bean
  21. @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  22. public RabbitTemplate rabbitTemplate() {
  23. RabbitTemplate template = new RabbitTemplate(connectionFactory());
  24. return template;
  25. }
  26. //省略getter setter
[java] view plain copy
 
  1. }

定义Service接口如下:

暂时不考虑延时队列,定义发送消息接口

[java] view plain copy
 
  1. /**
  2. *
  3. * @author victor
  4. * @desc 消息队列服务接口
  5. */
  6. public interface IMessageQueueService {
  7. /**
  8. * 发送消息到队列
  9. * @param queue 队列名称
  10. * @param message 消息内容
  11. */
  12. public void send(String queueName,String message);
  13. }

Service实现

[java] view plain copy
 
  1. package com.ks.server.service.impl.queue;
  2. import org.springframework.amqp.AmqpException;
  3. import org.springframework.amqp.core.Message;
  4. import org.springframework.amqp.core.MessagePostProcessor;
  5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Service;
  8. import com.base.common.codec.JSONUtils;
  9. import com.ks.common.constant.MQConstant;
  10. import com.ks.common.service.queue.IMessageQueueService;
  11. import com.ks.modal.queue.DLXMessage;
  12. /**
  13. *
  14. * @author victor
  15. * @desc 消息队列服务接口实现
  16. */
  17. @Service("messageQueueService")
  18. public class MessageQueueServiceImpl implements IMessageQueueService{
  19. @Autowired
  20. private RabbitTemplate rabbitTemplate;
  21. @Override
  22. public void send(String queueName, String msg) {
  23. rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE,queueName, msg);
  24. }
  25. }

相关常量类:

[java] view plain copy
 
  1. package com.ks.common.constant;
  2. /**
  3. *
  4. * @author victor
  5. * @desc Rabbit消息队列相关常量
  6. */
  7. public final class MQConstant {
  8. private MQConstant(){
  9. }
  10. //exchange name
  11. public static final String DEFAULT_EXCHANGE = "KSHOP";
  12. //DLX QUEUE
  13. public static final String DEFAULT_DEAD_LETTER_QUEUE_NAME = "kshop.dead.letter.queue";
  14. //DLX repeat QUEUE 死信转发队列
  15. public static final String DEFAULT_REPEAT_TRADE_QUEUE_NAME = "kshop.repeat.trade.queue";
  16. //Hello 测试消息队列名称
  17. public static final String HELLO_QUEUE_NAME = "HELLO";
  18. }

到现在为止,队列相关配置,以及使用以及封装完成,接下来是创建队列,

这里我是单独创建一个配置类,用于队列配置, 创建Hello队列示例如下:

[java] view plain copy
 
  1. package com.ks.ons.config;
  2. import java.util.HashMap;
  3. import java.util.Map;
  4. import org.springframework.amqp.core.Binding;
  5. import org.springframework.amqp.core.BindingBuilder;
  6. import org.springframework.amqp.core.DirectExchange;
  7. import org.springframework.amqp.core.Queue;
  8. import org.springframework.context.annotation.Bean;
  9. import org.springframework.context.annotation.Configuration;
  10. import com.ks.common.constant.MQConstant;
  11. /**
  12. *
  13. * @author victor
  14. * @desc 队列配置
  15. */
  16. @Configuration
  17. public class QueueConfiguration {
  18. //信道配置
  19. @Bean
  20. public DirectExchange defaultExchange() {
  21. return new DirectExchange(MQConstant.DEFAULT_EXCHANGE, true, false);
  22. }
  23. /*********************    hello 队列  测试    *****************/
  24. @Bean
  25. public Queue queue() {
  26. Queue queue = new Queue(MQConstant.HELLO_QUEUE_NAME,true);
  27. return queue;
  28. }
  29. @Bean
  30. public Binding binding() {
  31. return BindingBuilder.bind(queue()).to(defaultExchange()).with(MQConstant.HELLO_QUEUE_NAME);
  32. }
  33. }

通过配置队列bean,在程序启动时会在rabbitmq中创建相关队列,启动程序,可以在rabbtmq管理界面看到信道和队列信息:

spring boot Rabbitmq集成,延时消息队列实现

spring boot Rabbitmq集成,延时消息队列实现

众所周知,既然有了队列,用来处理业务的最终还是需要消费者,消费者创建示例如下:

[java] view plain copy
 
  1. package com.ks.ons.processor.hello;
  2. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  5. import com.ks.common.constant.MQConstant;
  6. /**
  7. *
  8. * @author victor
  9. * @desc hello 消息队列消费者
  10. */
  11. @Component
  12. @RabbitListener(queues = MQConstant.HELLO_QUEUE_NAME)
  13. public class HelloProcessor {
  14. @RabbitHandler
  15. public void process(String content) {
  16. System.out.println("接受消息:" + content);
  17. }
  18. }

注入service

[java] view plain copy
 
  1. @Autowired
  2. private IMessageQueueService messageQueueService;

发送消息

[java] view plain copy
 
  1. messageQueueService.send(MQConstant.HELLO_QUEUE_NAME, "测试发送消息");

接下来展示如何实现延时队列,在此之前如果读者像我一样对rabbitmq队列了解程度并不深入的话,-->推荐文章, 可以对rabbitmq延时队列实现思路有大概了解.

在本文中,主要是通过rabbitmq的DLX特性来实现发送延时队列:

思路如下:

spring boot Rabbitmq集成,延时消息队列实现

客户端:指具体往MQ发生消息端, 客户端将消息内容进行自定义包装, 将消息中附带目标队列名称。如:客户端向队列Q1发送字符串“hello” , 延时时间为60秒, 包装后修改为{"queueName":"Q1","body": “hello”},此时,将消息发送到DLX死信队列,而非Q1队列,并将消息设置为60秒超时。

DLX:死信队列,用来存储有超时时间信息的消息, 并且可以设置当消息超时时,转发到另一个指定队列(此处设置转发到router), 无消费者,当接收到客户端消息之后,等待消息超时,将消息转发到指定的Router队列

Router: 转发队列,用来接收死信队列超时消息, 如上示例消息,在接收到之后,消费者将消息解析,获取queueName,body,再向所获取的queueName队列发送一条消息,内容为body.

Q1,Q2,Q3.: 用户业务队列,当Q1收到hello,已经是60秒之后,再进行消费

修改上面代码 , 新增两个队列,

死信队列:存放发送的延时消息,

路由转发队列:用于接受死信消息死亡, 并将消息转发到业务目标队列

修改之后代码如下:

[java] view plain copy
 
  1. /**
  2. *
  3. * @author victor
  4. * @desc 队列配置
  5. */
  6. @Configuration
  7. public class QueueConfiguration {
  8. //信道配置
  9. @Bean
  10. public DirectExchange defaultExchange() {
  11. return new DirectExchange(MQConstant.DEFAULT_EXCHANGE, true, false);
  12. }
  13. @Bean
  14. public Queue repeatTradeQueue() {
  15. Queue queue = new Queue(MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME,true,false,false);
  16. return queue;
  17. }
  18. @Bean
  19. public Binding  drepeatTradeBinding() {
  20. return BindingBuilder.bind(repeatTradeQueue()).to(defaultExchange()).with(MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME);
  21. }
  22. @Bean
  23. public Queue deadLetterQueue() {
  24. Map<String, Object> arguments = new HashMap<>();
  25. arguments.put("x-dead-letter-exchange", MQConstant.DEFAULT_EXCHANGE);
  26. arguments.put("x-dead-letter-routing-key", MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME);
  27. Queue queue = new Queue(MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME,true,false,false,arguments);
  28. System.out.println("arguments :" + queue.getArguments());
  29. return queue;
  30. }
  31. @Bean
  32. public Binding  deadLetterBinding() {
  33. return BindingBuilder.bind(deadLetterQueue()).to(defaultExchange()).with(MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME);
  34. }
  35. /*********************    hello 队列  测试    *****************/
  36. @Bean
  37. public Queue queue() {
  38. Queue queue = new Queue(MQConstant.HELLO_QUEUE_NAME,true);
  39. return queue;
  40. }
  41. @Bean
  42. public Binding binding() {
  43. return BindingBuilder.bind(queue()).to(defaultExchange()).with(MQConstant.HELLO_QUEUE_NAME);
  44. }
  45. }

修改Service服务:

[java] view plain copy
 
  1. package com.ks.common.service.queue;
  2. /**
  3. *
  4. * @author victor
  5. * @desc 消息队列服务接口
  6. */
  7. public interface IMessageQueueService {
  8. /**
  9. * 发送消息到队列
  10. * @param queue 队列名称
  11. * @param message 消息内容
  12. */
  13. public void send(String queueName,String message);
  14. /**
  15. * 延迟发送消息到队列
  16. * @param queue 队列名称
  17. * @param message 消息内容
  18. * @param times 延迟时间 单位毫秒
  19. */
  20. public void send(String queueName,String message,long times);
  21. }
[java] view plain copy
 
  1. package com.ks.server.service.impl.queue;
  2. import org.springframework.amqp.AmqpException;
  3. import org.springframework.amqp.core.Message;
  4. import org.springframework.amqp.core.MessagePostProcessor;
  5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Service;
  8. import com.base.common.codec.JSONUtils;
  9. import com.ks.common.constant.MQConstant;
  10. import com.ks.common.service.queue.IMessageQueueService;
  11. import com.ks.modal.queue.DLXMessage;
  12. /**
  13. *
  14. * @author victor
  15. * @desc 消息队列服务接口实现
  16. */
  17. @Service("messageQueueService")
  18. public class MessageQueueServiceImpl implements IMessageQueueService{
  19. @Autowired
  20. private RabbitTemplate rabbitTemplate;
  21. @Override
  22. public void send(String queueName, String msg) {
  23. rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE,queueName, msg);
  24. }
  25. @Override
  26. public void send(String queueName, String msg, long times) {
  27. DLXMessage dlxMessage = new DLXMessage(queueName,msg,times);
  28. MessagePostProcessor processor = new MessagePostProcessor(){
  29. @Override
  30. public Message postProcessMessage(Message message) throws AmqpException {
  31. message.getMessageProperties().setExpiration(times + "");
  32. return message;
  33. }
  34. };
  35. dlxMessage.setExchange(MQConstant.DEFAULT_EXCHANGE);
  36. rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE,MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME, JSONUtils.toJson(dlxMessage), processor);
  37. }
  38. }

JSONUtils 为一个JSON工具类

新增消息实体,用于包装消息:

[java] view plain copy
 
  1. package com.ks.modal.queue;
  2. import java.io.Serializable;
  3. /**
  4. *
  5. * @author victor
  6. * @desc rabbit 死信消息载体
  7. */
  8. public class DLXMessage implements Serializable {
  9. private static final long serialVersionUID = 9956432152000L;
  10. public DLXMessage() {
  11. super();
  12. }
  13. public DLXMessage(String queueName, String content, long times) {
  14. super();
  15. this.queueName = queueName;
  16. this.content = content;
  17. this.times = times;
  18. }
  19. public DLXMessage(String exchange, String queueName, String content, long times) {
  20. super();
  21. this.exchange = exchange;
  22. this.queueName = queueName;
  23. this.content = content;
  24. this.times = times;
  25. }
  26. private String exchange;
  27. private String queueName;
  28. private String content;
  29. private long times;
  30. //省略getter setter
  31. }

路由转发队列消费者实现,负责接收超时消息,进行转发:

[java] view plain copy
 
  1. package com.ks.ons.processor.system;
  2. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.stereotype.Component;
  6. import com.base.common.codec.JSONUtils;
  7. import com.ks.common.constant.MQConstant;
  8. import com.ks.common.service.queue.IMessageQueueService;
  9. import com.ks.modal.queue.DLXMessage;
  10. /**
  11. *
  12. * @author victor
  13. * @desc 死信接收处理消费者
  14. */
  15. @Component
  16. @RabbitListener(queues = MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME)
  17. public class TradeProcessor {
  18. @Autowired
  19. private IMessageQueueService messageQueueService;
  20. @RabbitHandler
  21. public void process(String content) {
  22. DLXMessage message = JSONUtils.toBean(content, DLXMessage.class);
  23. messageQueueService.send(message.getQueueName(), message.getContent());
  24. }
  25. }

启动项目之后,管理界面如下:

spring boot Rabbitmq集成,延时消息队列实现

测试代码片段:

[java] view plain copy
 
    1. messageQueueService.send(MQConstant.HELLO_QUEUE_NAME,"测试延迟发送消息",60000);
上一篇:消息中间件-activemq实战之消息持久化(六)


下一篇:Spring boot实战项目整合阿里云RocketMQ (非开源版)消息队列实现发送普通消息,延时消息 --附代码