~~~ import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class TextConsumer { public String receiveTextMessage(){ String resultCode = ""; ConnectionFactory factory = null; Connection connection = null; Session session = null; Destination destination = null; // 消息的消費者，用于接受消息的對象 MessageConsumer consumer = null; Message message = null; try { factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://114.116.69.58:61616"); connection = factory.createConnection(); //消費者必須啟動連接，否則無法處理消息 connection.start(); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); destination = session.createQueue("first-mq"); //創建消息消費者對象，在指定的目的地中獲取消息 consumer = session.createConsumer(destination); //獲取隊列中的消息 message = consumer.receive(); //處理文本消息 resultCode = ((TextMessage)message).getText(); //System.in.read(); }catch (Exception e){ e.printStackTrace(); }finally { if (consumer != null){ try { consumer.close(); } catch (JMSException e) { e.printStackTrace(); } } if (session != null){ try { session.close(); } catch (JMSException e) { e.printStackTrace(); } } if (connection != null){ try { connection.close(); } catch (JMSException e) { e.printStackTrace(); } } } return resultCode; } public static void main(String[] args) { TextConsumer textConsumer = new TextConsumer(); String s = textConsumer.receiveTextMessage(); System.out.println(s); } } ~~~ ~~~ consumer = session.createConsumer(destination); //注冊監聽器，注冊成功后，隊列中的消息變化會自動觸發監聽器代碼，接收消息并處理 consumer.setMessageListener(new MessageListener() { /** * 監聽器一旦注冊，永久有效，consumer線程不關閉 * 只要有消息未處理，自動調用onMessage方法處理消息 * 監聽器可以注冊多個，注冊多個監聽器，相當于集群 * ActiveMQ自動的循環調用多個監聽器，處理隊列中的消息，實現并行處理 */ @Override public void onMessage(Message message) { try { // ObjectMessage om = (ObjectMessage)message; // Object data = om.getObject(); // System.out.println(data); } catch (Exception e) { e.printStackTrace(); } } }); System.in.read(); ~~~