activemq釋出訂閱者模式createTopic
阿新 • • 發佈:2018-11-22
一個釋出者,多個訂閱者
1.釋出者JMSProducer
package com.java1234.activemq2; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 訊息生產者-訊息釋出者 * @author Administrator * */ public class JMSProducer { private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 預設的連線使用者名稱 private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 預設的連線密碼 private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 預設的連線地址 private static final int SENDNUM=10; // 傳送的訊息數量 public static void main(String[] args) { ConnectionFactory connectionFactory; // 連線工廠 Connection connection = null; // 連線 Session session; // 會話 接受或者傳送訊息的執行緒 Destination destination; // 訊息的目的地 MessageProducer messageProducer; // 訊息生產者 // 例項化連線工廠 connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL); try { connection=connectionFactory.createConnection(); // 通過連線工廠獲取連線 connection.start(); // 啟動連線 session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 建立Session // destination=session.createQueue("FirstQueue1"); // 建立訊息佇列 destination=session.createTopic("FirstTopic1"); messageProducer=session.createProducer(destination); // 建立訊息生產者 sendMessage(session, messageProducer); // 傳送訊息 session.commit(); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } finally{ if(connection!=null){ try { connection.close(); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } } /** * 傳送訊息 * @param session * @param messageProducer * @throws Exception */ public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{ for(int i=0;i<JMSProducer.SENDNUM;i++){ TextMessage message=session.createTextMessage("ActiveMQ 傳送的訊息"+i); System.out.println("傳送訊息:"+"ActiveMQ 釋出的訊息"+i); messageProducer.send(message); } } }
訂閱者一
package com.java1234.activemq2; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.Session; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 訊息消費者-訊息訂閱者一 * @author Administrator * */ public class JMSConsumer { private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 預設的連線使用者名稱 private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 預設的連線密碼 private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 預設的連線地址 public static void main(String[] args) { ConnectionFactory connectionFactory; // 連線工廠 Connection connection = null; // 連線 Session session; // 會話 接受或者傳送訊息的執行緒 Destination destination; // 訊息的目的地 MessageConsumer messageConsumer; // 訊息的消費者 // 例項化連線工廠 connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL); try { connection=connectionFactory.createConnection(); // 通過連線工廠獲取連線 connection.start(); // 啟動連線 session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 建立Session // destination=session.createQueue("FirstQueue1"); // 建立連線的訊息佇列 destination=session.createTopic("FirstTopic1"); messageConsumer=session.createConsumer(destination); // 建立訊息消費者 messageConsumer.setMessageListener(new Listener()); // 註冊訊息監聽 } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } package com.java1234.activemq2; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; /** * 訊息監聽-訂閱者一 * @author Administrator * */ public class Listener implements MessageListener{ @Override public void onMessage(Message message) { // TODO Auto-generated method stub try { System.out.println("訂閱者一收到的訊息:"+((TextMessage)message).getText()); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
訂閱者二
package com.java1234.activemq2; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.Session; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 訊息消費者-訊息訂閱者二 * @author Administrator * */ public class JMSConsumer2 { private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 預設的連線使用者名稱 private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 預設的連線密碼 private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 預設的連線地址 public static void main(String[] args) { ConnectionFactory connectionFactory; // 連線工廠 Connection connection = null; // 連線 Session session; // 會話 接受或者傳送訊息的執行緒 Destination destination; // 訊息的目的地 MessageConsumer messageConsumer; // 訊息的消費者 // 例項化連線工廠 connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL); try { connection=connectionFactory.createConnection(); // 通過連線工廠獲取連線 connection.start(); // 啟動連線 session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 建立Session // destination=session.createQueue("FirstQueue1"); // 建立連線的訊息佇列 destination=session.createTopic("FirstTopic1"); messageConsumer=session.createConsumer(destination); // 建立訊息消費者 messageConsumer.setMessageListener(new Listener2()); // 註冊訊息監聽 } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } package com.java1234.activemq2; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; /** * 訊息監聽-訂閱者二 * @author Administrator * */ public class Listener2 implements MessageListener{ @Override public void onMessage(Message message) { // TODO Auto-generated method stub try { System.out.println("訂閱者二收到的訊息:"+((TextMessage)message).getText()); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }