ActiveMQ——1.初识ActiveMQ

梦里梦外; 2021-09-26 09:20 500阅读 0赞

# 前言 #

## 中间件 ##

由于业务的不同、技术的发展、硬件和软件的选择有所差别，导致了异构组件或应用并存的局面。要使这些异构的组件协同工作，一个有效的方式就是提供一个允许它们进行通信的层，该层即为中间件。

在概念上，中间件位于应用程序层与平台层（操作系统和底层网络服务）之间：

![Center][]

分类：

*  基于远程过程调用(Remote Procedure Call, RPC)的中间件  
    允许一个应用程序中的过程调用远程应用程序中的过程，就好像它们是本地调用一样。
 *  基于对象请求代理(Object Request Broker, ORB)的中间件
 *  面向消息（Message-Oriented Middleware，MOM）的中间件  
    使分布式应用程序可以通过发送和接收消息来进行通信和交换数据。

## **MOM** ##

全称：Message-Oriented Middleware 中文：面向消息的中间件。

MOM通信模型：

![Center 1][]

MOM模型中的四个基本元素：消息传递提供者、目的地、客户端（发送方或接收方）、消息。  
发送方客户端将消息发送到消息提供者维护的目的地，这些消息一直被保留直到接收方客户端将消息取出。

## **JMS** ##

全称：Java Message Service 中文：Java消息服务。

JMS是Java的一套API标准，最初的目的是为了使应用程序能够访问现有的MOM系统（客户端通过JMS访问消息提供者）；后来被许多现有的MOM供应商采用，并实现为MOM系统。  
基于JMS实现的MOM，又被称为JMS Provider。

## ActiveMQ ##

Apache出品，完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。

它有很多功能、特色，所以它的详细信息很复杂。但是基于这一刻，我们只需要知道两件事：(1) 它是MOM；(2) 它是JMS的一个实现。

## 相互关系 ##

![Center 2][]

参考：[点击打开链接][Link 1]

# **正文** #

## [JMS客户端编程模型][JMS] ##

### **消息传送模式** ###

消息传送模式 又称为消息传送域，JMS API定义了两种模式：PTP和Pub/Sub。

## PTP ##

全称：Point-to-Point 中文：点对点

![Center 3][]

上图描述了这样的内容： Sender发送 Message到 代理维护的 Queue，然后 Receiver可以从这个 Queue中获取这个 Message。

这个模式的特点是：

1.  一个Message只能交给一个Receiver；这里的Message像是一个消耗品。
2.  只要Queue中有Message，Receiver就能获取到。

## Pub/Sub ##

全称：Publish/Subscribe 中文：发布/订阅

![Center 4][]

图描述了这样的内容： Publisher发布 Message到 代理维护的 Topic， Subscriber从 代理那里订阅的 Topic，从而可以获取对应的 Message。

这个模式的特点是：

1.  一个Message可以复制多份，交给多个Subscriber。
2.  先订阅，再获取 - Subscriber只能获取订阅之后发送到Topic的Message。

##     ##

## 角色定位 ##

MOM包含四个基本元素：消息传递提供者、目的地、客户端（发送方或接收方）、消息。上述两种模式在MOM中的角色定位，可以表现为：

![Center 5][]

## JMS API ##

JavaEE提供的javax.jms包中，涵盖了JMS的API。

### 统一域和特定于域的API ###

早期的规范中，PTP和Pub/Sub各有不同的接口体系，后来的JMS1.1在两个接口体系的上层，又定义了一层统一的接口体系，称为统一域的API；而之前的两套接口体系，称为特定于域的API。对比情况如下：

![Center 6][]

注意： JMS当然不是只提供了上述的接口。上述接口，只是在统一之后，有对比意义的接口。

## 编程模型 ##

![Center 7][]

说明：连接工厂、目的地通常作为受管理对象来创建、配置、管理，驻留在对象存储库中；客户端程序通过JNDI查找获取对象，而不建议显示的创建。所以连接工厂、目的地在上图的表示有所不同。

## 对象简介 ##

