《SpringBoot2.0 实战》系列-整合Activemq实现点对点、发布订阅模式共存、消息持久化、重试等

电玩女神 2023-05-28 12:43 8阅读 0赞

## 简介 ##

**Apache ActiveMQ**是[Apache软件基金会][Apache]所研发的开放源代码[消息中间件][Link 1]；由于ActiveMQ是一个纯[Java][]程序，因此只需要[操作系统][Link 2]支持[Java虚拟机][Java 1]，ActiveMQ便可执行。

## 作用 ##

*  应用解耦
 *  异步通信
 *  流量削峰
 *  消息通讯

## 安装  ##

提供docker的安装方式，其他方式网上也都有很多

> [https://blog.csdn.net/HXNLYW/article/details/88950291][https_blog.csdn.net_HXNLYW_article_details_88950291]

## ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hYTkxZVw_size_16_color_FFFFFF_t_70][] ##

## 与springboot整合  ##

### jar依赖 ###

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-pool</artifactId>
    </dependency>

### 配置文件 ###

spring: 
      # activemq
      activemq:
        broker-url: tcp://xxx:61616
        user: admin
        password: admin
        pool:
          enabled: true
          max-connections: 100
        packages:
          #注意 对象传输需开启包白名单 否则会报错- -
          trust-all: true 
        in-memory: true
        # 自定义queue名称
        queue-name: gourd-queue
        # 自定义topic名称
        topic-name: gourd-topic
      jms:
        # 开启发布订阅模式
        pub-sub-domain: true

### 配置类 ###

配置类加上@EnableJms注解，注册重试、持久化配置bean

/**
     * active消息队列配置
     * @author gour.hu
     */
    @Configuration
    @EnableJms
    public class ActiveMqConfig {
    
        /**
         * 定义存放消息的队列
         * @return
         */
        @Bean
        public Queue queue(@Value("${spring.activemq.queue-name}")String queueName) {
            return new ActiveMQQueue(queueName);
        }
    
        /**
         * 发布订阅模式
         * @return
         */
        @Bean
        public Topic topic(@Value("${spring.activemq.topic-name}")String topicName) {
            return new ActiveMQTopic(topicName);
        }
    
        @Bean
        public RedeliveryPolicy redeliveryPolicy(){
            RedeliveryPolicy  redeliveryPolicy=   new RedeliveryPolicy();
            // 启用指数倍数递增的方式增加延迟时间。
            redeliveryPolicy.setUseExponentialBackOff(true);
            // 重连时间间隔递增倍数，只有值大于1和启用useExponentialBackOff参数时才生效。默认5
            redeliveryPolicy.setBackOffMultiplier(2);
            // 最大重传次数，达到最大重连次数后抛出异常。为-1时不限制次数，为0时表示不进行重传。默认6
            redeliveryPolicy.setMaximumRedeliveries(6);
            // 重发时间间隔,默认为1秒
            redeliveryPolicy.setInitialRedeliveryDelay(5000L);
            // 启用防止冲突功能，因为消息接收时是可以使用多线程并发处理的，应该是为了重发的安全性，避开所有并发线程都在同一个时间点进行消息接收处理。
            redeliveryPolicy.setUseCollisionAvoidance(true);
            // 设置重发最大拖延时间-1 表示没有拖延只有UseExponentialBackOff(true)为true时生效
            redeliveryPolicy.setMaximumRedeliveryDelay(-1);
    
            return redeliveryPolicy;
        }
    
    
        @Bean
        public ActiveMQConnectionFactory activeMQConnectionFactory (@Value("${spring.activemq.broker-url}")String url,
              @Value("${spring.activemq.user}")String user, @Value("${spring.activemq.password}")String password,
                                                                    RedeliveryPolicy redeliveryPolicy){
            ActiveMQConnectionFactory activeMQConnectionFactory =
                    new ActiveMQConnectionFactory(user,password,url);
            activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
            return activeMQConnectionFactory;
        }
    
        @Bean
        public JmsTemplate getJmsTemplate(ActiveMQConnectionFactory activeMQConnectionFactory,
            MessageConverter jacksonJmsMessageConverter,@Value("${spring.jms.pub-sub-domain}")Boolean subAdmin){
            //使用CachingConnectionFactory可以提高部分性能。
            CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
            cachingConnectionFactory.setSessionCacheSize(100);
            cachingConnectionFactory.setTargetConnectionFactory(activeMQConnectionFactory);
            JmsTemplate jmsTemplate = new JmsTemplate(cachingConnectionFactory);
            //设置deliveryMode（持久化）
            jmsTemplate.setExplicitQosEnabled(true);
            // 设置消息是否持久化
            jmsTemplate.setDeliveryPersistent(true);
            // 设置消息转换器
            jmsTemplate.setMessageConverter(jacksonJmsMessageConverter);
            // 设置消息是否以事务
            jmsTemplate.setSessionTransacted(true);
            jmsTemplate.setPubSubDomain(subAdmin);
            return jmsTemplate;
        }
    
        /**
         * topic模式的ListenerContainer
         *
         * @param activeMQConnectionFactory
         * @return
         */
        @Bean("jmsListenerContainerTopic")
        public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory activeMQConnectionFactory,
               DefaultJmsListenerContainerFactoryConfigurer configurer,@Value("${spring.application.name}")String applicationName) {
            DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
            configurer.configure(factory, activeMQConnectionFactory);
            // 设置消息转换器
            factory.setMessageConverter(jacksonJmsMessageConverter());
            factory.setPubSubDomain(true);
            factory.setSessionTransacted(true);
            factory.setAutoStartup(true);
            //开启持久化订阅
            factory.setSubscriptionDurable(true);
            factory.setClientId(applicationName);
            return factory;
        }
    
        /**
         * queue模式的ListenerContainer
         * @param activeMQConnectionFactory
         * @return
         */
        @Bean("jmsListenerContainerQueue")
        public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory activeMQConnectionFactory,
                                                                        DefaultJmsListenerContainerFactoryConfigurer configurer) {
            DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
            configurer.configure(factory, activeMQConnectionFactory);
            // 设置消息转换器
            factory.setMessageConverter(jacksonJmsMessageConverter());
            factory.setPubSubDomain(false);
            return factory;
        }
    
        /**
         * 消息转换器
         * @return
         */
        @Bean
        public MessageConverter jacksonJmsMessageConverter() {
            MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
            converter.setTargetType(MessageType.TEXT);
            converter.setTypeIdPropertyName("_type");
            return converter;
        }
    
    }

