springboot 集成activeMQ实现消息队列和双向队列

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
</dependency>

# ActiveMQ--------------------------------------------------------------------------------------------------------------
# Specify if the default broker URL should be in memory. Ignored if an explicit broker has been specified.
# 默认为true表示使用内存的activeMQ，不需要安装activeMQ server
#spring.activemq.in-memory=false
# URL of the ActiveMQ broker. Auto-generated by default. For instance `tcp://localhost:61616`
spring.activemq.broker-url=tcp://localhost:61616
# Login user of the broker.
spring.activemq.user=admin
# Login password of the broker.
spring.activemq.password=admin
# Trust all packages.
#spring.activemq.packages.trust-all=false
# Comma-separated list of specific packages to trust (when not trusting all packages).
#spring.activemq.packages.trusted=
# See PooledConnectionFactory.
#spring.activemq.pool.configuration.*=
# Whether a PooledConnectionFactory should be created instead of a regular ConnectionFactory.
spring.activemq.pool.enabled=true
# Maximum number of pooled connections.
spring.activemq.pool.max-connections=50
# Connection expiration timeout in milliseconds.
spring.activemq.pool.expiry-timeout=10000
# Connection idle timeout in milliseconds.
spring.activemq.pool.idle-timeout=30000
# 如果为True，则是Topic；如果是false或者默认，则是queue。
spring.jms.pub-sub-domain=false

@Service("producer")
public class Producer {

@Resource
private JmsTemplate jmsTemplate;

/**
* 发送消息
*
* @param destination 发送到的队列
* @param message     待发送的消息
*/
public void convertAndSend(Destination destination, final String message) {
jmsTemplate.convertAndSend(destination, message);
}
}

@Component
public class ConsumerA {

/**
* 使用JmsListener配置消费者监听的队列
*
* @param text 接收到的消息
*/
@JmsListener(destination = "suimh_queue")
public void receiveQueue(String text) {
System.out.println("Consumer-A : 收到的报文为:" + text);
}
}

@Component
public class ConsumerB {

/**
* 使用JmsListener配置消费者监听的队列
*
* @param text 接收到的消息
*/
@JmsListener(destination = "suimh_queue")
@SendTo("out.queue")
public String receiveQueue(String text) {
System.out.println("Consumer-B : 收到的报文为:" + text);
return text;
}
}

@Component
public class ConsumerC {

/**
* 使用JmsListener配置消费者监听的队列
*
* @param text 接收到的消息
*/
@JmsListener(destination = "out.queue")
public void consumerMessage(String text) {
System.out.println("Consumer-C : 从out.queue队列收到的回复报文为:" + text);
}
}

/**
* activeMQ测试方法（queue队列模式）
* 如果想用Topic发布订阅模式，需要新建ActiveMQTopic实例，并修改配置文件spring.jms.pub-sub-domain=true
*
* @return String
*/
@GetMapping(value = "/activeMqSendMes")
@ResponseBody
public String activeMqSendMes() {
int num = 10;
try {
Destination destinationQueue = new ActiveMQQueue("suimh_queue");
for (int i = 1; i <= num; i++) {
producer.convertAndSend(destinationQueue, "这是queueProducer发送的第" + i + "个消息！");
}
return "activeMQ生产成功！";
} catch (Exception e) {
return "activeMQ生产失败！";
}
}