Artemis采取了不同的方法。它只在内部实现队列，所有消息传送概念都是通过使用消息地址路由到适当的队列来实现的。如：发布订阅（topics）和点对点（queues）之类的消息传递概念是使用不同类型的路由机制实现。广播（Multicast）路由用于实现发布订阅（publish-subscribe）规程，所有订阅者将通过路由将获得自己的内部队列和消息。选播（Anycast）路由用于实现点对点规程，其中每一个地址只有一个队列，所有消费者都订阅它。所有协议都使用寻址和路由方案。例如，您可以将JMS topic作为广播地址，用于发送给消费者。

消息队列

点对点消息（queue）：

通常将消息发送到队列，队列将消息持久化后，将消息推送给多个消费者，但只有其中一个能够接收成功。如果生产者在生产消息的时候，并无消费者在线，那么会将消息进行持久化，待消费者上线后，将消息发送给消费者。当消费者接收消息成功后，会隐式地发送一个ack给消息中间件，消息中间件收到ack后将消息从消息队列中标记为已消费并从队列中移除。

发布订阅（topic）：

将消息发送到称为Topic的实体上，一个Topic可以有许多消费者（订阅者），每个订阅者都能收到发布者（生产者）的消息。生产订阅消息分为持久性订阅和非持久性订阅。
持久性订阅会将消息进行持久化而保存每一条消息，直到所有订阅者都收到了它们。
非持久订阅仅持续到创建它们的连接的最长生命周期。
消息传递可靠性，消息的可靠性有三种传输保障：

  • At most once，至多一次，消息可能丢失但是不会重复发送；
  • At least once，至少一次，消息不会丢失，但是可能会重复；
  • Exactly once，精确一次，每条消息肯定会被传输一次且仅一次。

事务消息：
事务消息指的是针对事务开始和事务结束期间的全体操作，要不全部成功，要不全部失败。Artemis支持本地的多个消息的发送和确认事务。
消息持久化：
需要持久化的消息存储在磁盘中，不需要持久化的消息存储在内存中。

使用Artemis

首先我们我们下载并安装了broker，我们就会遇到与ActiveMQ第一个不同之处。使用Artemis，您需要显式地创建一个broker实例，而在ActiveMQ上，这个步骤是可选的。这个步骤的意思是将broker的安装和配置分开，这使得以后升级和维护代理更加容易。
所以，在启动Artemis之前，你需要执行这样的操作：

  1. $ bin/artemis create clown /data/clown

不管你在什么操作系统上安装broker二进制文件，broker实例都会位于像“/opt/artemis”目录中。对每个ActiveMQ用户都熟悉这个文件夹的内容：

  • bin – 包含管理broker的shell脚本(start, stop, etc.)
  • data – 每一个broker数据存储 (message store)
  • etc – 包含broker配置文件(it’s what conf directory is in ActiveMQ)
  • log – Artemis logs存储目录, 而不像ActiveMQ将日志保存在data目录中
  • tmp – 临时文件目录

现在让我们更详细地了解一下配置。

  • etc/bootstrap.xml
    • 文件用于设置基本信息，如：主broker配置文件的位置、实用程序（如Web服务器）和JAAS安全性。
  • etc/broker.xml
    • 与ActiveMQ’s中的conf/activemq.xml类似，用于配置broker的大部分信息，如：端口、名称、安全策略等。
  • etc/artemis.profile
    • 文件与ActiveMQ中的bin/env文件类似。您可以在这个文件中配置broker环境变量，主要：SSL、调试等相关的JVM常规参数。
  • etc/logging.properties
    • 两个brokers之间的日志配置没有多大差别，所以任何熟悉Java日志记录系统的人都会配置这个文件。

最后，我们有JAAS配置文件 (login.config, artemis-users.properties and artemis-roles.properties)，它们与ActiveMQ中的配置相同。
现在我们可以启动Artemis：

  1. $ bin/artemis run

