activemq的入門及使用方法

阿新 • • 發佈：2018-12-23

1、什麼是activemq

ActiveMQ 是Apache出品，最流行的、能力強勁的開源訊息匯流排。ActiveMQ 是一個完全支援JMS1.1和J2EE 1.4規範的JMS Provider實現。

主要特點：

1. 多種語言和協議編寫客戶端。

2. 對Spring的支援,ActiveMQ可以很容易內嵌到使用Spring的系統裡面去。

3. 支援多種傳送協議:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA

4. 支援Ajax

5. 支援與Axis的整合

6. 可以很容易得呼叫內嵌JMS provider,進行測試

2、ActiveMQ的訊息形式

對於訊息的傳遞有兩種型別：

一種是點對點的，即一個生產者和一個消費者一一對應，這種方式基於queue佇列

一種是釋出/訂閱模式，即一個生產者產生訊息並進行傳送後，可以由多個消費者進行接收（一對多，這種方式基於topic）

對於點對點的訊息傳輸模型來說，可以在同一個queue上面註冊多個生產者和多個消費者，當訊息的生產者傳送一條訊息時，只有其中的一個訊息消費者接收到訊息生產者傳送的訊息，而不是所有訊息消費者都會接收到該訊息。

對於釋出/訂閱者模式來說，訊息的釋出者需將訊息投遞給topic,而訊息的訂閱者則需要在相應的topic進行註冊，以便接受相應topic的訊息。與點對點傳輸模型不同的是，訊息釋出者的訊息將被自動傳送給所有訂閱了該topic的訊息訂閱者。當訊息訂閱者某段時間由於某種原因斷開了與訊息釋出者的連線時，這個時間段的訊息將會丟失，除非將訊息的訂閱模式設定為持久訂閱

，這時訊息的釋出者將會為訊息的訂閱者保留這段時間所產生的訊息。當訊息的訂閱者重新連線訊息釋出者時，訊息訂閱者仍然可以獲得這部分訊息，而不至於丟失這部分訊息。

JMS定義了五種不同的訊息正文格式，以及呼叫的訊息型別，允許你傳送並接收一些不同形式的資料，提供現有訊息格式的一些級別的相容性。

　　· StreamMessage -- Java原始值的資料流

　　· MapMessage--一套名稱-值對

　　· TextMessage--一個字串物件

　　· ObjectMessage--一個序列化的 Java物件

　　· BytesMessage--一個位元組的資料流

2、如何安裝activemq

第一步：進入http://activemq.apache.org/下載ActiveMQ

第二步：將下載好的安裝包上傳到linux系統中，並解壓（tar -zxvf命令）

第三步：啟動。 ./activema start

使用bin目錄下的activemq命令啟動：
[[email protected] bin]# ./activemq start

關閉：
[[email protected] bin]# ./activemq stop

檢視狀態：
[[email protected] bin]# ./activemq status

第四步：在瀏覽器中輸入http://你的ip:8161/admin，使用者名稱：admin，密碼：admin，可以看到activemq的後臺管理頁面

3、activemq的使用

在maven專案中匯入activemq的依賴包：

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

3.1 點對點（queue）

3.1.1 生產者程式碼

 public void testQueueProducer() throws Exception {
        //第一步：建立connectionFactory, 指定ip和埠
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://你的ip:61616");
        //第二步：通過connectionFactory建立connection
        Connection connection = connectionFactory.createConnection();
        // 第三步：開啟連線，呼叫Connection物件的start方法。
        connection.start();
        // 第四步：使用Connection物件建立一個Session物件。
        //第一個引數：是否開啟事務。true：開啟事務，第二個引數忽略。
        //第二個引數：當第一個引數為false時，才有意義。訊息的應答模式。1、自動應答2、手動應答。一般是自動應答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Queue物件。
        //引數：佇列的名稱。
        Destination destination = session.createQueue("test-queue");
        // 第六步：使用Session物件建立一個Producer物件。
        MessageProducer producer = session.createProducer(destination);
        // 第七步：建立一個Message物件，建立一個TextMessage物件。
        /*TextMessage message = new ActiveMQTextMessage();
        message.setText("hello activemq");*/
        TextMessage message = session.createTextMessage("hello activemq queue");
        // 第八步：使用Producer物件傳送訊息。
        producer.send(destination, message);
        // 第九步：關閉資源。
        producer.close();
        session.close();
        connection.close();
    }