*  连接工厂（ConnectionFactory)  
    客户端使用连接工厂对象(ConnectionFactory)创建连接。
 *  连接（Connection）  
    连接对象 (Connection) 表示客户端与代理之间的活动连接。创建连接时会分配通信资源并对客户端进行验证。这是一个相当重要的对象，大多数客户端均使用一个连接来完成所有的消息传送。连接支持并发使用：一个连接可由任意数量的生成方和使用方共享。
 *  会话（Session）  
    如果连接代表客户端与代理之间的通信渠道，则会话标记客户端与代理之间的单次对话。会话对象主要用于创建消息、消息生成方和消息使用方。
 *  消息（Message）  
    消息封装一次通信的内容。消息由三部分组成：消息头、属性和主体。
 *  消息生成方（MessageProducer）  
    由Session创建，负责发送Message到目的地。
 *  消息使用方（MessageConsumer）  
    由Session创建，负责从目的地中消费Message。
 *  目的地（Destination）  
    JMS Provider负责维护，用于对Message进行管理的对象。MessageProducer需要指定Destination才能发送消息，MessageReceiver需要指定Destination才能接收消息。

## demo ##

--------------------

将上文的客户端编程模型应用起来。两个客户端：producer-client负责发送消息到ActiveMQ；consumer-client负责从ActiveMQ中获取消息。

在此之前，你需要下载ActiveMQ，并确保成功启动。

下载：不同的版本对JDK版本的要求不同，当前最新稳定版本是5.13.2，要求JDK1.7以上。

安装：直接解压缩

启动：(1) 在cmd窗口，切换到解压缩的路径；(2) bin\\activemq start \[1\]

确认：ActiveMQ提供了监控器，访问http://localhost:8161/admin，默认账号：admin/admin。可以访问，表示启动成功。

补充：

\[1\] 在启动的时候可能会遇到下面的错误

> Error occurred during initialization of VM  
> Could not reserve enough space for object heap  
> Error: Could not create the Java Virtual Machine.  
> Error: A fatal exception has occurred. Program will exit.

修改bin\\activemq.bat文件：

if "%ACTIVEMQ_OPTS%" == "" set ACTIVEMQ_OPTS=-Xms1G -Xmx1G

将Xms和Xmx适当调小，如256M。

## 1.producer-client ##

基于Maven的simple project。

### 文件目录结构 ###

pom.xml
    src/main/resources/
        |---- jndi.properties
    src/main/java/
        |---- cn.sinobest.asj.producer.jms.clientmode
                  |---- SimpleProducer.java # 基于客户端编程模型，发送消息给ActiveMQ

### 文件内容 ###

#### 1.pom.xml ####

<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>
        <groupId>cn.sinobest.asj</groupId>
        <artifactId>jms-producer</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>jms-producer</name>
        <description>基于ActiveMQ的Producer Client。</description>
        <dependencies>
            
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-client</artifactId>
                <version>5.13.2</version>
            </dependency>
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-simple</artifactId>
                <version>1.7.19</version>
            </dependency>
        </dependencies>
    </project>

#### 2.jndi.properties ####

java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory
    
    # use the following property to configure the default connector
    java.naming.provider.url=tcp://localhost:61616
    
    # register some queues in JNDI using the form
    # queue.[jndiName] = [physicalName]
    queue.exampleQueue=example.queue
    
    # register some topics in JNDI using the form
    # topic.[jndiName] = [physicalName]
    topic.exampleTopic = example.topic

说明：

*  java.naming.factory.initial  
    定义ConnectionFactory的实例化类。
 *  java.naming.provider.url  
    定义broker的url，这个是ActiveMQ默认的url。
 *  queue.exampleQueue  
    定义了一个Queue Destination。name为example.queue，JNDI的name为exampleQueue。
 *  topic.exampleTopic  
    定义了一个Topic Destination。name为example.topic，JNDI的name为exampleTopic。

#### 3.SimpleProcedure.java ####

