SpringBoot集成消息队列的项目实践
更新时间：2024年02月21日 10:42:02 作者：清醒的人最荒唐
本文主要介绍了SpringBoot集成消息队列的项目实践,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
背景
最近在对公司开发框架进行优化，框架内涉及到多处入库的日志记录，例如登录日志/操作日志/访问日志/业务执行日志，集成在业务代码中耦合度较高且占用业务操作执行时间，所以准备集成相关消息队列进行代码解耦
方案规划
现有的成熟消息队列组件非常多，例如RabbitMQ，ActiveMQ，Kafka等，考虑到业务并发量不高且框架已经应用于多个项目平稳运行，准备提供基于Redis的消息队列和集成ActiveMQ两种方案，Redis消息队列的好处是无需额外安装部署存量项目可平稳过度但消息无法持久化可能丢失，ActiveMQ解决方案成熟可以保证消息持久化但是需要实施人员额外掌握操作部署
统一设计
增加自定义配置指定消息队列方式
system: #消息队列方式 redis/activemq messageChannel: 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 {
@Autowired
private SystemConfig systemConfig;
@Autowired
private RedisUtil redisUtil;
@Autowired
private 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: 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 = "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: admin
password: admin
pool:
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{
@Autowired
private F_RequestLogService requestLogService;
@Override
public 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集成消息队列的项目实践的文章就介绍到这了,更多相关SpringBoot集成消息队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了Mybatis generator自动生成代码插件实例解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2020-09-09
字符串拼接问题应该是每个Java程序员都熟知的事情了，几乎每个Java程序员都读过关于StringBuffer/StringBuilder来拼接字符串。下面这篇文章主要给大家介绍了关于JDK8中的字符串拼接的相关资料，需要的朋友可以参考下。2018-04-04
这篇文章主要为大家详细介绍了java动态代理模式，总结一下代理模式，以及jdk,cglib代理模式用法，来理解代理模式，感兴趣的小伙伴们可以参考一下2016-02-02
本文主要介绍了IntelliJ IDEA快速查看某个类/接口的子类或父类，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2022-07-07
这篇文章主要给大家介绍了关于java四舍五入方法的基础学习,这是最近做算法题的时候碰到的这个问题,文中通过实例代码介绍的非常详细,需要的朋友可以参考下2023-07-07
这篇文章主要给大家介绍了关于Java8中如何通过方法引用获取属性名的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2020-09-09
这篇文章主要介绍了java正则匹配HTML中a标签里的中文字符,涉及java中文正则及HTML元素操作技巧,具有一定参考借鉴价值,需要的朋友可以参考下2017-01-01
Drools 就是一个开源的业务规则引擎，可以很容易地与 spring boot 应用程序集成，这篇文章就来和大家详细聊聊Drools的具体应用，需要的可以参考一下2023-03-03
通常注册账户上传用户图像时需要进行预览，这篇文章就是教我们如何用 Java 实现图片裁剪预览功能，需要的朋友可以参考下2015-07-07
这篇文章主要介绍了java通过模拟post方式提交表单实现图片上传功能实例,涉及Java针对表单的提交操作响应及文件传输的相关技巧,具有一定参考借鉴价值,需要的朋友可以参考下2015-11-11
最新评论