3.1.2 消費者程式碼

public void testQueueConsumer() throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://你的ip:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createQueue("test-queue");
        MessageConsumer consumer = session.createConsumer(destination);

        //為consumer設定一個監聽器
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    TextMessage msg = (TextMessage)message;
                    System.out.println(msg.getText());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });


        //等待鍵盤輸入
        System.in.read();

        //關閉資源
        consumer.close();
        session.close();
        connection.close();
    }

首先執行生產者的程式碼，在後臺管理介面看到已經建立了一個叫test-queue的佇列，並且有一條的訊息入隊。

再執行消費者程式碼：可以看到剛才未傳送的訊息已經出隊了，並且在控制檯接收到了傳送的訊息

3.2 釋出/訂閱模式（topic）

3.2.1 生產者

 public void testTopicProducer() throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://你的ip:61616");
        //第二步：通過connectionFactory建立connection
        Connection connection = connectionFactory.createConnection();
        // 第三步：開啟連線，呼叫Connection物件的start方法。
        connection.start();
        // 第四步：使用Connection物件建立一個Session物件。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session物件建立一個Destination物件
        Destination destination = session.createTopic("test-topic");
        // 第六步：使用Session物件建立一個Producer物件。
        MessageProducer producer = session.createProducer(destination);
        // 第七步：建立一個Message物件，建立一個TextMessage物件。
        TextMessage message = session.createTextMessage("hello activemq topic");
        // 第八步：使用Producer物件傳送訊息。
        producer.send(destination, message);
        // 第九步：關閉資源。
        producer.close();
        session.close();
        connection.close();
    }

3.2.2 消費者(為了便於測試，設定瞭如下的3個消費者)

 public void testTopicConsumer01() throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://你的ip:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createTopic("test-topic");
        MessageConsumer consumer = session.createConsumer(destination);

        //為consumer設定一個監聽器
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    TextMessage msg = (TextMessage)message;
                    System.out.println(msg.getText());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        System.out.println("topic的消費端01。。。。。");

        //等待鍵盤輸入
        System.in.read();

        //關閉資源
        consumer.close();
        session.close();
        connection.close();
    }

由於點對點這種模式，訊息預設是持久化到activemq中的，即使消費者暫時不線上，但是當消費者上線時，還是會收到訊息。而釋出者/訂閱者模式預設則不會持久化，當消費者暫時斷線時，則這段時間的訊息就會丟失。因此我們先將3個消費者執行起來，再執行生產者。

4、activemq與spring整合

生產者配置檔案：

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
       xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
	http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd
	http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd
	http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">

    <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://你的ip:61616" />
    </bean>
    <!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory"
          class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->
        <property name="targetConnectionFactory" ref="targetConnectionFactory" />
    </bean>

    <!-- 配置生產者 -->
    <!-- Spring提供的JMS工具類，它可以進行訊息傳送、接收等 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"></property>
    </bean>

    <!--佇列目的地址-->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg name="name" value="item-change-queue"></constructor-arg>
    </bean>

    <!-- 主題目的地址 -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg name="name" value="item-change-topic"></constructor-arg>
    </bean>
</beans>

業務邏輯：在新增新的商品時，同時生成該商品的索引。

@Override
    public TaotaoResult addItem(TbItem item, String desc, String itemParams) {
        //生成商品id
        long itemId = IDUtils.genItemId();
        item.setId(itemId);
        //設定商品狀態 '商品狀態，1-正常，2-下架，3-刪除',
        item.setStatus((byte) 1);
        //設定商品建立時間
        Date date = new Date();
        item.setCreated(date);
        item.setUpdated(date);

        itemMapper.insert(item);

        //商品描述
        TbItemDesc itemDesc = new TbItemDesc();
        itemDesc.setItemId(itemId);
        itemDesc.setItemDesc(desc);
        itemDesc.setCreated(date);
        itemDesc.setUpdated(date);
        itemDescMapper.insert(itemDesc);

        //新增一個商品資訊後，向activemq的topic中傳送商品id
        jmsTemplate.send(topicDestination, new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage message = session.createTextMessage(itemId + "");
                return message;
            }
        });

        return TaotaoResult.ok();
    }

