srpingboot+activeMQ pub/sub持久化订阅+重发消息
1.srpingboot+activeMQ非持久化订阅
设置配置文件
spring.activemq.broker-url=tcp://xxxx:61616 spring.activemq.user=admin spring.activemq.password=admin spring.activemq.in-memory=true spring.jms.pub-sub-domain=true #如果此处设置为true,需要加如下的依赖包,否则会自动配置失败,报JmsMessagingTemplate注入失败 spring.activemq.pool.enabled=false server.port=8082
创建消费者
import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Service; /** * Created by shenlu on 2018/5/7. */ @Service public class Consumer { @JmsListener(destination = "testTopic") public void receiveMsg(String text) { System.out.println("<<<<<<============ 收到消息: " + text); } }
2.srpingboot+activeMQ pub/sub持久化订阅
配置文件的方式无法进行配置持久化订阅。所以需要自己去生成一个持久化订阅
import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.RedeliveryPolicy; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.jms.annotation.EnableJms; import org.springframework.jms.config.DefaultJmsListenerContainerFactory; import org.springframework.stereotype.Component; import javax.jms.ConnectionFactory; /** * Created by shenlu on 2018/5/24. */ @Component @EnableJms public class JmsConfiguration { @Value("${top.ActiveMQ_URL}") private String url; @Value("${top.ActiveMQ_USER}") private String user; @Value("${top.ActiveMQ_PASSWORD}") private String password; public RedeliveryPolicy redeliveryPolicy(){ RedeliveryPolicy redeliveryPolicy= new RedeliveryPolicy(); //是否在每次尝试重新发送失败后,增长这个等待时间 redeliveryPolicy.setUseExponentialBackOff(true); //重发次数,默认为6次,这里设置为10次,-1表示不限次数 redeliveryPolicy.setMaximumRedeliveries(-1); //重发时间间隔,默认为1毫秒,设置为10000毫秒 redeliveryPolicy.setInitialRedeliveryDelay(10000); //表示没有拖延只有UseExponentialBackOff(true)为true时生效 //第一次失败后重新发送之前等待10000毫秒,第二次失败再等待10000 * 2毫秒 //第三次翻倍10000 * 2 * 2,以此类推 redeliveryPolicy.setBackOffMultiplier(2); //是否避免消息碰撞 redeliveryPolicy.setUseCollisionAvoidance(true); //设置重发最大拖延时间360000毫秒 表示没有拖延只有UseExponentialBackOff(true)为true时生效 redeliveryPolicy.setMaximumRedeliveryDelay(360000); return redeliveryPolicy; } public ConnectionFactory connectionFactory(){ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(); connectionFactory.setBrokerURL(url); connectionFactory.setUserName(user); connectionFactory.setPassword(password); //设置重发属性 connectionFactory.setRedeliveryPolicy(redeliveryPolicy()); return connectionFactory; } /** * JMS 队列的监听容器工厂 */ @Bean(name = "jmsTopicListener") public DefaultJmsListenerContainerFactory jmsTopicListenerContainerFactory() { DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); factory.setConnectionFactory(connectionFactory()); factory.setPubSubDomain(true); factory.setSessionTransacted(true); factory.setAutoStartup(true); //开启持久化订阅 factory.setSubscriptionDurable(true); //重连间隔时间 factory.setRecoveryInterval(1000L); factory.setClientId("mysub"); return factory; } }
注意:factory.setSubscriptionDurable(true);factory.setClientId("mysub");这两个是开启持久化订阅的关键。
重发消息当次数设置不为-1时,重发达到设置的次数之后不会再重发。而设置为-1时,重启服务会继续重发。但是一般业务场景下不可能会一直去尝试重发一条消息。
配置文件
top.ActiveMQ_URL=tcp://xxx:61616 top.ActiveMQ_USER=admin top.ActiveMQ_PASSWORD=admin server.port=8081
创建消费者,及重发机制
import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Session; import javax.jms.TextMessage; /** * Created by shenlu on 2018/5/7. */ @Component public class Consumer { @JmsListener(destination = "testTopic", containerFactory = "jmsTopicListener") public void receive(TextMessage msg,Session session) throws JMSException { try { System.out.println("-- 接收到消息 -- " + msg.getText()); if (msg.getText().contains("2")) throw new JMSException(""); msg.acknowledge(); } catch (JMSException e) { System.out.println("-----测试重发-----"); session.rollback(); } } }
3.查看Subscribers
启动持久化订阅和非持久化订阅的两个工程,启动持久化订阅,在activeMQ工作台能看见Active Durable Topic Subscribers有一个用户
再启动非持久化订阅工程
关闭持久化订阅工程
持久化订阅,在订阅topic后,关闭工程,再启动,会继续接受生产者在关闭期间发送的消息。