一个jms activemq Topic 消息实例

import javax.jms.Connection; import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.ExceptionListener; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; /** * 此代码来自官方网站： 自己添加注释： 总结：使用JMS的步骤： （1）创建一個JMS连接工厂 （2）创建 一個JMS连接并启动连接 * （3）创建一個JMS的会话并建立相应的JMS目标對象 (队列) （4）通过相应的消息创建者创建消息和消息消费者接受消息。 * * Hello world! */ public class Test1 { public static void main(String[] args) throws Exception { thread(new HelloWorldProducer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); Thread.sleep(1000); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); Thread.sleep(1000); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldProducer(), false); Thread.sleep(1000); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); } public static void thread(Runnable runnable, boolean daemon) { Thread brokerThread = new Thread(runnable); brokerThread.setDaemon(daemon); brokerThread.start(); } /** * 消息的创建者的应用 * * @author longgangbai * */ public static class HelloWorldProducer implements Runnable { public void run() { try { // 创建一個ActionMQ的连接工厂 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory( "vm://localhost"); // 创建一個ActionMQ连接工厂 Connection connection = connectionFactory.createConnection(); connection.start(); // 创建一個ActionMQ会话 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 创建一個连接ActionMQ的目标(Topic or Queue) Destination destination = session.createQueue("TEST.FOO"); // 创建一個消息的创建者从会话中队列目标中 MessageProducer producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 创建一個消息 String text = "Hello world! From: " + Thread.currentThread().getName() + " : " + this.hashCode(); TextMessage message = session.createTextMessage(text); // 发送一個消息 System.out.println("Sent message: " + message.hashCode() + " : " + Thread.currentThread().getName()); producer.send(message); // 关闭会话和连接 session.close(); connection.close(); } catch (Exception e) { System.out.println("Caught: " + e); e.printStackTrace(); } } } /** * 消息的消费者 * * @author longgangbai * */ public static class HelloWorldConsumer implements Runnable, ExceptionListener { public void run() { try { // 创建一個JMS的连接工厂的 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory( "vm://localhost"); // 创建一個JMS的连接 Connection connection = connectionFactory.createConnection(); // 启动连接對象 connection.start(); connection.setExceptionListener(this); // 创建会话 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 创建目标的對象 Create the destination (Topic or Queue) Destination destination = session.createQueue("TEST.FOO"); // 创建消息的消费者的 Create a MessageConsumer from the Session to the // Topic or // Queue MessageConsumer consumer = session.createConsumer(destination); // 接受相应的消息對應 Wait for a message Message message = consumer.receive(1000); if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; String text = textMessage.getText(); System.out.println("Received: " + text); } else { System.out.println("Received: " + message); } // 关闭会话和连接 consumer.close(); session.close(); connection.close(); } catch (Exception e) { System.out.println("Caught: " + e); e.printStackTrace(); } } /* * 异步消息异常处理 (non-Javadoc) * * @see javax.jms.ExceptionListener#onException(javax.jms.JMSException) */ public synchronized void onException(JMSException ex) { System.out.println("JMS Exception occured. Shutting down client."); } } }