SpringBoot集成消息队列

code2roc

发布于 2023-07-19 15:27:02

3170

发布于 2023-07-19 15:27:02

文章被收录于专栏：IT民工的代码世界

背景

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

方案规划

现有的成熟消息队列组件非常多，例如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);

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2022-04-23，如有侵权请联系 cloudcommunity@tencent.com 删除

public

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

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

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

登录后参与评论

0 条评论

热度

SpringBoot集成消息队列

SpringBoot集成消息队列

背景

方案规划

统一设计

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

定义消息传输统一模型

定义标准消息处理接口

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

集成Redis消息队列

pom配置

连接配置

操作工具类

消息处理

配置注册

集成ActiveMQ消息队列

pom配置

连接配置

消息处理

配置注册

使用示例

消息处理

发送消息

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