### 提供者 ###

/**
     * 队列消息控制器
     * @author gourd.hu
     */
    @RestController
    @RequestMapping("/activeMq")
    @Api(tags = "activeMq", description = "队列消息控制器" )
    public class ActiveMqController {
    
        /**
         * 注入存放消息的队列，用于下列方法一
         */
        @Autowired
        private Queue queue;
    
        /**
         * 注入springboot封装的工具类
         */
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        @GetMapping("send")
        @ApiOperation(value = "发送消息到队列")
        public BaseResponse send(@RequestParam("msg") String msg) {
    
            jmsMessagingTemplate.convertAndSend(queue, msg);
            return BaseResponse.ok("success!");
        }
    
    
        @Autowired
        private Topic topic;
     
        @GetMapping("/topic")
        @ApiOperation(value = "发送消息到主题")
        public BaseResponse handlerActiveMq(@RequestParam("msg")String  msg) {
            jmsMessagingTemplate.convertAndSend(topic, msg);
            return BaseResponse.ok("success!");
        }
     
    }

### 消费者 ###

>     @SendTo 加上此注解消费方法可以有返回值，否则只能为void

/**
     * 消息消费
     *
     * @author gourdhu
     */
    @Component
    @Slf4j
    public class ConsumerService {
    
    
        /**
         * 使用JmsListener配置消费者监听的队列，其中name是接收到的消息
         * @param message
         * @return
         */
        @JmsListener(destination = "gourd-queue",containerFactory = "jmsListenerContainerQueue")
        @SendTo("SQueue")
        public void handleMessage(final TextMessage message) throws JMSException {
            log.info("queue-consumer收到的报文为:" + message.getText());
        }
    
        @JmsListener(destination = "gourd-topic",containerFactory = "jmsListenerContainerTopic")
        @SendTo("STopic")// 加上此注解可以有返回值，否则只能为void
        public void receiveTopic(final TextMessage message) throws JMSException {
            log.info("topic-consumer收到的报文为:" + message.getText());
    //        if(true){
    //            throw new BusinessException("测试消息重试");
    //        }
        }
     
    }

至此整合完成，经测试持久化和重试都是正常的。

===============================================

代码均已上传至本人的开源项目

![20191230131831499.jpeg][]

spring-cloud-plus：[https://blog.csdn.net/HXNLYW/article/details/104635673][https_blog.csdn.net_HXNLYW_article_details_104635673]

[Apache]: https://baike.baidu.com/item/Apache%E8%BD%AF%E4%BB%B6%E5%9F%BA%E9%87%91%E4%BC%9A
[Link 1]: https://baike.baidu.com/item/%E6%B6%88%E6%81%AF%E4%B8%AD%E9%97%B4%E4%BB%B6
[Java]: https://baike.baidu.com/item/Java
[Link 2]: https://baike.baidu.com/item/%E6%93%8D%E4%BD%9C%E7%B3%BB%E7%BB%9F
[Java 1]: https://baike.baidu.com/item/Java%E8%99%9A%E6%8B%9F%E6%9C%BA
[https_blog.csdn.net_HXNLYW_article_details_88950291]: https://blog.csdn.net/HXNLYW/article/details/88950291
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hYTkxZVw_size_16_color_FFFFFF_t_70]: /images/20230528/1b6406c70c63431d881d153c126af023.png
[20191230131831499.jpeg]: /images/20230528/40e509213e3d4cf0b4dfe99352ffeb76.png
[https_blog.csdn.net_HXNLYW_article_details_104635673]: https://blog.csdn.net/HXNLYW/article/details/104635673