Spring配置Artemis

maven配置

  1. <!-- https://mvnrepository.com/artifact/org.apache.activemq/artemis-jms-client -->
  2. <dependency>
  3.     <groupId>org.apache.activemq</groupId>
  4.     <artifactId>artemis-jms-client</artifactId>
  5.     <version>2.7.0</version>
  6. </dependency>

Java配置

  1. private ActiveMQConnectionFactory factory;
  2. private Connection connection;
  4. @Autowired
  5. private ActiveMQQueueListener listener;
  7. public ActiveMQConfig() {
  8.     factory = new ActiveMQConnectionFactory(tcp_uri);
  9.     factory.setCacheDestinations(true);
  11.     try {
  12.         connection = factory.createConnection();
  13.         connection.start();
  14.     } catch (JMSException e) {
  15.         e.printStackTrace();
  16.     }
  17. }

首先我们需要一个ActiveMQ的连接器工厂类，然后还需要连接到ActiveMQ。

  1. @Bean(destroyMethod = "close")
  2. public Session activeMQSession() throws JMSException {
  3.     Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  4.     return session;
  5. }
  7. /**
  8.  * Queue生产者
  9.  *
  10.  * @param session
  11.  * @return
  12.  * @throws JMSException
  13.  */
  14. @Bean
  15. public MessageProducer messageProducer(Session session) throws JMSException {
  16.     if (mqType == 0) {
  17.         return session.createProducer(session.createQueue(QUEUE_DEFAULT));
  18.     }
  19.     return session.createProducer(session.createTopic(TOPIC_DEFAULT));
  20. }

这里我们需要创建一个生产者（MessageProducer），mqType是一个int对象，如：private int mqType = 1;//0:queue|1:topic|2:mqtt
这里为了做实验，便于切换，但是在实际应用场合中，我们应该是既使用Queue也会使用Topic，我将会在最后贴出ActiveMQService服务类，此类实现了对Artemis的基本操作。

  1. /**
  2.  * Queue消费者
  3.  *
  4.  * @param session
  5.  * @return
  6.  * @throws JMSException
  7.  */
  8. @Bean
  9. public MessageConsumer messageConsumer(Session session) throws JMSException {
  10.     MessageConsumer consumer = null;
  11.     if (mqType == 0) {
  12.         consumer = session.createConsumer(session.createQueue(QUEUE_DEFAULT));
  13.     } else {
  14.         consumer = session.createConsumer(session.createTopic(TOPIC_DEFAULT));
  15.     }
  16.     consumer.setMessageListener(listener);
  17.     return consumer;
  18. }

最后我们只需要创建一个消费者（MessageConsumer），配置文件非常简单，现在可以用来测试。

  1. @RestController
  2. @RequestMapping(value = "/test")
  3. public class TestActiveMQController {
  5.     @Autowired
  6.     private IActiveMQProducerService producerService;
  8.     @RequestMapping(value = "/activemq/queue")
  9.     public String testQueue(HttpServletRequest request) {
  10.         String str = "test queue " + DateTimeUtils.getTimeInt();
  11.         producerService.send(str);
  12.         return str;
  13.     }
  15.     @RequestMapping(value = "/activemq/mqtt")
  16.     public String testMqtt(HttpServletRequest request) {
  17.         String str = "test mqtt " + DateTimeUtils.getTimeInt();
  18.         producerService.push(str);
  19.         return str;
  20.     }
  21. }
  23. //生产者服务类
  24. @Service
  25. public class ActiveMQProducerServiceImpl extends CompactService implements IActiveMQProducerService {
  27.     @Autowired
  28.     private MessageProducer producer;
  30.     @Autowired
  31.     private Session session;
  33.     @Autowired
  34.     private BlockingConnection blockingConnection;
  37.     public ActiveMQProducerServiceImpl() {
  39.     }
  41.     /**
  42.      * Queue & Topic send
  43.      *
  44.      * @param message
  45.      */
  46.     @Override
  47.     public void send(String message) {
  48.         try {
  49.             producer.send(session.createTextMessage(message));
  50.         } catch (JMSException e) {
  51.             e.printStackTrace();
  52.         }
  53.     }
  55.     /**
  56.      * MQTT push
  57.      *
  58.      * @param message
  59.      */
  60.     @Override
  61.     public void push(String message) {
  62.         try {
  63.             blockingConnection.publish(ActiveMQConfig.MQTT_DEFAULT, message.getBytes(), QoS.AT_LEAST_ONCE, false);
  64.         } catch (Exception e) {
  65.             e.printStackTrace();
  66.         }
  67.     }
  69. }

