SpringBoot整合ActiveMQ的详细步骤
gblfy 人气:01. 引入依赖
pom文件引入activemq依赖
<!--activeMq配置--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-activemq</artifactId> </dependency> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-pool</artifactId> <version>5.15.3</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>2.0.7</version> </dependency>
2. 配置文件
spring: activemq: user: admin password: admin broker-url: failover:(tcp://192.168.43.666:61616) #是否信任所有包(如果传递的是对象则需要设置为true,默认是传字符串) packages: trust-all: true #连接池 pool: enabled: true max-connections: 5 idle-timeout: 30000 # expiry-timeout: 0 jms: #默认使用queue模式,使用topic则需要设置为true pub-sub-domain: true # 是否信任所有包 #spring.activemq.packages.trust-all= # 要信任的特定包的逗号分隔列表(当不信任所有包时) #spring.activemq.packages.trusted= # 当连接请求和池满时是否阻塞。设置false会抛“JMSException异常”。 #spring.activemq.pool.block-if-full=true # 如果池仍然满,则在抛出异常前阻塞时间。 #spring.activemq.pool.block-if-full-timeout=-1ms # 是否在启动时创建连接。可以在启动时用于加热池。 #spring.activemq.pool.create-connection-on-startup=true # 是否用Pooledconnectionfactory代替普通的ConnectionFactory。 #spring.activemq.pool.enabled=false # 连接过期超时。 #spring.activemq.pool.expiry-timeout=0ms # 连接空闲超时 #spring.activemq.pool.idle-timeout=30s # 连接池最大连接数 #spring.activemq.pool.max-connections=1 # 每个连接的有效会话的最大数目。 #spring.activemq.pool.maximum-active-session-per-connection=500 # 当有"JMSException"时尝试重新连接 #spring.activemq.pool.reconnect-on-exception=true # 在空闲连接清除线程之间运行的时间。当为负数时,没有空闲连接驱逐线程运行。 #spring.activemq.pool.time-between-expiration-check=-1ms # 是否只使用一个MessageProducer #spring.activemq.pool.use-anonymous-producers=true
3. 生产者
package com.gblfy.producer; import org.apache.activemq.ScheduledMessage; import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ActiveMQTopic; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.jms.JmsProperties; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import javax.jms.*; import java.io.Serializable; /** * 发送消息 * * @author gblfy * @date 2022-11-02 */ @RestController @RequestMapping(value = "/active") public class SendController { //也可以注入JmsTemplate,JmsMessagingTemplate对JmsTemplate进行了封装 @Autowired private JmsMessagingTemplate jmsMessagingTemplate; /** * 发送消息接口 * 发送queue消息 :http://127.0.0.1:8080/active/send?msg=ceshi1234 * 发送topic 消息: http://127.0.0.1:8080/active/topic/send?msg=ceshi1234 * 发送queue消息(延迟time毫秒) :http://127.0.0.1:8080/active/send?msg=ceshi1234&time=5000 * * @param msg 消息 * @param type url中参数,非必须 * @param time * @return */ @RequestMapping({"/send", "/{type}/send"}) public String send(@PathVariable(value = "type", required = false) String type, String msg, Long time) { Destination destination = null; if (type == null) { type = ""; } switch (type) { case "topic": //发送广播消息 destination = new ActiveMQTopic("active.topic"); break; default: //发送 队列消息 destination = new ActiveMQQueue("active.queue"); break; } // System.out.println("开始请求发送:"+DateUtil.getStringDate(new Date(),"yyyy-MM-dd HH:mm:ss")); if (time != null && time > 0) { //延迟队列,延迟time毫秒 //延迟队列需要在 <broker>标签上增加属性 schedulerSupport="true" delaySend(destination, msg, time); } else { jmsMessagingTemplate.convertAndSend(destination, msg);//无序 //jmsMessagingTemplate.convertSendAndReceive();//有序 } return "activemq消息发送成功 队列消息:" + msg; } /** * 延时发送 * 说明:延迟队列需要在 <broker>标签上增加属性 schedulerSupport="true" * * @param destination 发送的队列 * @param data 发送的消息 * @param time 延迟时间 /毫秒 */ public <T extends Serializable> void delaySend(Destination destination, T data, Long time) { Connection connection = null; Session session = null; MessageProducer producer = null; // 获取连接工厂 ConnectionFactory connectionFactory = jmsMessagingTemplate.getConnectionFactory(); try { // 获取连接 connection = connectionFactory.createConnection(); connection.start(); // 获取session,true开启事务,false关闭事务 session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建一个消息队列 producer = session.createProducer(destination); producer.setDeliveryMode(JmsProperties.DeliveryMode.PERSISTENT.getValue()); ObjectMessage message = session.createObjectMessage(data); //设置延迟时间 message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time); // 发送消息 producer.send(message); session.commit(); } catch (Exception e) { e.printStackTrace(); } finally { try { if (producer != null) { producer.close(); } if (session != null) { session.close(); } if (connection != null) { connection.close(); } } catch (Exception e) { e.printStackTrace(); } } } }
4. 配置config
package com.gblfy.config; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.RedeliveryPolicy; import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ActiveMQTopic; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jms.annotation.EnableJms; import org.springframework.jms.config.DefaultJmsListenerContainerFactory; import org.springframework.jms.config.JmsListenerContainerFactory; import javax.jms.Queue; import javax.jms.Topic; /** * 描述: * activemq 有两种模式 queue 和 topic * queue 模式是单对单,有多个消费者的情况下则是使用轮询监听 * topic 模式/广播模式/发布订阅模式 是一对多,发送消息所有的消费者都能够监听到 * * @author gblfy * @date 2022-11-02 */ @EnableJms @Configuration public class ActiveMQConfig { //队列名 private static final String queueName = "active.queue"; //主题名 private static final String topicName = "active.topic"; @Value("${spring.activemq.user:}") private String username; @Value("${spring.activemq.password:}") private String password; @Value("${spring.activemq.broker-url:}") private String brokerUrl; @Bean public Queue acQueue() { return new ActiveMQQueue(queueName); } @Bean public Topic acTopic() { return new ActiveMQTopic(topicName); } @Bean public ActiveMQConnectionFactory connectionFactory() { return new ActiveMQConnectionFactory(username, password, brokerUrl); } @Bean public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory) { DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory(); // 关闭Session事务,手动确认与事务冲突 bean.setSessionTransacted(false); // 设置消息的签收模式(自己签收) /** * AUTO_ACKNOWLEDGE = 1 :自动确认 * CLIENT_ACKNOWLEDGE = 2:客户端手动确认 * DUPS_OK_ACKNOWLEDGE = 3: 自动批量确认 * SESSION_TRANSACTED = 0:事务提交并确认 * 但是在activemq补充了一个自定义的ACK模式: * INDIVIDUAL_ACKNOWLEDGE = 4:单条消息确认 **/ bean.setSessionAcknowledgeMode(4); //此处设置消息重发规则,redeliveryPolicy() 中定义 connectionFactory.setRedeliveryPolicy(redeliveryPolicy()); bean.setConnectionFactory(connectionFactory); return bean; } @Bean public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory) { DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory(); // 关闭Session事务,手动确认与事务冲突 bean.setSessionTransacted(false); bean.setSessionAcknowledgeMode(4); //设置为发布订阅方式, 默认情况下使用的生产消费者方式 bean.setPubSubDomain(true); bean.setConnectionFactory(connectionFactory); return bean; } /** * 消息的重发规则配置 */ @Bean public RedeliveryPolicy redeliveryPolicy() { RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy(); // 是否在每次尝试重新发送失败后,增长这个等待时间 redeliveryPolicy.setUseExponentialBackOff(true); // 重发次数五次, 总共六次 redeliveryPolicy.setMaximumRedeliveries(5); // 重发时间间隔,默认为1000ms(1秒) redeliveryPolicy.setInitialRedeliveryDelay(1000); // 重发时长递增的时间倍数2 redeliveryPolicy.setBackOffMultiplier(2); // 是否避免消息碰撞 redeliveryPolicy.setUseCollisionAvoidance(false); // 设置重发最大拖延时间-1表示无延迟限制 redeliveryPolicy.setMaximumRedeliveryDelay(-1); return redeliveryPolicy; } }
5. queue消费者
package com.gblfy.listener; import org.apache.activemq.command.ActiveMQMessage; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Session; /** * TODO * * @author gblfy * @Date 2022-11-02 **/ @Component public class QueueListener { /** * queue 模式 单对单,两个消费者监听同一个队列则通过轮询接收消息 * containerFactory属性的值关联config类中的声明 * * @param msg */ @JmsListener(destination = "active.queue", containerFactory = "jmsListenerContainerQueue") public void queueListener(ActiveMQMessage message, Session session, String msg) throws JMSException { try { System.out.println("active queue 接收到消息 " + msg); //手动签收 message.acknowledge(); } catch (Exception e) { //重新发送 session.recover(); } } }
6. topic消费者
package com.gblfy.listener; import org.apache.activemq.command.ActiveMQMessage; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Session; /** * TODO * * @author gblfy * @Date 2022-11-02 **/ @Component public class TopicListener { /** * topic 模式/广播模式/发布订阅模式 一对多,多个消费者可同时接收到消息 * topic 模式无死信队列,死信队列是queue模式 * containerFactory属性的值关联config类中的声明 * * @param msg */ @JmsListener(destination = "active.topic", containerFactory = "jmsListenerContainerTopic") public void topicListener(ActiveMQMessage message, Session session, String msg) throws JMSException { try { // System.out.println("接收到消息:" + DateUtil.getStringDate(new Date(), "yyyy-MM-dd HH:mm:ss")); System.out.println("active topic 接收到消息 " + msg); System.out.println(""); //手动签收 message.acknowledge(); } catch (Exception e) { //重新发送 session.recover(); } } @JmsListener(destination = "active.topic", containerFactory = "jmsListenerContainerTopic") public void topicListener2(ActiveMQMessage message, Session session, String msg) throws JMSException { try { // System.out.println("接收到消息:" + DateUtil.getStringDate(new Date(), "yyyy-MM-dd HH:mm:ss")); System.out.println("active topic2 接收到消息 " + msg); System.out.println(""); //手动签收 message.acknowledge(); } catch (Exception e) { //重新发送 session.recover(); } } }
6. ActiveMQ 消息存储规则
QUEUE 点对点:
特点:消息遵循先到先得,消息只能被一个消费者消费。
消息存储规则:消费者消费消息成功,MQ服务端消息删除
TOPIC订阅模式: 消息属于广播(订阅)模式,消息会被所有的topic消费者消费消息。
消息存储规则:所有消费者消费成功,MQ服务端消息删除,有一个消息没有没有消费完成,消息也会存储在MQ服务端。
举例:
已经处于运行topic消费者5个,5个消费者消费完成后,MQ服务端消息删除。
扩展点补充:如果想额外添加topic消费者,如果MQ服务端消息没有被消费完毕,新增topic消费者可以消费以前未被消费的消息,
正常新增的只会消费新的topic消息。
总结
加载全部内容