ActiveMQ

下載下傳位址

ActiveMQ官網：

http://activemq.apache.org/

ActiveMQ的所有版本下載下傳位址：

http://activemq.apache.org/download-archives.html

ActiveMQ原理

首先來看本地通訊的情況，應用程式A和應用程式B運作于同一系統A，它們之間可以借助消息隊列技術進行彼此的通訊：應用程式A向隊列1發送一條資訊，而當應用程式B需要時就可以得到該資訊。

其次是遠端通訊的情況，如果資訊傳輸的目标改為在系統B上的應用程式C，這種變化不會對應用程式A産生影響，應用程式A向隊列2發送一條資訊，系統A的MQ發現Q2所指向的目的隊列實際上位于系統B，它将資訊放到本地的一個特殊隊列－傳輸隊列(Transmission Queue)。我們建立一條從系統A到系統B的消息通道，消息通道代理将從傳輸隊列中讀取消息，并傳遞這條資訊到系統B，然後等待确認。隻有MQ接到系統B成功收到資訊的确認之後，它才從傳輸隊列中真正将該資訊删除。如果通訊線路不通，或系統B不在運作，資訊會留在傳輸隊列中，直到被成功地傳送到目的地。這是MQ最基本而最重要的技術–確定資訊傳輸，并且是一次且僅一次(once-and-only-once)的傳遞。

MQ提供了用于應用內建的松耦合的連接配接方法，因為共享資訊的應用不需要知道彼此實體位置（網絡位址）；不需要知道彼此間怎樣建立通信；不需要同時處于運作狀态；不需要在同樣的作業系統或網絡環境下運作。

JMS架構基本角色和程式設計子產品

JMS Message Producers : 消息生産者，向消息隊列提供消息
JMS Provider(Broker) : JMS提供商，如ActiveMQ，提供JMS Queues/Topics服務
JMS Message listener/Consumer ：接收并使用消息

queue的使用

Product

開發步驟：

建立ActiveMQConnectionFactory，建立工廠需要ip和端口号，指定的是tcp協定
從工廠中擷取連接配接
使用連接配接的start方法，開啟連接配接
使用連接配接擷取session對象
使用session建立Destination，就是建立queue，（可以設定queue（點對點）和topic（訂閱）
使用session建立Product，消息的是生産者
建立消息對象TextMessage，設定消息
發送消息
釋放資源，關閉Product、session、connection

public static void main(String[] args) throws Exception {
  // 1. 建立連接配接工廠ActiveMQConnectionFactory，需要ip和端口61616
  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.37.130:61616");
  // 2. 從連接配接工廠中建立連接配接對象
  Connection connection = factory.createConnection();
  // 3. 執行start方法開啟連接配接
  connection.start();
  // 4. 從連接配接中建立session對象
  // 第一個參數，是否開啟事務，JTA分布式事務
  // 第二個參數，是否自動應答，如果第一個參數為true，第二個參數失效
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 從session中建立Destination對象，設定queue名稱（有兩種類型queue和topic）
  Queue queue = session.createQueue("test-queue");
  // 6. 從session中建立Product對象
  MessageProducer producer = session.createProducer(queue);
  // 7. 建立消息對象
  TextMessage textMessage = new ActiveMQTextMessage();
  // 設定消息内容
  textMessage.setText("開始發消息！");
  // 8. 發送消息
  producer.send(textMessage);
  // 9. 關閉session、連接配接
  producer.close();
  session.close();
  connection.close();
    }

Consumer

開發步驟

建立連接配接工廠ActiveMQConnectionFactory，需要ip和端口
從連接配接工廠中建立連接配接對象
執行start方法開始連接配接
從連接配接中建立session對象
從session中建立Destination對象，設定queue名稱（有兩種類型queue和topic）
從session中建立Consumer對象
接收消息
列印結果
關閉session、連接配接

public static void main(String[] args) throws Exception {
  // 1. 建立連接配接工廠ActiveMQConnectionFactory，需要ip和端口61616
  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.37.130:61616");
  // 2. 使用工廠建立連接配接
  Connection connection = factory.createConnection();
  // 3. 使用start開啟連接配接
  connection.start();
  // 4. 從連接配接中建立session對象
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 從session中建立Destination對象，設定queue名字
  Queue queue = session.createQueue("test-queue");
  // 6. 從session中建立Consumer
  MessageConsumer consumer = session.createConsumer(queue);
  // 7， 接收消息,直接擷取
  while (true) {
    // 消息逾時時間是20秒
    Message message = consumer.receive(20000);
    // 如果消息為空，則跳出死循環
    if (message == null) {
    break;
    }
    // 8. 列印消息
    if (message instanceof TextMessage) {
    // 擷取消息
    TextMessage textMessage = (TextMessage) message;
    // 列印
    System.out.println(textMessage.getText());      }
  }
  // 9. 關閉session、連接配接等
  consumer.close();
  session.close();
  connection.close();
    }