为了测试，随便整合了一下，但是后来觉得，其实没有必要使用config的方式进行整合，以为Apache ActiveMQ Artemis是一个非常大的消息队列，应该以服务层的概念来整合进去，所以后来写了一个Artemis服务对象类：

整合到服务层

接口IActiveMQService

  1. import javax.jms.*;
  4. public interface IActiveMQService {
  6.     Session createSession() throws JMSException;
  8.     Queue createQueue(Session session, String name) throws JMSException;
  10.     Topic createTopic(Session session, String name) throws JMSException;
  12.     MessageProducer getMessageProducer(Queue queue) throws JMSException;
  14.     MessageProducer getMessageProducer(Topic topic) throws JMSException;
  16.     MessageProducer createMessageProducer(Destination destination, String name) throws JMSException;
  18.     MessageConsumer getMessageConsumer(Session session, Queue queue) throws JMSException;
  20.     MessageConsumer getMessageConsumer(Session session, Topic topic) throws JMSException;
  22.     MessageConsumer createMessageConsumer(Session session, Destination destination, String name) throws JMSException;
  24.     boolean send(String name, String type, String message);
  26.     boolean send(String name, String type, String message, CompletionListener listener);
  28.     boolean send(MessageProducer producer, String message);
  30.     boolean send(MessageProducer producer, String message, CompletionListener listener);
  32.     void addListener(String name, String type, MessageListener listener);
  35. }

