脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → SpringBoot消息队列

详解SpringBoot集成消息队列的案例应用

 更新时间：2022年04月25日 11:20:23   作者：code2roc  
Message Queue又名消息队列，是一种异步通讯的中间件。可以理解为邮局，发送者将消息投递到邮局，然后邮局帮我们发送给具体的接收者，具体发送过程和时间与我们无关。 消息队列是分布式系统中重要的组件，消息队列主要解决了应用耦合、异步处理、流量削锋等问题

背景

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

方案规划

现有的成熟消息队列组件非常多，例如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消息队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Spring-boot原理及spring-boot-starter实例和代码

    Spring-boot原理及spring-boot-starter实例和代码

    spring-boot的starter是一个通过maven完成自包含并通过annotation配置使得可被spring上下文发现并实例化的一个可插拔的组件或服务。这篇文章主要介绍了Spring-boot原理及spring-boot-starter实例和代码 ,需要的朋友可以参考下
    2019-06-06
  • java中Date和Timestamp类型的相互转换方式

    java中Date和Timestamp类型的相互转换方式

    这篇文章主要介绍了java中Date和Timestamp类型的相互转换方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2023-07-07
  • MyBatis写入Json字段以及Json字段转对象示例详解

    MyBatis写入Json字段以及Json字段转对象示例详解

    这篇文章主要给大家介绍了关于MyBatis写入Json字段以及Json字段转对象的相关资料,文中通过实例代码介绍的非常详细,对大家的学习或者工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2023-07-07
  • Java分布式锁、分布式ID和分布式事务的实现方案

    Java分布式锁、分布式ID和分布式事务的实现方案

    在分布式系统中，分布式锁、分布式ID和分布式事务是常用的组件，用于解决并发控制、唯一标识和数据一致性的问题，本文将介绍Java中常用的分布式锁、分布式ID和分布式事务的实现方案，并通过具体的示例代码演示它们的用法和应用场景
    2023-06-06
  • 基于springboot搭建的web系统架构的方法步骤

    基于springboot搭建的web系统架构的方法步骤

    这篇文章主要介绍了基于springboot搭建的web系统架构的方法步骤，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
    2019-04-04
  • Java静态泛型使用方法实例解析

    Java静态泛型使用方法实例解析

    这篇文章主要介绍了Java静态泛型使用方法实例解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-06-06
  • IDEA使用入门小白操作教程

    IDEA使用入门小白操作教程

    IntelliJ IDEA，就是Java语言开发工具 （IDE） 功能与Eclipse同类！本文通过图文并茂的形式给大家介绍IntelliJ IDEA使用入门教程，非常适合新手小白操作，感兴趣的朋友一起看看吧
    2020-10-10
  • Jenkins自动化打包为war包

    Jenkins自动化打包为war包

    这篇文章主要介绍了Jenkins自动化打包为war包，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2021-01-01
  • 使用SpringMVC在redirect重定向的时候携带参数的问题

    使用SpringMVC在redirect重定向的时候携带参数的问题

    这篇文章主要介绍了使用SpringMVC在redirect重定向的时候携带参数的问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2022-03-03
  • java中 Set与Map排序输出到Writer详解及实例

    java中 Set与Map排序输出到Writer详解及实例

    这篇文章主要介绍了 java中 Set与Map排序输出到Writer详解及实例的相关资料,需要的朋友可以参考下
    2017-03-03

最新评论