spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.in-memory=true
spring.activemq.pool.enabled=false
spring.activemq.password=admin
spring.activemq.user=admin
#默認(rèn)值false，表示point to point（點(diǎn)到點(diǎn)）模式，true時(shí)代表發(fā)布訂閱模式，需要手動(dòng)開(kāi)啟
#spring.jms.pub-sub-domain=true

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.jms.Destination;

/**
 * 生產(chǎn)者
 */
@Service
public class Publisher {
  @Autowired
  private JmsMessagingTemplate jmsMsgTemplate;

  /**
   * 發(fā)送topic
   *
   * @param destination
   * @param message
   */
  public void publish(Destination destination, String message) {
    jmsMsgTemplate.convertAndSend(destination, message);
  }
}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;

/**
 * 消費(fèi)者
 */
@Service
public class Subscriber2 {
  private static Logger logger = LoggerFactory.getLogger(Subscriber2.class);

  @JmsListener(destination = "topicListener2")
  public void subscriber(String text) {
    logger.info("Subscriber2 收到的報(bào)文：{}", text);
  }
}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import javax.jms.JMSException;

/**
 * 消費(fèi)者
 */
@Component
public class Subscriber1 {
  private static Logger logger = LoggerFactory.getLogger(Subscriber1.class);

  /**
   * 訂閱 topicListener1
   *
   * @param text
   * @throws JMSException
   */
  @JmsListener(destination = "topicListener1")
  public void subscriber(String text) {
    logger.info("Subscriber1 收到的報(bào)文：{}", text);
  }

}

@Test
  public void topicTest() {
    // 設(shè)置話題監(jiān)聽(tīng)者，可以自由切換
    Destination destination = new ActiveMQTopic("topicListener2");
    for (int i = 0; i < 6; i++) {
      publisher.publish(destination, "Topic Message " + i);
    }
    try {
      Thread.sleep(300);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("使線程睡 300 毫秒，保證消費(fèi)者消費(fèi)完畢!");
  }