基于SpringBoot开发的ActiveMQ虚拟主题客户端（包含发布者和订阅者）

/***************************************************************** * Copyright (c) 2017 www.noryar.com Inc. All rights reserved. *****************************************************************/ package subscriber; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ActiveMQTopic; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.core.env.Environment; import org.springframework.jms.annotation.JmsListener; import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.TextMessage; import java.util.concurrent.atomic.AtomicInteger; /** * 类描述:订阅者. * * @author leon. */ @Component public class Subscriber { private static final Logger LOGGER = LoggerFactory.getLogger(Subscriber.class); private AtomicInteger cnt = new AtomicInteger(); // #############################注解启动方式############################# // @JmsListener(destination = "test", containerFactory = "myFactory") public String receive(Message message) throws JMSException { String msg = ((TextMessage) message).getText(); LOGGER.info("get message: {}", msg); cnt.addAndGet(1); return msg; } public int getDealCnt() { return cnt.get(); } // #############################Container############################# @Autowired private Environment env; @Autowired private CtpListener ctpListener; @Autowired private VirtualTopicAListener virtualTopicAListener; @Autowired private VirtualTopicBListener virtualTopicBListener; @Bean public ActiveMQConnectionFactory myFactory() { LOGGER.info(env.getProperty("spring.activemq.broker-url")); ActiveMQConnectionFactory myFactory = new ActiveMQConnectionFactory(); myFactory.setBrokerURL(env.getProperty("spring.activemq.broker-url")); myFactory.setAlwaysSyncSend(false); myFactory.setUserName(env.getProperty("username")); myFactory.setPassword(env.getProperty("password")); return myFactory; } @Bean public DefaultMessageListenerContainer myContainer() { DefaultMessageListenerContainer myContainer = new DefaultMessageListenerContainer(); myContainer.setConnectionFactory(myFactory()); myContainer.setDestination(new ActiveMQQueue(env.getProperty("queue"))); myContainer.setMessageListener(ctpListener); myContainer.setClientId("ctp_id_2"); myContainer.setSessionAcknowledgeModeName("CLIENT_ACKNOWLEDGE"); return myContainer; } @Bean public DefaultMessageListenerContainer myVirtualATopicContainer() { DefaultMessageListenerContainer myVirtualATopicContainer = new DefaultMessageListenerContainer(); myVirtualATopicContainer.setConnectionFactory(myFactory()); myVirtualATopicContainer.setDestination(new ActiveMQTopic(env.getProperty("topic.queue"))); myVirtualATopicContainer.setMessageListener(virtualTopicAListener); myVirtualATopicContainer.setSubscriptionDurable(true); myVirtualATopicContainer.setDurableSubscriptionName("client_A"); myVirtualATopicContainer.setClientId("client_A_name"); myVirtualATopicContainer.setSessionAcknowledgeModeName("CLIENT_ACKNOWLEDGE"); return myVirtualATopicContainer; } @Bean public DefaultMessageListenerContainer myVirtualBTopicContainer() { DefaultMessageListenerContainer myVirtualBTopicContainer = new DefaultMessageListenerContainer(); myVirtualBTopicContainer.setConnectionFactory(myFactory()); myVirtualBTopicContainer.setDestination(new ActiveMQTopic(env.getProperty("topic.queue"))); myVirtualBTopicContainer.setMessageListener(virtualTopicAListener); myVirtualBTopicContainer.setSubscriptionDurable(true); myVirtualBTopicContainer.setDurableSubscriptionName("client_B"); myVirtualBTopicContainer.setClientId("client_B_name"); myVirtualBTopicContainer.setSessionAcknowledgeModeName("CLIENT_ACKNOWLEDGE"); return myVirtualBTopicContainer; } }