前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >SpringBoot集成消息队列

SpringBoot集成消息队列

作者头像
code2roc
发布2023-07-19 15:27:02
2460
发布2023-07-19 15:27:02
举报

背景

最近在对公司开发框架进行优化,框架内涉及到多处入库的日志记录,例如登录日志/操作日志/访问日志/业务执行日志,集成在业务代码中耦合度较高且占用业务操作执行时间,所以准备集成相关消息队列进行代码解耦

方案规划

现有的成熟消息队列组件非常多,例如RabbitMQ,ActiveMQ,Kafka等,考虑到业务并发量不高且框架已经应用于多个项目平稳运行,准备提供基于Redis的消息队列和集成ActiveMQ两种方案,Redis消息队列的好处是无需额外安装部署存量项目可平稳过度但消息无法持久化可能丢失,ActiveMQ解决方案成熟可以保证消息持久化但是需要实施人员额外掌握操作部署

统一设计

增加自定义配置指定消息队列方式

代码语言:javascript
复制
system:
  #消息队列方式 redis/activemq
  messageChannel: redis

定义消息传输统一模型

代码语言:javascript
复制
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;
    }
}

定义标准消息处理接口

代码语言:javascript
复制
public interface IMessageReceiver {
    void handleMessage(Object bodyObject, HashMap extraParam);
}

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

代码语言:javascript
复制
@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配置

代码语言:javascript
复制
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
            <version>2.0.1.RELEASE</version>
        </dependency>

连接配置

代码语言:javascript
复制
spring:
  redis:
    host: localhost
    port: 6379
    password:

操作工具类

代码语言:javascript
复制
    @Autowired
    private RedisTemplate redisTemplate;
    public void sendMessage(String channel, Object message) {
        redisTemplate.convertAndSend(channel, message);
    }

消息处理

代码语言:javascript
复制
@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());
    }
}

配置注册

代码语言:javascript
复制
@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配置

代码语言:javascript
复制
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.0</version>
        </dependency>

注意:jdk1.8对应版本5.15.0

连接配置

代码语言:javascript
复制
spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616 #MQ服务器地址
    user: admin
    password: admin
    pool:
      enabled: true

消息处理

代码语言:javascript
复制
@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());
    }
}

配置注册

代码语言:javascript
复制
@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;
    }
}

使用示例

消息处理

代码语言:javascript
复制
@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);
    }
}

发送消息

代码语言:javascript
复制
@AutoWired
private MessageUtil messageUtil;
messageUtil.sendMessage(requestLogDO,RequestLogMessageReceiver.class,null);
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2022-04-23,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 背景
  • 方案规划
  • 统一设计
    • 增加自定义配置指定消息队列方式
      • 定义消息传输统一模型
        • 定义标准消息处理接口
          • 定义统一对外发送消息工具类
          • 集成Redis消息队列
            • pom配置
              • 连接配置
                • 操作工具类
                  • 消息处理
                    • 配置注册
                    • 集成ActiveMQ消息队列
                      • pom配置
                        • 连接配置
                          • 消息处理
                            • 配置注册
                            • 使用示例
                              • 消息处理
                                • 发送消息
                                相关产品与服务
                                消息队列
                                腾讯云消息队列 TDMQ 是分布式架构中的重要组件,提供异步通信的基础能力,通过应用解耦降低系统复杂度,提升系统可用性和可扩展性。TDMQ 产品系列提供丰富的产品形态,包含 CKafka、RocketMQ、RabbitMQ、Pulsar、CMQ 五大产品,覆盖在线和离线场景,满足金融、互联网、教育、物流、能源等不同行业和场景的需求。
                                领券
                                问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档