package cn.slimsmart.activemq.demo.topic;  import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;  import org.apache.activemq.ActiveMQConnectionFactory;  public class Producer {  public static void main(String[] args) throws JMSException {  // 连接到ActiveMQ服务器  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.18.43:61616");  Connection connection = factory.createConnection();  connection.start();  Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  // 创建主题  Topic topic = session.createTopic("slimsmart.topic.test");  MessageProducer producer = session.createProducer(topic);  // NON_PERSISTENT 非持久化 PERSISTENT 持久化,发送消息时用使用持久模式
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);  TextMessage message = session.createTextMessage();  message.setText("topic 消息。");  message.setStringProperty("property", "消息Property");  // 发布主题消息
        producer.send(message);  System.out.println("Sent message: " + message.getText());  session.commit();  session.close();  connection.close();  }
}  

package cn.slimsmart.activemq.demo.topic;  import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;  import org.apache.activemq.ActiveMQConnectionFactory;  /** * 持久订阅设置唯一的客户端ID和订阅者ID。 */
public class ConsumerPersistent {  public static void main(String[] args) throws JMSException {  String clientId = "client_id";  // 连接到ActiveMQ服务器  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.18.43:61616");  Connection connection = factory.createConnection();  //客户端ID,持久订阅需要设置
        connection.setClientID(clientId);  connection.start();  Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);  // 创建主题  Topic topic = session.createTopic("slimsmart.topic.test");  // 创建持久订阅,指定客户端ID。  MessageConsumer consumer = session.createDurableSubscriber(topic,clientId);  consumer.setMessageListener(new MessageListener() {  // 订阅接收方法  public void onMessage(Message message) {  TextMessage tm = (TextMessage) message;  try {  System.out.println("Received message: " + tm.getText()+":"+tm.getStringProperty("property"));  } catch (JMSException e) {  e.printStackTrace();  }  }  });  }
}