本文主要是介绍SpringBoot集成消息队列，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
背景
最近在对公司开发框架进行优化，框架内涉及到多处入库的日志记录，例如登录日志/操作日志/访问日志/业务执行日志，集成在业务代码中耦合度较高且占用业务操作执行时间，所以准备集成相关消息队列进行代码解耦
方案规划
现有的成熟消息队列组件非常多，例如RabbitMQ，ActiveMQ，Kafka等，考虑到业务并发量不高且框架已经应用于多个项目平稳运行，准备提供基于Redis的消息队列和集成ActiveMQ两种方案，Redis消息队列的好处是无需额外安装部署存量项目可平稳过度但消息无法持久化可能丢失，ActiveMQ解决方案成熟可以保证消息持久化但是需要实施人员额外掌握操作部署
统一设计
增加自定义配置指定消息队列方式
system:#消息队列方式 redis/activemqmessageChannel: redis
定义消息传输统一模型
public class MessageModel {private Class<? extends IMessageReceiver> handleClazz;private String bodyContent;private Class bodyClass;private HashMap extraParam;public MessageModel(){extraParam = new HashMap();}public Class<? extends IMessageReceiver> getHandleClazz() {return handleClazz;}public void setHandleClazz(Class<? extends IMessageReceiver> 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 {@Autowiredprivate SystemConfig systemConfig;@Autowiredprivate RedisUtil redisUtil;@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;public void sendMessage(Object messageBody, Class<? extends IMessageReceiver> handleClass, HashMap<String,Object> 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配置
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>2.0.1.RELEASE</version></dependency>
连接配置
spring:redis:host: localhostport: 6379password:
操作工具类
@Autowiredprivate RedisTemplate redisTemplate;public void sendMessage(String channel, Object message) {redisTemplate.convertAndSend(channel, message);}
消息处理
@Component
@ConditionalOnProperty(name = "system.messageChannel", havingValue = "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 = "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 = "redis", matchIfMissing = true)MessageListenerAdapter listenerAdapter(RedisMessageReceiver receiver) {return new MessageListenerAdapter(receiver, "receiveMessage");}
}
集成ActiveMQ消息队列
pom配置
<dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId><version>5.15.0</version></dependency>
注意：jdk1.8对应版本5.15.0
连接配置
spring:activemq:broker-url: tcp://127.0.0.1:61616 #MQ服务器地址user: adminpassword: adminpool:enabled: true
消息处理
@Component
@ConditionalOnProperty(name = "system.messageChannel", havingValue = "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 = "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{@Autowiredprivate F_RequestLogService requestLogService;@Overridepublic void handleMessage(Object bodyObject, HashMap extraParam) {F_RequestLogDO requestLogDO = (F_RequestLogDO)bodyObject;requestLogService.insert(requestLogDO);}
}
发送消息
@AutoWired
private MessageUtil messageUtil;
messageUtil.sendMessage(requestLogDO,RequestLogMessageReceiver.class,null);
这篇关于SpringBoot集成消息队列的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
