springboot 使用 ActiveMQ

pom文件添加依赖

<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.17.1</version></dependency>

配置 application.yml

server:   port:8080#springBoot项目访问端口   spring:   activemq:     broker-url: tcp://192.168.140.xx:30005#你activeMQ的ip和端口号     user: admin#activeMq账号     password: admin#activeMq密码     pool:       enabled:true#连接池启动       max-connections:10#最大连接数

SpringBoot 的启动类添加解 @EnableJms

@SpringBootApplication@EnableJmspublicclassMqconverterApplication{publicstaticvoidmain(String[] args){SpringApplication.run(MqconverterApplication.class, args);}}

创建 activeMQ 配置类

mportorg.apache.activemq.ActiveMQConnectionFactory;importorg.apache.activemq.command.ActiveMQQueue;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.jms.config.JmsListenerContainerFactory;importorg.springframework.jms.config.SimpleJmsListenerContainerFactory;importjavax.jms.ConnectionFactory;importjavax.jms.Queue;/**  * 配置类  */@ConfigurationpublicclassActiveMQConfig{@Value("${spring.activemq.broker-url}")privateString brokerUrl;@Value("${spring.activemq.user}")privateString userName;@Value("${spring.activemq.password}")privateString password;@BeanpublicConnectionFactoryconnectionFactory(){returnnewActiveMQConnectionFactory(userName, password, brokerUrl);}/**      * 在 Queue 模式中，对消息的监听需要对containerFactory进行配置      */@Bean("queueListener")publicJmsListenerContainerFactory<?>queueJmsListenerContainerFactory(ConnectionFactory connectionFactory){SimpleJmsListenerContainerFactory factory=newSimpleJmsListenerContainerFactory();         factory.setConnectionFactory(connectionFactory);         factory.setPubSubDomain(false);return factory;}/**      * 在 topic 模式中，对消息的监听需要对containerFactory进行配置      */@Bean("topicListener")publicJmsListenerContainerFactory<?>topicJmsListenerContainerFactory(ConnectionFactory connectionFactory){SimpleJmsListenerContainerFactory factory=newSimpleJmsListenerContainerFactory();         factory.setConnectionFactory(connectionFactory);         factory.setPubSubDomain(true);return factory;}}

pubSubDomain：为true 代表 topic 模式，false 代表队列模式，默认是false

生产者

@RestControllerpublicclassActiveMQProducerController{/**      * 队列模式      * @param msg 消息      * @return      */@PostMapping("/queue/{msg}")publicStringsendQueue(@PathVariable("msg")String msg){ActiveMQQueue queue=newActiveMQQueue("testq");         jmsMessagingTemplate.convertAndSend(queue, msg);return"success";}/**      * topic 模式      * @param msg 消息      * @return      */@PostMapping("/topic/{msg}")publicStringsendTopic(@PathVariable("msg")String msg){ActiveMQTopic topic=newActiveMQTopic("test");         jmsMessagingTemplate.convertAndSend(topic, msg);return"success";}}

消费者

importcom.njc.mqconverter.property.KafkaTopicProperty;importlombok.extern.slf4j.Slf4j;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.jms.annotation.JmsListener;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.stereotype.Component;@Component@Slf4jpublicclassActiveMQConsumer{@AutowiredKafkaTemplate kafkaTemplate;/**      * queue模式的消费者      */@JmsListener(destination="testq", containerFactory="queueListener")publicvoidreadActiveQueue(String message){String topic=KafkaTopicProperty.getSap();         kafkaTemplate.send(topic, message);         log.info("activeMQ 消费者接收：{}", message);}/**      * topic 模式的消费者      */@JmsListener(destination="test", containerFactory="topicListener")publicvoidreadActivTopic1(String message){         log.info("activeMQ topic 消费者1接收：{}", message);}/**      * topic 模式的消费者      */@JmsListener(destination="test", containerFactory="topicListener")publicvoidreadActivTopic2(String message){         log.info("activeMQ topic 消费者2接收：{}", message);}}

监听者的队列模式可以不配 containerFactory ，但topic的需要配置