实现ActiveMQService：

  1. import com.lanxinbase.system.basic.CompactService;
  2. import com.lanxinbase.system.service.resource.IActiveMQService;
  3. import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
  4. import org.springframework.beans.factory.DisposableBean;
  5. import org.springframework.beans.factory.InitializingBean;
  6. import org.springframework.stereotype.Service;
  8. import javax.jms.*;
  9. import java.util.Map;
  10. import java.util.concurrent.ConcurrentHashMap;
  12. @Service
  13. public class ActiveMQService extends CompactService implements IActiveMQService, InitializingBean, DisposableBean {
  15.     public static final String TYPE_QUEUE = "queue";
  16.     public static final String TYPE_TOPIC = "topic";
  18.     private static final String TOPIC_DEFAULT = "topic/default/";
  19.     private static final String QUEUE_DEFAULT = "queue/default/";
  20.     private static final String tcp_uri = "tcp://0.0.0.0:61616";
  22.     private ActiveMQConnectionFactory factory;
  23.     private Connection connection;
  24.     private Session session;
  26.     private Map<String, MessageProducer> producerMap = new ConcurrentHashMap<>();
  28.     public ActiveMQService() {
  30.     }
  32.     /**
  33.      * 首次启动运行
  34.      *
  35.      * @throws Exception
  36.      */
  37.     @Override
  38.     public void afterPropertiesSet() throws Exception {
  39.         factory = new ActiveMQConnectionFactory(tcp_uri);
  40.         factory.setCacheDestinations(true);
  42.         connection = factory.createConnection();
  43.         connection.start();
  44.         createSession();
  45.     }
  47.     /**
  48.      * 私有的session，主要用于生产者
  49.      *
  50.      * @return
  51.      * @throws JMSException
  52.      */
  53.     private Session getSession() throws JMSException {
  54.         if (session == null) {
  55.             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  56.         }
  57.         return session;
  58.     }
  60.     /**
  61.      * 创建一个新的session，主要应用于消费者
  62.      *
  63.      * @return
  64.      * @throws JMSException
  65.      */
  66.     @Override
  67.     public Session createSession() throws JMSException {
  68.         return connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  69.     }
  71.     /**
  72.      * 创建一个Queue
  73.      *
  74.      * @param s    session
  75.      * @param name 如：test
  76.      * @return
  77.      * @throws JMSException
  78.      */
  79.     @Override
  80.     public Queue createQueue(Session s, String name) throws JMSException {
  81.         if (name == null) {
  82.             name = "";
  83.         }
  84.         return s.createQueue(QUEUE_DEFAULT + name);
  85.     }
  87.     /**
  88.      * 创建一个Topic
  89.      *
  90.      * @param s    session
  91.      * @param name 如：test
  92.      * @return
  93.      * @throws JMSException
  94.      */
  95.     @Override
  96.     public Topic createTopic(Session s, String name) throws JMSException {
  97.         if (name == null) {
  98.             name = "";
  99.         }
  100.         return s.createTopic(TOPIC_DEFAULT + name);
  101.     }
  103.     /**
  104.      * 获取一个生产者
  105.      *
  106.      * @param queue
  107.      * @return
  108.      * @throws JMSException
  109.      */
  110.     @Override
  111.     public MessageProducer getMessageProducer(Queue queue) throws JMSException {
  112.         return createMessageProducer(queue, queue.getQueueName());
  113.     }
  115.     /**
  116.      * 获取一个生产者
  117.      *
  118.      * @param topic
  119.      * @return
  120.      * @throws JMSException
  121.      */
  122.     @Override
  123.     public MessageProducer getMessageProducer(Topic topic) throws JMSException {
  124.         return createMessageProducer(topic, topic.getTopicName());
  125.     }
  127.     /**
  128.      * 创建一个生产者
  129.      *
  130.      * @param destination
  131.      * @param name        缓存key，同时等于queue&topic的name
  132.      * @return
  133.      * @throws JMSException
  134.      */
  135.     @Override
  136.     public MessageProducer createMessageProducer(Destination destination, String name) throws JMSException {
  137.         logger(name);
  138.         MessageProducer producer = producerMap.get(name);
  139.         if (producer == null) {
  140.             producer = session.createProducer(destination);
  141.             producerMap.put(name, producer);
  142.         }
  143.         return producer;
  144.     }
  146.     /**
  147.      * 获取一个消费者
  148.      *
  149.      * @param s     可以通过createSession创建
  150.      * @param queue
  151.      * @return
  152.      * @throws JMSException
  153.      */
  154.     @Override
  155.     public MessageConsumer getMessageConsumer(Session s, Queue queue) throws JMSException {
  156.         return createMessageConsumer(s, queue, queue.getQueueName());
  157.     }
  159.     /**
  160.      * 获取一个消费者
  161.      *
  162.      * @param s     可以通过createSession创建
  163.      * @param topic
  164.      * @return
  165.      * @throws JMSException
  166.      */
  167.     @Override
  168.     public MessageConsumer getMessageConsumer(Session s, Topic topic) throws JMSException {
  169.         return createMessageConsumer(s, topic, topic.getTopicName());
  170.     }
  172.     /**
  173.      * 创建一个消费者
  174.      *
  175.      * @param s           session（本来是把session做成单例，但是消费者应该是动态的，不同于生产者，所以这里需要随时创建一个session）
  176.      * @param destination
  177.      * @param name        废弃
  178.      * @return
  179.      * @throws JMSException
  180.      */
  181.     @Override
  182.     public MessageConsumer createMessageConsumer(Session s, Destination destination, String name) throws JMSException {
  183. //        MessageConsumer consumer = consumerMap.get(name);
  184.         MessageConsumer consumer = null;
  185. //        if (consumer == null) {
  186. //            consumer = session.createConsumer(destination);
  187. //            consumerMap.put(name, consumer);
  188. //        }
  189.         consumer = s.createConsumer(destination);
  190.         return consumer;
  191.     }
  193.     /**
  194.      * 生产一条消息
  195.      *
  196.      * @param name    Queue|Topic的name
  197.      * @param type    类型：Queue|Topic
  198.      * @param message 字符串消息，通常应该是JSON字符串
  199.      * @return
  200.      */
  201.     @Override
  202.     public boolean send(String name, String type, String message) {
  203.         return this.send(name, type, message, null);
  204.     }
  206.     @Override
  207.     public boolean send(String name, String type, String message, CompletionListener listener) {
  208.         MessageProducer producer;
  209.         try {
  210.             if (TYPE_QUEUE.equals(type)) {
  211.                 producer = getMessageProducer(createQueue(getSession(), name));
  212.             } else {
  213.                 producer = getMessageProducer(createTopic(getSession(), name));
  214.             }
  215.             if (listener == null) {
  216.                 return this.send(producer, message);
  217.             } else {
  218.                 return this.send(producer, message, listener);
  219.             }
  221.         } catch (JMSException e) {
  222.             e.printStackTrace();
  223.         }
  224.         return false;
  225.     }
  228.     /**
  229.      * 发送消息
  230.      *
  231.      * @param producer 生产者
  232.      * @param message  消息字符串
  233.      * @return
  234.      */
  235.     @Override
  236.     public boolean send(MessageProducer producer, String message) {
  237.         try {
  238.             producer.send(session.createTextMessage(message));
  239.             return true;
  240.         } catch (JMSException e) {
  241.             e.printStackTrace();
  242.         }
  243.         return false;
  244.     }
  246.     @Override
  247.     public boolean send(MessageProducer producer, String message, CompletionListener listener) {
  248.         try {
  249.             producer.send(session.createTextMessage(message), listener);
  250.             return true;
  251.         } catch (JMSException e) {
  252.             e.printStackTrace();
  253.         }
  254.         return false;
  255.     }
  257.     /**
  258.      * 消费者监听
  259.      *
  260.      * @param name     Queue|Topic的name
  261.      * @param type     类型：Queue|Topic
  262.      * @param listener 监听回调
  263.      */
  264.     @Override
  265.     public void addListener(String name, String type, MessageListener listener) {
  266.         MessageConsumer consumer = null;
  267.         try {
  268.             Session session = createSession();
  269.             if (TYPE_QUEUE.equals(type)) {
  270.                 consumer = getMessageConsumer(session, createQueue(session, name));
  271.             } else {
  272.                 consumer = getMessageConsumer(session, createTopic(session, name));
  273.             }
  274.             this.addListener(consumer, listener);
  275.         } catch (JMSException e) {
  276.             e.printStackTrace();
  277.         }
  278.     }
  280.     private void addListener(MessageConsumer consumer, MessageListener listener) {
  281.         try {
  282.             consumer.setMessageListener(listener);
  283.         } catch (JMSException e) {
  284.             e.printStackTrace();
  285.         }
  286.     }
  288.     /**
  289.      * 程序退出时需要关闭或停止连接
  290.      *
  291.      * @throws Exception
  292.      */
  293.     @Override
  294.     public void destroy() throws Exception {
  295.         if (session != null) {
  296.             session.close();
  297.         }
  298.         connection.stop();
  299.         connection.close();
  300.         factory.close();
  301.     }
  304. }

