中国红客联盟国际站联盟介绍联盟架构联盟章程发展历程联盟邮箱积分VIP捐助联盟收藏本站
请登录
立即注册
门户首页
交流社区BBS
技术圈子Group
关于我们
红客大事记
插件Plugin
搜索
当前位置:»交流社区 › 中国红客联盟【技术分享区】 › 编程脚本交流 › SpringBoot整合ActiveMQ的详细步骤

[JAVA] SpringBoot整合ActiveMQ的详细步骤

1846 0
黑夜隐士 2022-11-6 11:23:47 | 显示全部楼层 |阅读模式
目录

    1. 引入依赖2. 配置文件3. 生产者4. 配置config5. queue消费者6. topic消费者6. ActiveMQ 消息存储规则总结


1. 引入依赖

pom文件引入activemq依赖
  1.     <!--activeMq配置-->
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter-activemq</artifactId>
  5.         </dependency>
  6.         <dependency>
  7.             <groupId>org.apache.activemq</groupId>
  8.             <artifactId>activemq-pool</artifactId>
  9.             <version>5.15.3</version>
  10.         </dependency>
  12.         <dependency>
  13.             <groupId>org.projectlombok</groupId>
  14.             <artifactId>lombok</artifactId>
  15.         </dependency>
  16.         <dependency>
  17.             <groupId>org.springframework.boot</groupId>
  18.             <artifactId>spring-boot-starter-web</artifactId>
  19.         </dependency>
  20.         <dependency>
  21.             <groupId>com.alibaba</groupId>
  22.             <artifactId>fastjson</artifactId>
  23.             <version>2.0.7</version>
  24.         </dependency>