使用監聽器

public static void main(String[] args) throws Exception {
  // 1. 建立連接配接工廠ActiveMQConnectionFactory，需要ip和端口61616
  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.37.130:61616");
  // 2. 使用工廠建立連接配接
  Connection connection = factory.createConnection();
  // 3. 使用start開啟連接配接
  connection.start();
  // 4. 從連接配接中建立session對象
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 從session中建立Destination對象，設定queue名字
  Queue queue = session.createQueue("test-queue");
  // 6. 從session中建立Consumer
  MessageConsumer consumer = session.createConsumer(queue);
  // 7.接收消息
  // 監聽器的方式實際上是開啟了一個新的線程，專門處理消息的接受
  // 現在的情況是，主線程執行完就結束了，新的線程也跟着沒了
  consumer.setMessageListener(new MessageListener() {
    @Override
    public void onMessage(Message message) {
    if (message instanceof TextMessage) {
        // 擷取消息
        TextMessage textMessage = (TextMessage) message;
        try {
      // 列印
      System.out.println(textMessage.getText());
        } catch (JMSException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
        }
    }
    }
  });
  // 讓主線程等待一會，監聽器能夠有時間執行
  Thread.sleep(10000);
  // 9. 關閉session、連接配接等
  consumer.close();
  session.close();
  connection.close();
    }

topic的使用

public static void main(String[] args) throws Exception {
  // 1. 建立連接配接工廠ActiveMQConnectionFactory
  ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
    "tcp://192.168.37.130:61616");
  // 2. 使用工廠建立連接配接
  Connection connection = activeMQConnectionFactory.createConnection();
  // 3. 使用start方法開啟連接配接
  connection.start();
  // 4. 從連接配接建立session
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 從session建立Destination對象，設定topic名稱
  Topic topic = session.createTopic("test-topic");
  // 6. 從session建立Product
  MessageProducer producer = session.createProducer(topic);
  // 7. 建立消息對象
  TextMessage textMessage = new ActiveMQTextMessage();
  textMessage.setText("topic消息");
  // 8. 發送消息
  producer.send(textMessage);
  // 9. 關閉session、連接配接等
  producer.close();
  session.close();
  connection.close();
    }
}

Consumer1

public static void main(String[] args) throws Exception {
  // 1. 建立連接配接工廠ActiveMQConnectionFactory
  ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
    "tcp://192.168.37.130:61616");
  // 2. 從連接配接工廠建立連接配接
  Connection connection = activeMQConnectionFactory.createConnection();
  // 3. 使用start方法開啟連接配接
  connection.start();
  // 4. 從連接配接建立session對象
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 從session建立安Destination，設定topic名稱
  Topic topic = session.createTopic("test-topic");
  // 6. 從session建立Consumer對象
  MessageConsumer consumer = session.createConsumer(topic);
  // 7. 接收消息，直接接受
  while (true) {
    Message message = consumer.receive(20000);
    if (message == null) {
    break;
    }
    if (message instanceof TextMessage) {
    TextMessage textMessage = (TextMessage) message;
    // 8. 列印消息
    System.out.println(textMessage.getText());
    }
  }
  // 9. 關閉session、消息等
  consumer.close();
  session.close();
  connection.close();
    }

Consumer2

public static void main(String[] args) throws Exception {
  // 1. 建立連接配接工廠ActiveMQConnectionFactory
  ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
    "tcp://192.168.37.130:61616");
  // 2. 從連接配接工廠建立連接配接
  Connection connection = activeMQConnectionFactory.createConnection();
  // 3. 使用start方法開啟連接配接
  connection.start();
  // 4. 從連接配接建立session對象
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 從session建立安Destination，設定topic名稱
  Topic topic = session.createTopic("test-topic");
  // 6. 從session建立Consumer對象
  MessageConsumer consumer = session.createConsumer(topic);
  // 7. 接收消息，直接接受
  // while (true) {
  // Message message = consumer.receive(20000);
  //
  // if (message == null) {
  // break;
  // }
  //
  // if (message instanceof TextMessage) {
  // TextMessage textMessage = (TextMessage) message;
  // // 8. 列印消息
  // System.out.println(textMessage.getText());
  // }
  // }
  // 7.接受消息，使用監聽器
  consumer.setMessageListener(new MessageListener() {
    @Override
    public void onMessage(Message message) {
    if (message instanceof TextMessage) {
        TextMessage textMessage = (TextMessage) message;
        try {
      // 列印消息
      System.out.println(textMessage.getText());
        } catch (JMSException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
        }
    }
    }
  });
  // 等待監聽器執行
  Thread.sleep(10000);
  // 9. 關閉session、消息等
  consumer.close();
  session.close();
  connection.close();
    }

ActiveMQ整合Spring

