ActiveMQ使用分為兩大塊：生產者和消費者 一、準備 項目導入jar包：activemq-all-5.15.3.jar 并buildpath? 二、生產者

創建連接工廠

ActiveMQConnectionFactory mqf = new ActiveMQConnectionFactory(userName, password, brokerURL); 注： userName是ActiveMQ的用戶名，默認可以通過：ActiveMQConnection.DEFAULT_USER password是ActiveMQ的密碼，默認可以通過： ActiveMQConnection.DEFAULT_PASSWORD brokerURL是ActiveMQ的連接，指定格式為：tcp://主機名:61616

獲取連接

connection = mqf.createConnection();

生成會話

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

生成對應的topic

Destination destination = session.createTopic("mytopic");

創建生產者

MessageProducer producer = session.createProducer(destination);

設置發送消息使用的模式

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); 默認是：DeliveryMode.PERSISTENT

生成消息

TextMessage msg = session.createTextMessage(“message");

啟動連接

connection.start();

發送消息

producer.send(msg);

關閉生產者

producer.close();

關閉會話

session.close();

關閉連接

connection.close(); 三、消費者

繼承接口

MessageListener ExceptionListener 并實現onException(JMSException exception)和onMessage(Message message)方法

創建連接工廠

ActiveMQConnectionFactory mqf = new ActiveMQConnectionFactory(userName, password, brokerURL); 具體參數同上

獲取連接

Connection connection = mqf.createConnection();

生成會話

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

生成對應的topic

Destination destination = session.createTopic("mytopic”);

創建消費者

MessageConsumer consumer = session.createConsumer(destination);

啟動連接

connection.start();

設置消息監聽

consumer.setMessageListener(this);

設置異常監聽

connection.setExceptionListener(this);

實現onMessage方法

改方法有一個參數Message message,這個參數是從ActiveMQ上拿到的消息，可以通過如下方法解析出來： TextMessage tm = (TextMessage)message; String result = tm.getText();

關閉消費者

consumer.close();

關閉會話

session.close();

關閉連接

connection.close(); 四、例程

生產者實現程序

1 package activemq_test; 2 3 import javax.jms.Connection; 4 import javax.jms.DeliveryMode; 5 import javax.jms.Destination; 6 import javax.jms.JMSException; 7 import javax.jms.MessageProducer; 8 import javax.jms.Session; 9 import javax.jms.TextMessage; 10 11 import org.apache.activemq.ActiveMQConnection; 12 import org.apache.activemq.ActiveMQConnectionFactory; 13 14 public class Producer_tool { 15 16 private final static String userName = ActiveMQConnection.DEFAULT_USER; 17 private final static String password = ActiveMQConnection.DEFAULT_PASSWORD; 18 private final static String brokerURL = "tcp://192.168.0.5:61616"; 19 private MessageProducer producer = null; 20 private Connection connection = null; 21 private Session session = null; 22 23 public void initialize() throws JMSException { 24 ActiveMQConnectionFactory mqf = new ActiveMQConnectionFactory(userName, password, brokerURL); 25 connection = mqf.createConnection(); 26 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 27 Destination destination = session.createTopic("mytopic"); 28 producer = session.createProducer(destination); 29 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); 30 } 31 32 public void send(String message) throws JMSException { 33 initialize(); 34 TextMessage msg = session.createTextMessage(message); 35 System.out.println("sending message: " + message); 36 connection.start(); 37 producer.send(msg); 38 } 39 40 public void close() throws JMSException { 41 if(producer != null) { 42 producer.close(); 43 } 44 if(session != null) { 45 session.close(); 46 } 47 if(connection != null) { 48 connection.close(); 49 } 50 System.out.println("closed"); 51 } 52 53 }

?

生產者主程序

1 package activemq_test; 2 import javax.jms.JMSException; 3 public class Producer_test { 4 public static void main(String[] args) throws JMSException { 5 Producer_tool producer = null; 6 for(int i = 0; i < 10; i++) { 7 producer = new Producer_tool(); 8 producer.send("message" + i); 9 producer.close(); 10 } 11 } 12 }

消費者實現程序

1 package activemq_test; 2 3 import javax.jms.Connection; 4 import javax.jms.Destination; 5 import javax.jms.ExceptionListener; 6 import javax.jms.JMSException; 7 import javax.jms.Message; 8 import javax.jms.MessageConsumer; 9 import javax.jms.MessageListener; 10 import javax.jms.Session; 11 import javax.jms.TextMessage; 12 13 import org.apache.activemq.ActiveMQConnection; 14 import org.apache.activemq.ActiveMQConnectionFactory; 15 16 public class Consumer_tool implements MessageListener,ExceptionListener{ 17 18 private final static String userName = ActiveMQConnection.DEFAULT_USER; 19 private final static String password = ActiveMQConnection.DEFAULT_PASSWORD; 20 private final static String brokerURL = "tcp://192.168.0.5:61616"; 21 private Connection connection = null; 22 private Session session = null; 23 private MessageConsumer consumer = null; 24 static boolean isConnection = false; 25 26 public void initialize() throws JMSException { 27 ActiveMQConnectionFactory mqf = new ActiveMQConnectionFactory(userName, password, brokerURL); 28 connection = mqf.createConnection(); 29 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 30 Destination destination = session.createTopic("mytopic"); 31 consumer = session.createConsumer(destination); 32 } 33 34 public void consumeMessage() throws JMSException { 35 initialize(); 36 connection.start(); 37 consumer.setMessageListener(this); 38 connection.setExceptionListener(this); 39 isConnection = true; 40 System.out.println("consumer is listening"); 41 42 } 43 44 @Override 45 public void onException(JMSException exception) { 46 isConnection = false; 47 } 48 49 @Override 50 public void onMessage(Message message) { 51 if(message instanceof TextMessage) { 52 TextMessage tm = (TextMessage)message; 53 try { 54 System.out.println("consumer received " + tm.getText()); 55 } catch (JMSException e) { 56 e.printStackTrace(); 57 } 58 } 59 else { 60 System.out.println(message); 61 } 62 } 63 64 public void close() throws JMSException { 65 if(consumer != null) { 66 consumer.close(); 67 } 68 if(session != null) { 69 session.close(); 70 } 71 if(connection != null) { 72 connection.close(); 73 } 74 System.out.println("consumer has closed"); 75 } 76 }

消費者主程序

1 package activemq_test; 2 import javax.jms.JMSException; 3 public class Consumer_test { 4 public static void main(String[] args) throws JMSException { 5 Consumer_tool consumer = new Consumer_tool(); 6 consumer.consumeMessage(); 7 while(Consumer_tool.isConnection) { 8 9 } 10 consumer.close(); 11 } 12 }

轉載于:https://www.cnblogs.com/xiatianyu/p/9055647.html

總結

以上是生活随笔為你收集整理的ActiveMQ的使用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。