由于每个方法都加了注释，所以这里就不更多的废话了。但是在public MessageConsumer createMessageConsumer()方法中我使用的是创建了一个全新的Session来创建消费者，为什么要这样子做呢？
在生产环境中应该跟createMessageProducer()方法中的session一样使用单列的session，这里需要注意就是这个session我并没有研究过有没有过期时间，在生产环境中，需要实时传输消息，理论上是不会过期。
之所以这样做，我认为在用户登陆APP上线之后会执行一个会话（创建一个新的session），当APP下线之后也应该移除这个Session，示例代码只是为了做测试，所以没有过多的实现，在实际应用中应该根据自身的业务需求来改进逻辑代码。

最后测试TestActiveMQController：

  1. @RestController
  2. @RequestMapping(value = "/test")
  3. public class TestActiveMQController {
  5.     private static final Logger logger = Logger.getLogger("TestActiveMQController>");
  7.     @Resource
  8.     private IActiveMQService activeMQService;
  11.     /**
  12.      * /test/activemq/pub/add?name=testQueue&type=queue
  13.      * /test/activemq/pub/add?name=testQueue1&type=queue
  14.      * <p>
  15.      * /test/activemq/pub/add?name=testTopic&type=topic
  16.      * /test/activemq/pub/add?name=testTopic1&type=topic
  17.      *
  18.      * @param request
  19.      * @return
  20.      */
  21.     @RequestMapping(value = "/activemq/pub/add")
  22.     public ResultResp<Void> pubAdd(HttpServletRequest request) {
  23.         ResultResp<Void> resp = new ResultResp<>();
  25.         String name = request.getParameter("name");
  26.         String type = request.getParameter("type");
  28.         String msg = "test mqtt " + DateTimeUtils.getTimeInt();
  30.         activeMQService.send(name, type, msg);
  32.         resp.setInfo(msg);
  33.         return resp;
  34.     }
  36.     /**
  37.      * /test/activemq/sub/add?id=100&name=testQueue&type=queue
  38.      * /test/activemq/sub/add?id=101&name=testQueue1&type=queue
  39.      * <p>
  40.      * /test/activemq/sub/add?id=100&name=testTopic&type=topic
  41.      * /test/activemq/sub/add?id=101&name=testTopic&type=topic
  42.      *
  43.      * @param request
  44.      * @return
  45.      */
  46.     @RequestMapping(value = "/activemq/sub/add")
  47.     public ResultResp<Void> subAdd(HttpServletRequest request) {
  48.         ResultResp<Void> resp = new ResultResp<>();
  50.         String id = request.getParameter("id");
  51.         String name = request.getParameter("name");
  52.         String type = request.getParameter("type");
  54.         activeMQService.addListener(name, type, (m) -> {
  55.             TextMessage textMessage = (TextMessage) m;
  56.             try {
  57.                 logger.info(id + " : " + textMessage.getText());
  58.             } catch (JMSException e) {
  59.                 e.printStackTrace();
  60.             }
  61.         });
  63.         return resp;
  64.     }
  65. }