加入依賴

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
  <groupId>com.taotao</groupId>
  <artifactId>taotao-parent</artifactId>
  <version>0.0.1-SNAPSHOT</version>
    </parent>
    <groupId>cn.itcast.activemq</groupId>
    <artifactId>itcast-activemq</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <dependencies>
  <!-- 加入ActiveMQ依賴 -->
  <dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
  </dependency>
  <!-- 加入spring-jms依賴 -->
  <dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
  </dependency>
  <!-- Spring -->
  <dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-webmvc</artifactId>
  </dependency>
    </dependencies>
</project>

消息發送

建立spring容器
從容器中擷取JMSTemplate對象，發送消息
定義Destination
使用JMSTemplate對象發送消息

public class Product {
    public static void main(String[] args) {
  // 1. 建立spring容器
  ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
    "applicationContext-activemq.xml");
  // 2. 從容器中擷取JMSTemplate對象
  JmsTemplate jmsTemplate = context.getBean(JmsTemplate.class);
  // 3. 從容器中擷取Destination對象
  Destination destination = context.getBean(Destination.class);
  // 4. 使用JMSTemplate發送消息
  jmsTemplate.send(destination, new MessageCreator() {
    @Override
    public Message createMessage(Session session) throws JMSException {
    // 建立消息對象
    TextMessage textMessage = new ActiveMQTextMessage();
    // 設定消息内容
    textMessage.setText("spring整合ActiveMQ");
    // 列印消息
    System.out.println(textMessage.getText());
    return textMessage;
    }
  });
    }
}

消息接受

建立一個類實作MessageListener 接口。業務處理在此類中實作。
在spring容器中配置DefaultMessageListenerContainer對象，引用MessageListener 實作類對象接收消息。

public class MyMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
  if (message instanceof TextMessage) {
    TextMessage textMessage = (TextMessage) message;
    try {
    // 擷取消息内容
    String msg = textMessage.getText();
    // 列印消息
    System.out.println("接受消息:" + msg);
    } catch (JMSException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    }
  }
    }
}

queue方式配置Spring

<?xml version="1.0" encoding="UTF-8"?>
<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:jms="http://www.springframework.org/schema/jms"
    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.0.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
    http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
    http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.0.xsd">
    <!-- 真正可以産生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL" value="tcp://192.168.37.130: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">
  <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 -->
  <property name="connectionFactory" ref="connectionFactory" />
    </bean>
    <!--這個是隊列目的地，點對點的 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
  <constructor-arg>
    <value>queue</value>
  </constructor-arg>
    </bean>
    <!--這個是主題目的地，一對多的 -->
    <!-- <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> -->
    <!-- <constructor-arg value="topic" /> -->
    <!-- </bean> -->
    <!-- messageListener實作類 -->
    <bean id="myMessageListener" class="top.simba.activemq.spring.MyMessageListener"></bean>
    <!-- 配置一個jsm監聽容器 -->
    <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="queueDestination" />
  <property name="messageListener" ref="myMessageListener" />
    </bean>
</beans>

topic方式配置Spring

<?xml version="1.0" encoding="UTF-8"?>
<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:jms="http://www.springframework.org/schema/jms" 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.0.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
    http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
    http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.0.xsd">
    <!-- 真正可以産生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL" value="tcp://192.168.37.130: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">
  <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 -->
  <property name="connectionFactory" ref="connectionFactory" />
    </bean>
    <!--這個是隊列目的地，點對點的 -->
    <!-- <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> -->
    <!-- <constructor-arg> -->
    <!-- <value>queue</value> -->
    <!-- </constructor-arg> -->
    <!-- </bean> -->
    <!--這個是主題目的地，一對多的 -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
  <constructor-arg value="topic" />
    </bean>
    <!-- messageListener實作類 -->
    <bean id="myMessageListener" class="top.simba.activemq.spring.MyMessageListener"></bean>
    <!-- messageListener實作類 -->
    <bean id="myMessageListener2" class="top.simba.activemq.spring.MyMessageListener2"></bean>
    <!-- 配置一個jsm監聽容器 -->
    <bean id="jmsContainer"
  class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="topicDestination" />
  <property name="messageListener" ref="myMessageListener" />
    </bean>
    <!-- 配置一個jsm監聽容器 -->
    <bean id="jmsContainer2"
  class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="topicDestination" />
  <property name="messageListener" ref="myMessageListener2" />
    </bean>
</beans>

ActiveMQ安裝

1.上傳ActiveMQ，并解壓

2.進入解壓後的檔案夾的bin目錄，啟動ActiveMQ

[root@root bin]# ./activemq start

3.通路

http://192.168.37.130:8161/admin/

原始賬号：admin

原始密碼：admin

ActiveMQ

ActiveMQ

下載下傳位址

ActiveMQ原理

JMS架構基本角色和程式設計子產品

queue的使用

Product

Consumer

使用監聽器

topic的使用

Consumer1

Consumer2

ActiveMQ整合Spring

消息發送

消息接受

queue方式配置Spring

topic方式配置Spring

ActiveMQ安裝

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method