基于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.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; @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("ctp.topic.queue"))); myContainer.setMessageListener(ctpListener); myContainer.setSubscriptionDurable(true); myContainer.setDurableSubscriptionName("ctp_name_2"); myContainer.setClientId("ctp_id_2"); myContainer.setSessionAcknowledgeModeName("CLIENT_ACKNOWLEDGE"); return myContainer; } }