SpringBoot集成消息队列 springboot消息队列

背景最近在对公司开发框架进行优化，框架内涉及到多处入库的日志记录，例如登录日志/操作日志/访问日志/业务执行日志，集成在业务代码中耦合度较高且占用业务操作执行时间，所以准备集成相关消息队列进行代码解耦
方案规划 【SpringBoot集成消息队列】现有的成熟消息队列组件非常多，例如RabbitMQ，ActiveMQ，Kafka等，考虑到业务并发量不高且框架已经应用于多个项目平稳运行，准备提供基于Redis的消息队列和集成ActiveMQ两种方案，Redis消息队列的好处是无需额外安装部署存量项目可平稳过度但消息无法持久化可能丢失，ActiveMQ解决方案成熟可以保证消息持久化但是需要实施人员额外掌握操作部署
统一设计增加自定义配置指定消息队列方式

system: #消息队列方式 redis/activemq messageChannel: redis

定义消息传输统一模型

public class MessageModel { private Class handleClazz; private String bodyContent; private Class bodyClass; private HashMap extraParam; public MessageModel(){ extraParam = new HashMap(); }public Class getHandleClazz() { return handleClazz; }public void setHandleClazz(Class handleClazz) { this.handleClazz = handleClazz; }public HashMap getExtraParam() { return extraParam; }public void setExtraParam(HashMap extraParam) { this.extraParam = extraParam; }public String getBodyContent() { return bodyContent; }public void setBodyContent(String bodyContent) { this.bodyContent = bodyContent; }public Class getBodyClass() { return bodyClass; }public void setBodyClass(Class bodyClass) { this.bodyClass = bodyClass; } }

定义标准消息处理接口

public interface IMessageReceiver { void handleMessage(Object bodyObject, HashMap extraParam); }

定义统一对外发送消息工具类

@Component public class MessageUtil { @Autowired private SystemConfig systemConfig; @Autowired private RedisUtil redisUtil; @Autowired private JmsMessagingTemplate jmsMessagingTemplate; public void sendMessage(Object messageBody, Class handleClass, HashMap extraParam) { MessageModel messageModel = new MessageModel(); messageModel.setHandleClazz(handleClass); messageModel.setBodyClass(messageBody.getClass()); messageModel.setBodyContent(JSON.toJSONString(messageBody)); if (extraParam != null) { for (String key:extraParam.keySet()) { messageModel.getExtraParam().put(key,extraParam.get(key)); } } if(systemConfig.getMessageChannel().equals("redis")){ redisUtil.sendMessage("message", JSON.toJSON(messageModel)); }else{ jmsMessagingTemplate.convertAndSend("message",JSON.toJSONString(messageModel)); } } }

集成Redis消息队列 pom配置

org.springframework.boot spring-boot-starter-data-redis 2.0.1.RELEASE

连接配置

spring: redis: host: localhost port: 6379 password:

操作工具类

@Autowired private RedisTemplate redisTemplate; public void sendMessage(String channel, Object message) { redisTemplate.convertAndSend(channel, message); }

消息处理

@Component @ConditionalOnProperty(name = "system.messageChannel", havingValue = "https://www.it610.com/article/redis", matchIfMissing = true) public class RedisMessageReceiver { public void receiveMessage(String message) { MessageModel messageModel = JSON.parseObject(message, MessageModel.class); IMessageReceiver receiver = SpringBootBeanUtil.getBean(messageModel.getHandleClazz()); receiver.handleMessage(JSON.parseObject(messageModel.getBodyContent(), messageModel.getBodyClass()), messageModel.getExtraParam()); } }

配置注册

@Configuration public class MessageCenter { @Bean @ConditionalOnProperty(name = "system.messageChannel", havingValue = "https://www.it610.com/article/redis", matchIfMissing = true) RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); // 可以添加多个 messageListener，配置不同的交换机 container.addMessageListener(listenerAdapter, new PatternTopic("message")); return container; }/** * 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法 * * @param receiver * @return */ @Bean @ConditionalOnProperty(name = "system.messageChannel", havingValue = "https://www.it610.com/article/redis", matchIfMissing = true) MessageListenerAdapter listenerAdapter(RedisMessageReceiver receiver) { return new MessageListenerAdapter(receiver, "receiveMessage"); } }

集成ActiveMQ消息队列 pom配置

org.apache.activemq activemq-pool 5.15.0

注意：jdk1.8对应版本5.15.0
连接配置

spring: activemq: broker-url: tcp://127.0.0.1:61616 #MQ服务器地址 user: admin password: admin pool: enabled: true

消息处理

@Component @ConditionalOnProperty(name = "system.messageChannel", havingValue = "https://www.it610.com/article/activemq", matchIfMissing = false) public class ActiveMQMessageReceiver { @JmsListener(destination = "message", containerFactory = "customQueueListener") public void handleMessage(String message) { MessageModel messageModel = JSON.parseObject(message, MessageModel.class); IMessageReceiver receiver = SpringBootBeanUtil.getBean(messageModel.getHandleClazz()); receiver.handleMessage(JSON.parseObject(messageModel.getBodyContent(), messageModel.getBodyClass()), messageModel.getExtraParam()); } }

配置注册

@Configuration @EnableJms public class MessageCenter { @Bean(name = "customQueueListener") @ConditionalOnProperty(name = "system.messageChannel", havingValue = "https://www.it610.com/article/activemq", matchIfMissing = false) public JmsListenerContainerFactory customQueueListener(ConnectionFactory connectionFactory) { DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); factory.setPubSubDomain(false); factory.setConnectionFactory(connectionFactory); //重连间隔时间 factory.setRecoveryInterval(1000L); factory.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE); //连接数 factory.setConcurrency("5-10"); //指定任务线程池 factory.setTaskExecutor(new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.CallerRunsPolicy())); return factory; } }

使用示例消息处理

@Service public class RequestLogMessageReceiver implements IMessageReceiver{ @Autowired private F_RequestLogService requestLogService; @Override public void handleMessage(Object bodyObject, HashMap extraParam) { F_RequestLogDO requestLogDO = (F_RequestLogDO)bodyObject; requestLogService.insert(requestLogDO); } }

发送消息