SpringBoot整合ActiveMQ

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<!-- spring boot web支持：mvc,aop... -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

server:
port: 8080
#activemq配置
spring:
activemq:
broker-url: tcp://127.0.0.1:61616
user: admin
password: admin

@Component
public class ActiveMQProvider {
//注入JMSTemplate模板
@Resource
private JmsTemplate jmsTemplate;
//创建方法
public void sendMessage(){
//点对点，创建队列
ActiveMQQueue queue=new ActiveMQQueue("boot_queue");
//发送消息
jmsTemplate.convertAndSend(queue,"生产者产生的消息~");
}
}

@RestController
public class ProController {
@Resource
private ActiveMQProvider activeMQProvider;

@RequestMapping("/sendMessage")
public String sendMessage(){
activeMQProvider.sendMessage();
return "success";
}
}

server:
port: 8081
spring:
activemq:
broker-url: tcp://127.0.0.1:61616
user: admin
password: admin

//消费者消费
@JmsListener(destination = "boot_queue")
public void getMessage(TextMessage message) throws JMSException {
System.out.println("消费者获取到消息："+message.getText());
}

@Component
public class ActiveMQProvider {
//注入JMSTemplate模板
@Resource
private JmsTemplate jmsTemplate;
//创建方法
public void sendMessage(){
//点对点，创建队列
//ActiveMQQueue queue=new ActiveMQQueue("boot_queue");
//发布订阅：创建主题
ActiveMQTopic topic=new ActiveMQTopic("boot_topic");
//springboot默认是queue
jmsTemplate.setPubSubDomain(true);
//发送消息
jmsTemplate.convertAndSend(topic,"生产者产生主题的消息~");

}
}

//springboot默认只配置queue类型消息，如果要使用topic类型的消息，则需要配置该bean
@Bean
public JmsListenerContainerFactory jmsTopicListenerContainerFactory(ConnectionFactory connectionFactory){
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
//这里必须设置为true，false则表示是queue类型
factory.setPubSubDomain(true);
return factory;
}

@JmsListener(destination = "boot_topic",containerFactory = "jmsTopicListenerContainerFactory")
public void getMessage(TextMessage message) throws JMSException {

System.out.println("消费者获取到消息："+message.getText());
}

jmsTemplate.setDeliveryMode(2);
jmsTemplate.setExplicitQosEnabled(true);
jmsTemplate.setDeliveryPersistent(true);

@Bean(name = "topicListenerFactory")
public JmsListenerContainerFactory<DefaultMessageListenerContainer> topicListenerFactory(ConnectionFactory connectionFactory){
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();

factory.setSubscriptionDurable(true);// Set this to "true" to register a durable subscription,

factory.setClientId("B");

factory.setConnectionFactory(connectionFactory);
return factory;

}

//消费者消费 destination队列或者主题的名字
@JmsListener(destination = "boot_topic",containerFactory = "topicListenerFactory")
public void getMessage(TextMessage message,Session session) throws JMSException {
System.out.println("消费者获取到消息："+message.getText());
}