复制代码
2. 配置文件
  1. spring:
  2.   activemq:
  3.     user: admin
  4.     password: admin
  5.     broker-url: failover:(tcp://192.168.43.666:61616)
  6.     #是否信任所有包(如果传递的是对象则需要设置为true，默认是传字符串)
  7.     packages:
  8.       trust-all: true
  9.     #连接池
  10.     pool:
  11.       enabled: true
  12.       max-connections: 5
  13.       idle-timeout: 30000
  14. #      expiry-timeout: 0
  15.     jms:
  16.       #默认使用queue模式，使用topic则需要设置为true
  17.       pub-sub-domain: true
  19.       # 是否信任所有包
  20.       #spring.activemq.packages.trust-all=
  21.       # 要信任的特定包的逗号分隔列表（当不信任所有包时）
  22.       #spring.activemq.packages.trusted=
  23.       # 当连接请求和池满时是否阻塞。设置false会抛“JMSException异常”。
  24.       #spring.activemq.pool.block-if-full=true
  25.       # 如果池依旧满，则在抛出异常前阻塞时间。
  26.       #spring.activemq.pool.block-if-full-timeout=-1ms
  27.       # 是否在启动时创建连接。可以在启动时用于加热池。
  28.       #spring.activemq.pool.create-connection-on-startup=true
  29.       # 是否用Pooledconnectionfactory代替普通的ConnectionFactory。
  30.       #spring.activemq.pool.enabled=false
  31.       # 连接过期超时。
  32.       #spring.activemq.pool.expiry-timeout=0ms
  33.       # 连接空闲超时
  34.       #spring.activemq.pool.idle-timeout=30s
  35.       # 连接池最大连接数
  36.       #spring.activemq.pool.max-connections=1
  37.       # 每个连接的有效会话的最大数目。
  38.       #spring.activemq.pool.maximum-active-session-per-connection=500
  39.       # 当有"JMSException"时尝试重新连接
  40.       #spring.activemq.pool.reconnect-on-exception=true
  41.       # 在空闲连接清除线程之间运行的时间。当为负数时，没有空闲连接驱逐线程运行。
  42.       #spring.activemq.pool.time-between-expiration-check=-1ms
  43.       # 是否只使用一个MessageProducer
  44.       #spring.activemq.pool.use-anonymous-producers=true
复制代码
3. 生产者
  1. package com.gblfy.producer;
  3. import org.apache.activemq.ScheduledMessage;
  4. import org.apache.activemq.command.ActiveMQQueue;
  5. import org.apache.activemq.command.ActiveMQTopic;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.boot.autoconfigure.jms.JmsProperties;
  8. import org.springframework.jms.core.JmsMessagingTemplate;
  9. import org.springframework.web.bind.annotation.PathVariable;
  10. import org.springframework.web.bind.annotation.RequestMapping;
  11. import org.springframework.web.bind.annotation.RestController;
  13. import javax.jms.*;
  14. import java.io.Serializable;
  16. /**
  17. * 发送消息
  18. *
  19. * @author gblfy
  20. * @date 2022-11-02
  21. */
  22. @RestController
  23. @RequestMapping(value = "/active")
  24. public class SendController {
  25.     //也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装
  26.     @Autowired
  27.     private JmsMessagingTemplate jmsMessagingTemplate;
  29.     /**
  30.      * 发送消息接口
  31.      * 发送queue消息 ：http://127.0.0.1:8080/active/send?msg=ceshi1234
  32.      * 发送topic 消息： http://127.0.0.1:8080/active/topic/send?msg=ceshi1234
  33.      * 发送queue消息(延迟time毫秒) ：http://127.0.0.1:8080/active/send?msg=ceshi1234&time=5000
  34.      *
  35.      * @param msg  消息
  36.      * @param type url中参数,非必须
  37.      * @param time
  38.      * @return
  39.      */
  40.     @RequestMapping({"/send", "/{type}/send"})
  41.     public String send(@PathVariable(value = "type", required = false) String type, String msg, Long time) {
  42.         Destination destination = null;
  43.         if (type == null) {
  44.             type = "";
  45.         }
  46.         switch (type) {
  47.             case "topic":
  48.                 //发送广播消息
  49.                 destination = new ActiveMQTopic("active.topic");
  50.                 break;
  51.             default:
  52.                 //发送 队列消息
  53.                 destination = new ActiveMQQueue("active.queue");
  54.                 break;
  55.         }
  56.         // System.out.println("开始请求发送："+DateUtil.getStringDate(new Date(),"yyyy-MM-dd HH:mm:ss"));
  57.         if (time != null && time > 0) {
  58.             //延迟队列，延迟time毫秒
  59.             //延迟队列需要在 <broker>标签上增加属性 schedulerSupport="true"
  60.             delaySend(destination, msg, time);
  61.         } else {
  62.             jmsMessagingTemplate.convertAndSend(destination, msg);//无序
  63.             //jmsMessagingTemplate.convertSendAndReceive();//有序
  64.         }
  65.         return "activemq消息发送成功 队列消息：" + msg;
  66.     }
  68.     /**
  69.      * 延时发送
  70.      * 说明：延迟队列需要在 <broker>标签上增加属性 schedulerSupport="true"
  71.      *
  72.      * @param destination 发送的队列
  73.      * @param data        发送的消息
  74.      * @param time        延迟时间 /毫秒
  75.      */
  76.     public <T extends Serializable> void delaySend(Destination destination, T data, Long time) {
  77.         Connection connection = null;
  78.         Session session = null;
  79.         MessageProducer producer = null;
  80.         // 获取连接工厂
  81.         ConnectionFactory connectionFactory = jmsMessagingTemplate.getConnectionFactory();
  82.         try {
  83.             // 获取连接
  84.             connection = connectionFactory.createConnection();
  85.             connection.start();
  86.             // 获取session，true开启事务，false关闭事务
  87.             session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
  88.             // 创建一个消息队列
  89.             producer = session.createProducer(destination);
  90.             producer.setDeliveryMode(JmsProperties.DeliveryMode.PERSISTENT.getValue());
  91.             ObjectMessage message = session.createObjectMessage(data);
  92.             //设置延迟时间
  93.             message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
  94.             // 发送消息
  95.             producer.send(message);
  96.             session.commit();
  97.         } catch (Exception e) {
  98.             e.printStackTrace();
  99.         } finally {
  100.             try {
  101.                 if (producer != null) {
  102.                     producer.close();
  103.                 }
  104.                 if (session != null) {
  105.                     session.close();
  106.                 }
  107.                 if (connection != null) {
  108.                     connection.close();
  109.                 }
  110.             } catch (Exception e) {
  111.                 e.printStackTrace();
  112.             }
  113.         }
  114.     }
  115. }
复制代码
4. 配置config
  1. package com.gblfy.config;
  3. import org.apache.activemq.ActiveMQConnectionFactory;
  4. import org.apache.activemq.RedeliveryPolicy;
  5. import org.apache.activemq.command.ActiveMQQueue;
  6. import org.apache.activemq.command.ActiveMQTopic;
  7. import org.springframework.beans.factory.annotation.Value;
  8. import org.springframework.context.annotation.Bean;
  9. import org.springframework.context.annotation.Configuration;
  10. import org.springframework.jms.annotation.EnableJms;
  11. import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
  12. import org.springframework.jms.config.JmsListenerContainerFactory;
  14. import javax.jms.Queue;
  15. import javax.jms.Topic;
  17. /**
  18. * 描述:
  19. * activemq 有两种模式 queue 和 topic
  20. * queue 模式是单对单，有多个消费者的情况下则是使用轮询监听
  21. * topic 模式/广播模式/发布订阅模式 是一对多，发送消息所有的消费者都能够监听到
  22. *
  23. * @author gblfy
  24. * @date 2022-11-02
  25. */
  26. @EnableJms
  27. @Configuration
  28. public class ActiveMQConfig {
  29.     //队列名
  30.     private static final String queueName = "active.queue";
  31.     //主题名
  32.     private static final String topicName = "active.topic";
  34.     @Value("${spring.activemq.user:}")
  35.     private String username;
  36.     @Value("${spring.activemq.password:}")
  37.     private String password;
  38.     @Value("${spring.activemq.broker-url:}")
  39.     private String brokerUrl;
  41.     @Bean
  42.     public Queue acQueue() {
  43.         return new ActiveMQQueue(queueName);
  44.     }
  46.     @Bean
  47.     public Topic acTopic() {
  48.         return new ActiveMQTopic(topicName);
  49.     }
  51.     @Bean
  52.     public ActiveMQConnectionFactory connectionFactory() {
  53.         return new ActiveMQConnectionFactory(username, password, brokerUrl);
  54.     }
  56.     @Bean
  57.     public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory) {
  58.         DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
  59.         // 关闭Session事务，手动确认与事务冲突
  60.         bean.setSessionTransacted(false);
  61.         // 设置消息的签收模式（自己签收）
  62.         /**
  63.          * AUTO_ACKNOWLEDGE = 1 ：自动确认
  64.          * CLIENT_ACKNOWLEDGE = 2：客户端手动确认
  65.          * DUPS_OK_ACKNOWLEDGE = 3： 自动批量确认
  66.          * SESSION_TRANSACTED = 0：事务提交并确认
  67.          * 但是在activemq补充了一个自定义的ACK模式:
  68.          * INDIVIDUAL_ACKNOWLEDGE = 4：单条消息确认
  69.          **/
  70.         bean.setSessionAcknowledgeMode(4);
  71.         //此处设置消息重发规则，redeliveryPolicy() 中定义
  72.         connectionFactory.setRedeliveryPolicy(redeliveryPolicy());
  73.         bean.setConnectionFactory(connectionFactory);
  74.         return bean;
  75.     }
  77.     @Bean
  78.     public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory) {
  79.         DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
  80.         // 关闭Session事务，手动确认与事务冲突
  81.         bean.setSessionTransacted(false);
  82.         bean.setSessionAcknowledgeMode(4);
  83.         //设置为发布订阅方式, 默认情况下使用的生产消费者方式
  84.         bean.setPubSubDomain(true);
  85.         bean.setConnectionFactory(connectionFactory);
  86.         return bean;
  87.     }
  89.     /**
  90.      * 消息的重发规则配置
  91.      */
  92.     @Bean
  93.     public RedeliveryPolicy redeliveryPolicy() {
  94.         RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
  95.         // 是否在每次尝试重新发送失败后,增长这个等待时间
  96.         redeliveryPolicy.setUseExponentialBackOff(true);
  97.         // 重发次数五次， 总共六次
  98.         redeliveryPolicy.setMaximumRedeliveries(5);
  99.         // 重发时间间隔,默认为1000ms（1秒）
  100.         redeliveryPolicy.setInitialRedeliveryDelay(1000);
  101.         // 重发时长递增的时间倍数2
  102.         redeliveryPolicy.setBackOffMultiplier(2);
  103.         // 是否避免消息碰撞
  104.         redeliveryPolicy.setUseCollisionAvoidance(false);
  105.         // 设置重发最大拖延时间-1表示无延迟限制
  106.         redeliveryPolicy.setMaximumRedeliveryDelay(-1);
  107.         return redeliveryPolicy;
  108.     }
  109. }