SpringBoot配置Artemis

POM文件

  1. // 引入activemq
  2. <dependency>
  3.     <groupId>org.springframework.boot</groupId>
  4.     <artifactId>spring-boot-starter-activemq</artifactId>
  5. </dependency>
  8. // 引入artemis
  9. <dependency>
  10.     <groupId>org.springframework.boot</groupId>
  11.     <artifactId>spring-boot-starter-artemis</artifactId>
  12. </dependency>

配置文件

配置activemq

  1. spring:
  2.   activemq:
  3.     broker-url: tcp://${ACTIVEMQ_HOST, localhost}:61616 # activemq连接地址
  4.     user: ${ACTIVEMQ_USER, admin} # 用户名
  5.     password: ${ACTIVEMQ_PASSWORD, admin} # 密码
  6.     send-timeout: # 发送超时时间
  7.     pool:
  8.       enabled: false # 是否创建 JmsPoolConnectionFactory 连接池
  9.       idle-timeout: 30s # 空闲连接超时时间
  10.       max-connections: 50 # 连接池中最大连接数
  11.       max-sessions-per-connection: 100 # 每个连接最大会话

配置artemis

  1. spring:
  2.   artemis:
  3.     mode: native
  4.     host: ${ARTEMIS_HOST, localhost} # artermis连接地址
  5.     port: ${ARTEMIS_PORT, 9876} # artermis连接端口
  6.     user: ${ARTEMIS_USER, admin} # 用户名
  7.     password: ${ARTEMIS_PASSWORD, admin} # 密码
  8.     send-timeout: # 发送超时时间
  9.     pool:
  10.       enabled: false # 是否创建 JmsPoolConnectionFactory 连接池
  11.       idle-timeout: 30s # 空闲连接超时时间
  12.       max-connections: 50 # 连接池中最大连接数
  13.       max-sessions-per-connection: 100 # 每个连接最大会话

