ActiveMQ學習之java代碼調用ActiveMQ隊列

時間 2020-01-24

標籤 activemq 學習 java 代碼調用隊列欄目 ActiveMQ 简体版

原文原文鏈接

1、基本概念：

ActiveMQ中共分爲兩種：queue和topicjava

queue：在點對點消息傳遞域中，目的地被稱爲隊列（一對一）spring

topic：在發佈訂閱消息中，目的地被稱爲主題（一對多）apache

2、建立maven工程，並引入依賴，這裏我建立的springboot項目，因此引入的依賴以下：

依賴：

<!--activemq-->
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-all</artifactId>
      <version>5.15.9</version>
    </dependency>

建立main方法測試鏈接，消息提供者。

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @ProjectName: springbootActiveMQ
 * @Package: cn.**.test
 * @Author: huat
 * @Date: 2020/1/2 17:04
 * @Version: 1.0
 */
public class ActiveMQTest {
    //url路徑
    private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
    //隊列名稱
    private static final String QUEUE_NAME="queue01";
  

    public static void main(String[] args) {
        //一、建立鏈接工廠
        //若是帳號密碼沒有修改的話，帳號密碼默認均爲admin
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
        //若是帳號密碼修改的話
        //第一個參數爲帳號，第二個爲密碼，第三個爲請求的url
        //ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
        try {
            //二、經過鏈接工廠獲取鏈接
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //三、建立session會話
            //裏面會有兩個參數，第一個爲事物，第二個是簽收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //四、建立目的地（具體是隊列仍是主題）,這裏是建立隊列
            Queue queue=session.createQueue(QUEUE_NAME);
           
            //五、建立消息生產者
            MessageProducer messageProducer = session.createProducer(queue);
            //六、經過messageProducer生產三條消息發送到MQ消息隊列中
            for (int i=0;i<3;i++){
                //七、建立消息
                TextMessage textMessage = session.createTextMessage("msg----->" + i);//建立一個文本消息
                //八、經過messageProducer發送給mq
                messageProducer.send(textMessage);
            }
            messageProducer.close();
            session.close();
            connection.close();
            System.out.println("消息發送成功");
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

查看管理頁面是否有消息

建立main方法測試鏈接，消息消費者。（同步調用）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @ProjectName: springbootActiveMQ
 * @Package: cn.**.test
 * @Author: huat
 * @Date: 2020/1/3 8:47
 * @Version: 1.0
 */
public class ActiveMQConsumer {
    //url路徑
    private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
    //隊列名稱
    private static final String QUEUE_NAME="queue01";

    public static void main(String[] args) {
        //一、建立鏈接工廠
        //若是帳號密碼沒有修改的話，帳號密碼默認均爲admin
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
        //若是帳號密碼修改的話
        //第一個參數爲帳號，第二個爲密碼，第三個爲請求的url
        //ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
        try {
            //二、經過鏈接工廠獲取鏈接
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //三、建立session會話
            //裏面會有兩個參數，第一個爲事物，第二個是簽收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //四、這裏接受的queue的名稱要和發送者的一致
            Queue queue = session.createQueue(QUEUE_NAME);
            //五、建立消費者
            MessageConsumer consumer = session.createConsumer(queue);
            //六、獲取消息
            while(true){
                //MessageConsumer 調用的receive方法爲同步調用，在消息到達以前一直阻塞線程
                //用什麼格式發送，這裏就用什麼格式接受
                //receive等待消息，不限制時間
                    TextMessage message=(TextMessage)consumer.receive();
                //receive帶參數等待消息，限制時間，單位毫秒
                //TextMessage message=(TextMessage)consumer.receive(4000L);

                if(null != message){
                    System.out.println("接受的消息爲------>"+message.getText());
                }else{
                    break;
                }
            }
            //七、閉資源
            consumer.close();
            session.close();
            connection.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

建立main方法測試鏈接，消息消費者。（異步調用）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @ProjectName: springbootActiveMQ
 * @Package: cn.**.test
 * @Author: huat
 * @Date: 2020/1/3 8:47
 * @Version: 1.0
 */
public class ActiveMQConsumer {
    //url路徑
    private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
    //隊列名稱
    private static final String QUEUE_NAME="queue01";

    public static void main(String[] args) {
        //一、建立鏈接工廠
        //若是帳號密碼沒有修改的話，帳號密碼默認均爲admin
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
        //若是帳號密碼修改的話
        //第一個參數爲帳號，第二個爲密碼，第三個爲請求的url
        //ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
        try {
            //二、經過鏈接工廠獲取鏈接
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //三、建立session會話
            //裏面會有兩個參數，第一個爲事物，第二個是簽收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //四、這裏接受的queue的名稱要和發送者的一致
            Queue queue = session.createQueue(QUEUE_NAME);
            //五、建立消費者
            MessageConsumer consumer = session.createConsumer(queue);
            //六、經過監聽的方式消費消息
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    //若是message不等於null而且屬於TextMessage類型（由於消息發送的類型是TextMessage，因此這裏判斷是不是這個類型）
                    if(null!=message&&message instanceof TextMessage){
                        TextMessage textMessage=(TextMessage)message;
                        try {
                            System.out.println(textMessage.getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            //七、保證控制檯一直在運行
            System.in.read();
            //八、閉資源
            consumer.close();
            session.close();
            connection.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}