复制代码
5. queue消费者
  1. package com.gblfy.listener;
  3. import org.apache.activemq.command.ActiveMQMessage;
  4. import org.springframework.jms.annotation.JmsListener;
  5. import org.springframework.stereotype.Component;
  7. import javax.jms.JMSException;
  8. import javax.jms.Session;
  10. /**
  11. * TODO
  12. *
  13. * @author gblfy
  14. * @Date 2022-11-02
  15. **/
  16. @Component
  17. public class QueueListener {
  19.     /**
  20.      * queue 模式 单对单，两个消费者监听同一个队列则通过轮询接收消息
  21.      * containerFactory属性的值关联config类中的声明
  22.      *
  23.      * @param msg
  24.      */
  25.     @JmsListener(destination = "active.queue", containerFactory = "jmsListenerContainerQueue")
  26.     public void queueListener(ActiveMQMessage message, Session session, String msg) throws JMSException {
  27.         try {
  28.             System.out.println("active queue 接收到消息 " + msg);
  29.             //手动签收
  30.             message.acknowledge();
  31.         } catch (Exception e) {
  32.             //重新发送
  33.             session.recover();
  34.         }
  35.     }
  36. }
复制代码
6. topic消费者
  1. package com.gblfy.listener;
  3. import org.apache.activemq.command.ActiveMQMessage;
  4. import org.springframework.jms.annotation.JmsListener;
  5. import org.springframework.stereotype.Component;
  7. import javax.jms.JMSException;
  8. import javax.jms.Session;
  10. /**
  11. * TODO
  12. *
  13. * @author gblfy
  14. * @Date 2022-11-02
  15. **/
  16. @Component
  17. public class TopicListener {
  19.     /**
  20.      * topic 模式/广播模式/发布订阅模式 一对多，多个消费者可同时接收到消息
  21.      * topic 模式无死信队列，死信队列是queue模式
  22.      * containerFactory属性的值关联config类中的声明
  23.      *
  24.      * @param msg
  25.      */
  26.     @JmsListener(destination = "active.topic", containerFactory = "jmsListenerContainerTopic")
  27.     public void topicListener(ActiveMQMessage message, Session session, String msg) throws JMSException {
  28.         try {
  29.             // System.out.println("接收到消息：" + DateUtil.getStringDate(new Date(), "yyyy-MM-dd HH:mm:ss"));
  30.             System.out.println("active topic 接收到消息 " + msg);
  31.             System.out.println("");
  32.             //手动签收
  33.             message.acknowledge();
  34.         } catch (Exception e) {
  35.             //重新发送
  36.             session.recover();
  37.         }
  38.     }
  40.     @JmsListener(destination = "active.topic", containerFactory = "jmsListenerContainerTopic")
  41.     public void topicListener2(ActiveMQMessage message, Session session, String msg) throws JMSException {
  42.         try {
  43.             // System.out.println("接收到消息：" + DateUtil.getStringDate(new Date(), "yyyy-MM-dd HH:mm:ss"));
  44.             System.out.println("active topic2 接收到消息 " + msg);
  45.             System.out.println("");
  46.             //手动签收
  47.             message.acknowledge();
  48.         } catch (Exception e) {
  49.             //重新发送
  50.             session.recover();
  51.         }
  52.     }
  53. }