Spring JMS 中默使用 CachingConnectionFactory 创建连接池，如果指定 JmsPoolConnectionFactory 连接池，测在 spring.jms.* 中配置连接池属性。

  1. spring:
  2.   jms:
  3.     session-cache-size: 5
  4.     # 是否允许发布订阅
  5.     pub-sub-domain: true

关于 Activemq (Artemis) 和 Apring Jms 更多配置，可参考：Spring Boot Integration Properties

使用

JmsTemplate 是 Spring JMS 中用来发送消息的类。做完以上配置，在 Spring Boot 启动入口加上 @EnableJms 注解，项目在启动时会自动装配，在项目中直接注入 JmsTemplate 对象。

  1. @SpringBootApplication
  2. @EnableJms
  3. public class Application {
  5.     @Autowired
  6.     private JmsTemplate jmsTemplate;
  8.    public static void main(String[] args) {
  9.         SpringApplication.run(Application.class);
  10.     }
  12. }

发送文本消息示例：

  1. public void send(String msg) {
  3.     log.info("发送消息：{}", msg);
  5.     jmsTemplate.send(DEST_NAME, new MessageCreator() {
  6.         @Override
  7.         public Message createMessage(Session session) throws JMSException {
  8.             // 也可以创建对象 session.createObjectMessage()
  9.             TextMessage textMessage = session.createTextMessage();
  10.             textMessage.setText(msg);
  11.             return textMessage;
  12.         }
  13.     });
  14. }

接收使用 @JmsListener 注解：

  1. /**
  2.  * 监听消息
  3.  * @param content
  4.  */
  5. @JmsListener(destination = DEST_NAME, concurrency = "消费线程数")
  6. public void recive(String content) {
  7.     log.info("收到消息：{}", content);
  8. }

发送消息和接收消息的 DEST_NAME 要保持一至。 concurrency 属性非必须，用来设置消费消息的程序数。关于 ActiveMQ 的详细介绍会在专门的章节讲解。

Producer

  1. @Component
  2. public class Producer {
  4.     @Autowired
  5.     private JmsMessagingTemplate jmsMessagingTemplate;
  7.     public void send(Destination destination, final String message) {
  8.         jmsMessagingTemplate.convertAndSend(destination, message + "from queue");
  9.     }
  11.     @JmsListener(destination="out.queue")
  12.     public void consumerMessage(String text){
  13.         System.out.println("从out.queue队列收到的回复信息为:"+text);
  14.     }
  15. }

Consumer

  1. @Component
  2. public class Consumer {
  3.     @JmsListener(destination = "mytest.queue")
  4.     @SendTo("out.queue")
  5.     public String receiveQueue(String text) {
  6.         System.out.println("Consumer收到的信息为:"+text);
  7.         return "return message "+text;
  8.     }
  9. }

Rest使用

  1. @ResponseBody
  2. @RequestMapping(value = "/mqtest", method = RequestMethod.GET)
  3. public Object mqtest() {
  4.     Destination destination = new ActiveMQQueue("mytest.queue");
  5.     producer.send(destination, "I am YeJiaWei");
  6.     return "OK";
  7. }