消費者配置檔案：

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
       xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
	http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd
	http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd
	http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">

    <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://你的ip:61616" />
    </bean>
    <!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory"
          class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->
        <property name="targetConnectionFactory" ref="targetConnectionFactory" />
    </bean>

    <!--佇列目的地址-->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg name="name" value="item-change-queue"></constructor-arg>
    </bean>
    <!-- 主題目的地址 -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg name="name" value="item-change-topic"></constructor-arg>
    </bean>

    <!-- 配置訊息監聽器 -->
    <bean id="messageListener" class="com.taotao.search.listener.ItemChangeMessageListener"></bean>
    <!-- 訊息監聽容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"></property>
        <property name="destination" ref="topicDestination"></property>
        <property name="messageListener" ref="messageListener"></property>
    </bean>
</beans>

監聽器程式碼：

public class ItemChangeMessageListener implements MessageListener {

    @Autowired
    private SearchItemService searchItemService;

    @Override
    public void onMessage(Message message) {
        try {
            TextMessage textMessage = null;
            Long itemId = null;
            if(message instanceof TextMessage) {
                textMessage = (TextMessage)message;
                itemId = Long.parseLong(textMessage.getText());
            }

            //向索引庫新增文件
            searchItemService.addDocument(itemId);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

監聽器監聽到來自新增商品的id後，在資料庫中查詢到該商品的詳細資訊，通過solrJ為該商品新建索引。

 @Override
    public TaotaoResult addDocument(long itemId) throws Exception {
        SearchItem searchItem = searchItemMapper.getItemById(itemId);
        SolrInputDocument document = new SolrInputDocument();
        document.addField("id", searchItem.getId());
        document.addField("item_title", searchItem.getTitle());
        document.addField("item_sell_point", searchItem.getSell_point());
        document.addField("item_price", searchItem.getPrice());
        document.addField("item_image", searchItem.getImage());
        document.addField("item_category_name", searchItem.getCategory_name());
        document.addField("item_desc", searchItem.getItem_desc());
        solrServer.add(document);
        solrServer.commit();
        return TaotaoResult.ok();
    }

activemq的入門及使用方法

1、什麼是activemq

2、ActiveMQ的訊息形式

2、如何安裝activemq

3、activemq的使用

3.1 點對點（queue）

3.1.1 生產者程式碼

3.1.2 消費者程式碼

3.2 釋出/訂閱模式（topic）

3.2.1 生產者

3.2.2 消費者(為了便於測試，設定瞭如下的3個消費者)

4、activemq與spring整合

activemq的入門及使用方法

activemq安裝及自啟動方法

javascript入門及數據類型

ObjectQuery查詢及方法

elasticsearch 官方入門及 API

列表的定義及方法

java 消息機制 ActiveMQ入門實例

轉載---編寫高質量代碼:改善Java程序的151個建議(第3章:類、對象及方法___建議47~51)

轉載---編寫高質量代碼:改善Java程序的151個建議(第3章:類、對象及方法___建議41~46)

solr搜索之入門及原理（一）

Linux入門及基礎知識

Linux入門及命令詳解

嘮一嘮Linux系統入門的方法和經驗

Laravel入門及實踐，快速上手社交系統ThinkSNS+二次開發

ActiveMQ 入門helloworld

Docker基礎入門及示例

manjaro 入門及基本配置安裝

VUE -- stylus入門使用方法

Redis快速入門及應用

Ext.grid.GridPanel屬性及方法等

activemq的入門及使用方法

1、什麼是activemq

2、ActiveMQ的訊息形式

2、如何安裝activemq

3、activemq的使用

3.1 點對點（queue）

3.1.1 生產者程式碼

3.1.2 消費者程式碼

3.2 釋出/訂閱模式（topic）

3.2.1 生產者

3.2.2 消費者(為了便於測試，設定瞭如下的3個消費者)

4、activemq與spring整合

相關推薦