分散式訊息中介軟體(一)——ActiveMQ點對點訊息模式
阿新 • • 發佈:2019-02-13
一、下載執行
1、官網下載
Windows版:apache-activemq-5.13.3-bin.zip
Linux版:apache-activemq-5.13.3-bin.tar.gz
2、Windows下根據作業系統位數,執行activemq.bat檔案,啟動ActiveMQ服務。
注意:需要先配置java環境變數,否則啟動一閃而過。
3、訪問http://localhost:8161/admin ,MQ服務正常訪問
二、ActiveMQ點對點訊息模式
編寫ActiveMQ點對點模式客戶端程式。
1、引入activemq-all-5.11.1.jar包
1、官網下載
Windows版:apache-activemq-5.13.3-bin.zip
Linux版:apache-activemq-5.13.3-bin.tar.gz
2、Windows下根據作業系統位數,執行activemq.bat檔案,啟動ActiveMQ服務。
注意:需要先配置java環境變數,否則啟動一閃而過。
3、訪問http://localhost:8161/admin ,MQ服務正常訪問
二、ActiveMQ點對點訊息模式
編寫ActiveMQ點對點模式客戶端程式。
1、引入activemq-all-5.11.1.jar包
2、訊息生產者
3、訊息消費者/** * 訊息生產者 * @author Max * */ 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;接收確認方式有3種 destination=session.createQueue("FirstQueue1"); // 建立訊息佇列 messageProducer=session.createProducer(destination); // 建立訊息生產者 sendMessage(session, messageProducer); // 傳送訊息 session.commit(); } catch (Exception e) { e.printStackTrace(); } finally{ if(connection!=null){ try { connection.close(); } catch (JMSException e) { 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); } }
Linsner監聽器:/** * 訊息消費者 * @author Max * */ 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"); // 建立連線的訊息佇列;佇列的名字要跟傳送的佇列名對應。 messageConsumer = session.createConsumer(destination); // 建立訊息消費者 while (true) { TextMessage textMessage = (TextMessage) messageConsumer .receive(100000);//1s鍾接收1次訊息 if (textMessage != null) { System.out.println("收到的訊息:" + textMessage.getText()); } else { break; } } } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
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"); // 建立連線的訊息佇列 messageConsumer=session.createConsumer(destination); // 建立訊息消費者 //採用監聽的方式 ,註冊監聽,自動的去監聽裡觸發訊息 messageConsumer.setMessageListener(new Listener()); // 註冊訊息監聽 } catch (JMSException e) { e.printStackTrace(); } } } /** * 訊息監聽 * @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(); } }}