ActiveMQ实战之 Topic发布订阅消息
jopen
13年前
<p>对于此类消息,其实就是指使用JMS中的发布订阅消息模型的消息,下面是一个简单的例子</p> <p>消息发布者</p> <pre class="brush:java; toolbar: true; auto-links: false;">package com.googlecode.garbagecan.jmsstudy.activemq.topic; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class TopicPublisher { public static void main(String[] args) throws JMSException { ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); Connection connection = factory.createConnection(); connection.start(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Topic topic = session.createTopic("myTopic.messages"); MessageProducer producer = session.createProducer(topic); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); while(true) { TextMessage message = session.createTextMessage(); message.setText("message_" + System.currentTimeMillis()); producer.send(message); System.out.println("Sent message: " + message.getText()); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } // session.close(); // connection.stop(); // connection.close(); } }</pre>消息订阅者(消息消费者) <pre class="brush:java; toolbar: true; auto-links: false;">package com.googlecode.garbagecan.jmsstudy.activemq.topic; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class TopicSubscriber { public static void main(String[] args) throws JMSException { ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); Connection connection = factory.createConnection(); connection.start(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Topic topic = session.createTopic("myTopic.messages"); MessageConsumer consumer = session.createConsumer(topic); consumer.setMessageListener(new MessageListener() { public void onMessage(Message message) { TextMessage tm = (TextMessage) message; try { System.out.println("Received message: " + tm.getText()); } catch (JMSException e) { e.printStackTrace(); } } }); // session.close(); // connection.stop(); // connection.close(); } }</pre>分别运行两个类,就可以看到Publisher发布的类,Subscriber都可以接受到。 <p></p>