package cn.sinobest.asj.producer.jms.clientmode;
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.naming.Context;
    import javax.naming.InitialContext;
    import javax.naming.NamingException;
    /**
     * A simple demo for producer client to send message to ActiveMQ.<br>
     * refer to <a href="http://activemq.apache.org/jndi-support.html">JNDI
     * Support</a>
     * 
     * @author lijinlong
     */
    public class SimpleProducer {
        /** JNDI name for ConnectionFactory */
        static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
        /** JNDI name for Queue Destination (use for PTP Mode) */
        static final String QUEUE_JNDI_NAME = "exampleQueue";
        /** JNDI name for Topic Destination (use for Pub/Sub Mode) */
        static final String TOPIC_JNDI_NAME = "exampleTopic";
        /**
         * @param args
         */
        public static void main(String[] args) {
            Context jndiContext = null;
            ConnectionFactory connectionFactory = null;
            Connection connection = null;
            Session session = null;
            Destination destination = null;
            MessageProducer producer = null;
            // create a JNDI API IntialContext object
            try {
                jndiContext = new InitialContext();
            } catch (NamingException e) {
                System.out.println("Could not create JNDI Context:"
                        + e.getMessage());
                System.exit(1);
            }
            
            // look up ConnectionFactory and Destination
            try {
                connectionFactory = (ConnectionFactory) jndiContext
                        .lookup(CONNECTION_FACTORY_JNDI_NAME);
                // look up QUEUE_JNDI_NAME for PTP Mode
                // look up TOPIC_JNDI_NAME for Pub/Sub Mode
                destination = (Destination) jndiContext.lookup(QUEUE_JNDI_NAME);
            } catch (NamingException e) {
                System.out.println("JNDI look up failed:" + e.getMessage());
                System.exit(1);
            }
            
            // send Messages and finally release the resources.
            try {
                connection = connectionFactory.createConnection();
                session = connection.createSession(Boolean.FALSE,
                        Session.AUTO_ACKNOWLEDGE);
                producer = session.createProducer(destination);
                TextMessage message = session.createTextMessage();
                for (int i = 0; i < 3; i++) {
                    message.setText(String.format("This is the %dth message.",
                            i + 1));
                    producer.send(message);
                }
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (session != null)
                        session.close();
                    if (connection != null)
                        connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    SimpleProcedure.java

说明：

1.  第23行声明的JNDI name for ConnectionFactory的值是固定的。
2.  第25行声明的JNDI name for Queue的值，对应于jndi.properties的queue的jndi name。
3.  第27行声明的JNDI name for Topic的值，对应于jndi.properties的topic的jndi name。
4.  第55行look up Destination的实例，参数取QUEUE\_JNDI\_NAME或TOPIC\_JNDI\_NAME，分别对应PTP模式和Pub/Sub模式。

## 2.consumer-client ##

基于Maven的simple project。

文件目录结构

pom.xml
    src/main/resources/
        |---- jndi.properties
    src/main/java/
        |---- cn.sinobest.asj.consumer.jms.clientmode
                  |---- SimpleConsumer.java # 基于客户端编程模型，从ActiveMQ接收消息

### 文件内容 ###

#### 1.pom.xml ####

<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>
        <groupId>cn.sinoebst.asj</groupId>
        <artifactId>jms-consumer</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>jms-consumer</name>
        <description>基于ActiveMQ的Consumer Client。</description>
        <dependencies>
            
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-client</artifactId>
                <version>5.13.2</version>
            </dependency>
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-simple</artifactId>
                <version>1.7.19</version>
            </dependency>
        </dependencies>
    </project>

#### 2.jndi.properties ####

#### 3.SimpleConsumer.java ####

package cn.sinobest.asj.consumer.jms.clientmode;
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.naming.Context;
    import javax.naming.InitialContext;
    import javax.naming.NamingException;
    /**
     * A simple demo for consumer to receive message from ActiveMQ.<br>
     * 
     * @author lijinlong
     * 
     */
    public class SimpleConsumer {
        /** JNDI name for ConnectionFactory */
        static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
        /** JNDI name for Queue Destination (use for PTP Mode) */
        static final String QUEUE_JNDI_NAME = "exampleQueue";
        /** JNDI name for Topic Destination (use for Pub/Sub Mode) */
        static final String TOPIC_JNDI_NAME = "exampleTopic";
        /**
         * @param args
         */
        public static void main(String[] args) {
            Context jndiContext = null;
            ConnectionFactory connectionFactory = null;
            Connection connection = null;
            Session session = null;
            Destination destination = null;
            MessageConsumer consumer = null;
            // create a JNDI API IntialContext object
            try {
                jndiContext = new InitialContext();
            } catch (NamingException e) {
                System.out.println("Could not create JNDI Context:"
                        + e.getMessage());
                System.exit(1);
            }
            // look up ConnectionFactory and Destination
            try {
                connectionFactory = (ConnectionFactory) jndiContext
                        .lookup(CONNECTION_FACTORY_JNDI_NAME);
                // look up QUEUE_JNDI_NAME for PTP Mode
                // look up TOPIC_JNDI_NAME for Pub/Sub Mode
                destination = (Destination) jndiContext.lookup(QUEUE_JNDI_NAME);
            } catch (NamingException e) {
                System.out.println("JNDI look up failed:" + e.getMessage());
                System.exit(1);
            }
            // receive Messages and finally release the resources.
            try {
                connection = connectionFactory.createConnection();
                connection.start(); // connection should be called in
                                    // receiver-client
                session = connection.createSession(Boolean.FALSE,
                        Session.AUTO_ACKNOWLEDGE);
                consumer = session.createConsumer(destination);
                long timeout = 10 * 1000;
                for (Message message = consumer.receive(timeout); message != null; message = consumer
                        .receive(timeout)) {
                    String text = ((TextMessage) message).getText();
                    System.out.println(String.format("receive a message:%s", text));
                }
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (session != null)
                        session.close();
                    if (connection != null)
                        connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    SimpleConsumer.java

说明：在第65行调用了Connection\#start方法，否则无法收到消息（目前不明白原因）。

## 3.测试 ##

### 3.1.基于PTP Mode测试 ###

1.  确保SimpleProducer和SimpleConsumer中，都是根据QUEUE\_JNDI\_NAME来look up Destination实例的。
2.  确保ActiveMQ启动正常。
3.  运行SimpleProducer。
4.  运行SimpleConsumer - 看到信息输出。

### 3.2.基于Pub/Sub测试 ###

1.  确保SimpleProducer和SimpleConsumer中，都是根据TOPIC\_JNDI\_NAME来look up Destination实例的。
2.  确保ActiveMQ启动正常。
3.  运行SimpleConsumer。  
    Pub/Sub的其中一个特点是：先订阅，再接收。所以要先运行Consumer。
4.  尽快运行SimpleProducer。  
    Consumer设置的timeout是10s，如果10s内没有收到消息就会退出。
5.  SimpleConsumer的控制台有输出信息。  
    输出内容和PTP的测试结果一致。

## 4.编程获取ConnectionnFactory和Destination ##

虽然不建议使用编程的方式获取ConnectionFactory和Destination，但是还是记录一下：

// 创建工厂实例
    // javax.jms.ConnectionFactory
    // org.apache.activemq.ActiveMQConnectionFactory
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
        ActiveMQConnectionFactory.DEFAULT_USER,
        ActiveMQConnectionFactory.DEFAULT_PASSWORD,
        ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
    
    // javax.jms.Connection
    Connection connection = connectionFactory.createConnection();
    // javax.jms.Session
    Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    
    // 创建目的地
    // javax.jms.Destination
    Destination destination = session.createQueue("example.queue");
    Destination destination = session.createTopic("example.topic");

参考文档：https://docs.oracle.com/cd/E19148-01/820-0533/aerbg/index.html

原文转自：http://www.cnblogs.com/ywjy/p/5433747.html

[Center]: /images/20210923/2b5c437a82144250bf07c72a33d86ed9.png
[Center 1]: /images/20210923/3c93c3b1ec2d451db3069a0171300d65.png
[Center 2]: /images/20210923/c4407369f92e4575a45a4aaa5c430550.png
[Link 1]: https://docs.oracle.com/cd/E19148-01/820-0533/aerap/index.html
[JMS]: http://www.cnblogs.com/ywjy/articles/5433747.html
[Center 3]: /images/20210923/471d3e39ac764aec91ccf6f28bb32823.png
[Center 4]: /images/20210923/b64caf13ce69419b8cad8b691dd1a141.png
[Center 5]: /images/20210923/75d2dbfcc0994d3b819aeb46071ef34a.png
[Center 6]: /images/20210923/9512bed9fec048a78bf20280a54ebf31.png
[Center 7]: /images/20210923/fb537f8bc9304803bfb349f73bb82d53.png