复制代码
6. ActiveMQ 消息存储规则

QUEUE 点对点：
特点：消息遵循先到先得，消息只能被一个消费者消费。
消息存储规则：消费者消费消息成功，MQ服务端消息删除
TOPIC订阅模式: 消息属于广播（订阅）模式，消息会被所有的topic消费者消费消息。
消息存储规则：所有消费者消费成功，MQ服务端消息删除，有一个消息没有没有消费完成，消息也会存储在MQ服务端。
举例：
已经处于运行topic消费者5个，5个消费者消费完成后，MQ服务端消息删除。
扩展点补充：如果想额外添加topic消费者，如果MQ服务端消息没有被消费完毕，新增topic消费者可以消费以前未被消费的消息，
正常新增的只会消费新的topic消息。

总结

到此这篇关于SpringBoot整合ActiveMQ的文章就介绍到这了,更多相关SpringBoot整合ActiveMQ内容请搜索中国红客联盟以前的文章或继续浏览下面的相关文章希望大家以后多多支持中国红客联盟！
回复

使用道具 举报

返回列表 发新帖
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

黑夜隐士

创始人

关注
  • 1059
    主题
  • 329
    粉丝
  • 354
    关注
这家伙很懒，什么都没留下！

中国红客联盟公众号

联系站长QQ：5520533

admin@chnhonker.com
Copyright © 2001-2025 Discuz Team. Powered by Discuz! X3.5 ( 粤ICP备13060014号 )|天天打卡 本站已运行天