JMS 和 Apache-ActiveMQ 简介

JMS（Java Message Service，Java 消息服务）是一个 Java 面向消息中间件（MOM）的 API，用于 Java 应用程序或分布式系统之间发送信息，异步通信；

JMS 具有以下优势

通信的异步性，客户端获取信息不需要主动发送请求，由 JMS 中间件自动推送信息；
消息发送的准确性，JMS 中间件可以保证信息只会发送一次，不会发送重复信息；

JMS 的消息传输模型有以下 2 种

1）点对点模型（P2P，queue）

P2P 模型中，应用程序有消息队列（queue），消息发送者，消息接收者组成，每一个消息发送给一个中间件队列，再由队列发送给单个接收者。具有以下的性质：

每一个消息只有一个接收者（当多个接收者使用同一个消息队列时，接收消息是竞争式的）；
发布者和接收者没有时间依赖，当消息发送者发送消息时，即使但是接收者不运行，当只要接收者上线，就可以接收到信息；
当接收者接收到消息时，会发送确认通知；

2）发布/订阅模型（topic）

分布/订阅模型，应用程序有主题（topic），消息发送者，一系列的消息接收者组成，topic 用于保存和传递消息，并且会一直保存到被传递到客户端。具有以下特性：

一个消息可以传递给多个消息接收者（当多个接收者接收到同一个消息时，每一个接收者都会接收到一份该消息的拷贝）；
发布者和接收者有时间依赖，只有当客户端创建订阅后，才可以开始接收到消息（而且是当前开始消息发送者发布的消息），同时订阅者需要一直保持活动状态才可以接收消息；
JMS 允许发布订阅者创建一个可持久化的订阅，这样即使订阅者没有上线，也可以在上线后接收到消息；

JMS 这是一个平台无关的 Java API，由 MOM 提供商对其进行支持，这里介绍其中比较轻量化，高效的 Apache-ActiveMQ；

ActiveMQ 项目官网：http://activemq.apache.org

ActiveMQ的安装和部署

由于 JMS 消息中间是独立于客户端，服务端程序的，在运行程序前要先部署好 JMS 服务程序；

Active-MQ 下载地址：http://activemq.apache.org/download.html

选择合适的系统版本下载，解压后，window 版本X64运行 apache-activemq/bin/win64/activemq.bat，Unix 版本运行 apache-activemq/bin/activemq.sh，即可启动 ActiveMQ；

ActiveMQ 提供了一个基于内嵌 Jetty 的管理程序，可以在启动 acgtivemq 后访问服务器的 http://127.0.0.1:8161/admin 访问该管理页面（默认密码：admin，admin）

如果要增加、修改 ActiveMQ 的程序用户认证，可以在 apache-activemq/conf/activemq.xml 文件中的<broker>节点下添加：

<broker>

    ....

     <!--增加用户认证-->

        <plugins>

            <simpleAuthenticationPlugin>

                <users>

                    <authenticationUser username="assad" password="assad123" groups="admins" />

                </users>

            </simpleAuthenticationPlugin>

        </plugins>

</broker>

要增加、修改程序访问的 openwire 路由端口，可以修改<transportConnectors>中相关的<transportConnector>节点：

   <transportConnectors>

            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->

            <transportConnector name="openwire" uri="tcp://127.0.0.1:23333?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>

        ......

  </transportConnectors>

ActiveMQ 支持 3 种类型的数据传输对象：

TextMessager：传输文本对象；
MapMessager：传输 Map 类型对象；
ObjectMessager：传输其他 Object，调用该 Object 的序列化方法序列化为二进制字节传输；

关于使用 Java 编写基于 ActiveMQ 的 JMS 基本程序，可以参考：https://www.cnblogs.com/jaycekon/p/6225058.html

但由于直接用 Java 直接编写步骤比较繁琐，可以使用 Spring 的集成，相对简单很多，详见下面：

Spring 集成 JMS（ActiveMQ）

Spring 提供的 org.springframework:spring-jms 包对 JMS 进行了封装，使用 Spring 风格的 JmsTemplate 封装了 Adminstred Object（管理对象）、ConnectionFactory（连接工厂）、Session（会话）、Destination（目的地）等的配置；

以下示例完整代码地址：https://gitee.com/assad/jms-spring-sample

异步模型快速示例（queue/topic）

示例代码位于以上地址的 basic_sample 路径下；

需要导入的相关依赖包：

//spring_sample 核心依赖

compile 'org.springframework:spring-core:4.3.11.RELEASE'

compile 'org.springframework:spring-beans:4.3.11.RELEASE'

//Apache-ActiveMQ ( JMS API 的一种思想) 依赖，Spring 相关支持依赖

compile 'org.apache.activemq:activemq-all:5.15.3'

compile 'org.apache.activemq:activemq-pool:5.15.3'

compile 'org.springframework:spring-jms:4.3.11.RELEASE'

服务端

在 spring 上下文配置文件中进行 ActiveMQ 相关的配置：basic_sample/appContext-server.xml

    <!--ActiveMQ 相关配置-->

    <!--第三方工厂-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"

          p:brokerURL="tcp://127.0.0.1:23333"

          p:userName="assad"

          p:password="assad123"

          p:trustAllPackages="true"

          p:useAsyncSend="true"/>

    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"

          p:connectionFactory-ref="targetConnectionFactory"

          p:maxConnections="100" />

    <!--spring 管理真正 ConnectionFactory 的连接工厂，使用PooledConnectionFactory 封装过的 targetConnectionFactory -->

    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"

          p:targetConnectionFactory-ref="pooledConnectionFactory"/>

    <!--目的地配置: queue(point-to-point模式) -->

    <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">

        <constructor-arg index="0" value="spring-queue" />

    </bean>

    <!--目的地配置：Topic（发布/订阅模式）-->

    <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">

        <constructor-arg index="0" value="spring-topic" />

    </bean>

    <!--配置 Spring jsmTemplate 模板，用于发送和接收消息 -->

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"

          p:connectionFactory-ref="connectionFactory">

        <!--配置默认目的地（queue 或 topic）-->

        <property name="defaultDestination" ref="destinationQueue" />

        <!--使用spring提供的默认消息转换器，也可以装载自己实现的消息转化器-->

        <property name="messageConverter">

            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />

        </property>

    </bean>

要上配置文件配置使用 P2P 模型作为目的地，如果要更改为 Topic 模型，只需要丢该 jmsTemplate bean 节点 defaultDestination-ref 属性为之前配置好的 "destinationTpoic" 即可，同样可以在服务端发送消息代码中手动指定目的地；

服务端程序 sample/MessageProducer ：

package basic_sample.server;

import bean.User;

import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;

import org.springframework.jms.core.JmsTemplate;

//演示各种类型消息的发送

public class MessageProducer {

    private static JmsTemplate jmsTemplate = null;

    private static final Logger log = LogManager.getLogger();

    public static void main(String[] args){

        //获取 JmsTemplate 对象

        jmsTemplate = new ClassPathXmlApplicationContext("basic_sample/appContext-server.xml").getBean("jmsTemplate",JmsTemplate.class);

        //发送文本信息

        String message = "basic_sample.server: Hello world!";

        jmsTemplate.convertAndSend(message);

        log.debug(message);

        //发送 Map 类型信息

        Map<String,Integer> info = new HashMap<>();

        info.put("Guangdong",64815);

        info.put("Jiangshu",62604);

        info.put("Shangdong",54868);

        info.put("Zhijiang",36958);

        jmsTemplate.convertAndSend(info);

        log.debug(info.keySet().stream().map(key -> key+":"+info.get(key)).collect(Collectors.joining(", ")));

        //发送 Object 类型信息（User对象）

        User user = new User();

        user.setId(20137);

        user.setName("assad");

        user.setCity("Guangzhou");

        jmsTemplate.convertAndSend(user);

        log.debug(user);

注意在传输对象时，spring 配置文件中的 connectionFactory 的 trustAllPackages 参数需要设置为 true；

客户端

在这个示例中客户端的 Spring 上下文配置文件 basic_sample/appContext-clinet.xml

    <!--ActiveMQ 相关配置-->

    <!--第三方工厂-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"

          p:brokerURL="tcp://127.0.0.1:23333"

          p:userName="assad"

          p:password="assad123"

          p:trustAllPackages="true" />

    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"

          p:connectionFactory-ref="targetConnectionFactory"

          p:maxConnections="100" />

    <!--spring 管理真正 ConnectionFactory 的连接工厂，使用PooledConnectionFactory 封装过的 targetConnectionFactory -->

    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"

          p:targetConnectionFactory-ref="pooledConnectionFactory"/>

    <!--目的地配置: queue(point-to-point模式) -->

    <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">

        <constructor-arg index="0" value="spring-queue" />

    </bean>

    <!--目的地配置：Topic（发布/订阅模式）-->

    <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">

        <constructor-arg index="0" value="spring-topic" />

    </bean>

    <!--配置 Spring jsmTemplate 模板，用于发送和接收消息 -->

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"

          p:connectionFactory-ref="connectionFactory">

        <!--配置默认目的地（queue 或 topic）-->

        <property name="defaultDestination" ref="destinationQueue" />

        <!--使用spring提供的默认消息转换器，也可以装载自己实现的消息转化器-->

        <property name="messageConverter">

            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />

        </property>

    </bean>

客户端程序，basic_sample/client/MessageConsumer

package basic_sample.client;

import bean.User;

import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;

import org.springframework.jms.core.JmsTemplate;

//演示接收各种类型的消息

public class MessageConsumer {

    private static JmsTemplate jmsTemplate = null;

    private static final Logger log = LogManager.getLogger();

    public static void main(String[] args){

        //创建 JmsTemplate 对象

        jmsTemplate = new ClassPathXmlApplicationContext("basic_sample/appContext-client.xml").getBean("jmsTemplate",JmsTemplate.class);

        //接收文本对象

        String msg = (String) jmsTemplate.receiveAndConvert();

        log.debug(msg);

        //接收 Map 对象

        Map<String,Integer> info = (HashMap<String,Integer>) jmsTemplate.receiveAndConvert();

        log.debug(info.keySet().stream().map(key -> key+":"+info.get(key)).collect(Collectors.joining(", ")));

        //接收 Object 类型对象

        User user = (User) jmsTemplate.receiveAndConvert();

        log.debug(user);

运行步骤，启动 activemq 后，分别运行客户端，服务端；客户端接收的信息如下：

关于 queue 模式和 topic 模式客户端接收信息的区别，可以分别修改 jmsTemplate 的 defaultDestination 参数，运行 basic_sample/MessageProducer2 和 basic_sample/MessageConsumer2 进行测试，结果和以上介绍 JMS 的 2 种传输模型相符合；

客户端使用监听器的方式接收信息

以上示例的客户端是使用 JmsTemplate 接收消息，如果客户端需要使用监听器的方式接收信息，可以如下步骤：

创建一个监听器 client/MyListener

package persistence_sample.client;

import javax.jms.*;

public class MyListener implements MessageListener {

    private static final Logger log = LogManager.getLogger();

    @Override

    public void onMessage(Message message) {

        try {

            //接收文本对象

            String msg = (String) jmsTemplate.receiveAndConvert();

            log.debug(msg);

            //接收 Map 对象

            Map<String,Integer> info = (HashMap<String,Integer>) jmsTemplate.receiveAndConvert();

            log.debug(info.keySet().stream().map(key -> key+":"+info.get(key)).collect(Collectors.joining(", ")));

            //接收 Object 类型对象

            User user = (User) jmsTemplate.receiveAndConvert();

            log.debug(user);

        } catch (JMSException e) {

            e.printStackTrace();

在 spring 上下文中配置文件 appContext-client.xml 中配置监听器：

    <!--ActiveMQ 相关配置-->

    <!--第三方工厂-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"

          p:brokerURL="tcp://127.0.0.1:23333"

          p:userName="assad"

          p:password="assad123"

          p:trustAllPackages="true" />

    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"

          p:connectionFactory-ref="targetConnectionFactory"

          p:maxConnections="100" />

    <!--spring 管理真正 ConnectionFactory 的连接工厂，使用PooledConnectionFactory 封装过的 targetConnectionFactory -->

    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"

          p:targetConnectionFactory-ref="pooledConnectionFactory"/>

    <!--目的地配置: queue(point-to-point模式) -->

    <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">

        <constructor-arg index="0" value="spring-queue" />

    </bean>

     <!--装载自定义消息监听器-->

    <bean id="myTopicListener" class="persistence_sample.client.TopicListener" />

    <!--消息监听容器-->

    <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"

          p:connectionFactory-ref="connectionFactory"

          p:destination-ref="destinationQueue"

          p:receiveTimeout="1000"

          p:messageListener-ref="myTopicListener">

        <!--messageListener-ref 配置消息监听器-->

        <property name="messageConverter">

            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />

        </property>

    </bean>

创建客户端 client/StartReciver

public class StartReceiver {

    public static void main(String[] args){

        ApplicationContext context = new ClassPathXmlApplicationContext("persistence_sample/appContext-client.xml");

        while(true){

Topic 消息持久化配置

topic 非持久化消息订阅会持续到订阅对象的生命周期，由于 topic 模型的实际产生场景一般会有多个订阅对象，网络 IO 开销会很大，ActiveMQ 提供了持久化实现，可以将 topic 消息储存到磁盘文件或者数据库的实现，如果持久化没有活动的订阅者，JMS 会保存该订阅消息，直到消息被订阅或过期；

以下是一个实现示例，代码模块在示例代码地址的 persistence 目录下；

服务端

spring 配置文件：persistence_sample/appContext-server.xml

 <!--ActiveMQ 相关配置-->

    <!--第三方工厂-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"

          p:brokerURL="tcp://127.0.0.1:23333"

          p:userName="assad"

          p:password="assad123"

          p:trustAllPackages="true" />

    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"

          p:connectionFactory-ref="targetConnectionFactory"

          p:maxConnections="100" />

    <!--spring 管理真正 ConnectionFactory 的连接工厂，使用PooledConnectionFactory 封装过的 targetConnectionFactory -->

    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"

          p:targetConnectionFactory-ref="pooledConnectionFactory"/>

    <!--目的地配置：queue（发布/订阅模式）-->

    <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">

        <constructor-arg index="0" value="spring-topic-persistence" />

    </bean>

    <!--配置 Spring jsmTemplate 模板，配置为 Topic 模式 -->

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"

          p:connectionFactory-ref="connectionFactory"

          p:defaultDestination-ref="destinationTopic"

          p:pubSubDomain="true"

          p:sessionAcknowledgeMode="1"

          p:explicitQosEnabled="true"

          p:deliveryMode="2"

          p:receiveTimeout="10000">

        <!--以上配置中，explicitQosEnabled用于生效deliveryMode配置

            deliveryMode 为发送模式，1 为非持久，2 为持久-->

        <!--使用spring提供的默认消息转换器-->

        <property name="messageConverter">

            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />

        </property>

    </bean>

服务端程序 persistence_sample/server/MessageProducer

ackage persistence_sample.server;

import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;

import org.springframework.jms.core.JmsTemplate;

public class MessageProducer {

    private static JmsTemplate jmsTemplate ;

    private static final Logger log = LogManager.getLogger();

    public static void main(String[] args) throws InterruptedException {

        jmsTemplate = new ClassPathXmlApplicationContext("persistence_sample/appContext-server.xml").getBean("jmsTemplate",JmsTemplate.class);

        //发送20条信息

        for(int i=1; i<=20; i++){

            String message = String.format("Message@<%d>",i);

            jmsTemplate.convertAndSend(message);

            log.debug(message);

            Thread.sleep(900);

客户端

示例代码客户端使用监听器的方式实现获取消息；

spring 配置文件：persistence_sample/appContext-client.xml

    <!--ActiveMQ 相关配置-->

    <!--第三方工厂-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"

          p:brokerURL="tcp://127.0.0.1:23333"

          p:userName="assad"

          p:password="assad123"

          p:trustAllPackages="true" />

    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"

          p:connectionFactory-ref="targetConnectionFactory"

          p:maxConnections="100" />

    <!--spring 管理真正 ConnectionFactory 的连接工厂，使用PooledConnectionFactory 封装过的 targetConnectionFactory -->

    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"

          p:clientId="client_001"

          p:targetConnectionFactory-ref="pooledConnectionFactory"/>

    <!--目的地配置：Topic（发布/订阅模式）-->

    <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">

        <constructor-arg index="0" value="spring-topic-persistence" />

    </bean>

    <!--装载自定义消息监听器-->

    <bean id="myTopicListener" class="persistence_sample.client.TopicListener" />

    <!--消息监听容器-->

    <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"

          p:connectionFactory-ref="connectionFactory"

          p:destination-ref="destinationTopic"

          p:pubSubDomain="true"

          p:subscriptionDurable="true"

          p:clientId="client_001"

          p:durableSubscriptionName="client_001"

          p:receiveTimeout="1000"

          p:messageListener-ref="myTopicListener">

        <!--pubSubDomain开启订阅模式，subscriptionDurable设置允许持久化，

        clientId、durableSubscriptionName 设置接收者ID-->

        <property name="messageConverter">

            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />

        </property>

    </bean>

消息监听器 persistence_sample/client/TopicListener

package persistence_sample.client;

import javax.jms.*;

public class TopicListener implements MessageListener {

    private static final Logger log = LogManager.getLogger();

    @Override

    public void onMessage(Message message) {

        try {

            String msg = ((TextMessage)message).getText();

            log.debug(msg);

        } catch (JMSException e) {

            e.printStackTrace();

客户端启动程序 persistence_sample/client/StartReceiver

package persistence_sample.client;

import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;

import org.springframework.context.ApplicationContext;

public class StartReceiver {

    public static void main(String[] args){

        ApplicationContext context = new ClassPathXmlApplicationContext("persistence_sample/appContext-client.xml");

        while(true){

ActiveMQ 的配置

具体持久化的实现实在 ActiveMQ 中进行配置的，ActiveMQ 支持以下几种持久化方式：

AMQ 文件储存；
KahaDB 文件储存；
LevelDB 储存
JDBC 数据库持久化；

这里演示使用 KahaDB 文件作为持久化文件储存格式，在 apache-activemq/conf/activemq.xml 中<brokler>节点下添加以下：

<broker>

    ....

    <!--持久化配置-->

    <persistenceAdapter>

            <kahaDB directory="${activemq.data}/kahadb"/>

        </persistenceAdapter>

    ....

</broker>

关于ActiveMQ 详细的持久化配置，参考：https://my.oschina.net/u/1455908/blog/310115

Spring framework(10)：集成 JMS 异步消息队列（ActiveMQ）相关推荐

八.利用springAMQP实现异步消息队列的日志管理
经过前段时间的学习和铺垫,已经对spring amqp有了大概的了解.俗话说学以致用,今天就利用springAMQP来完成一个日志管理模块.大概的需求是这样的:系统中有很多地方需要记录操作日志,比如登 ...
异步消息队列Celery
Celery是异步消息队列, 可以在很多场景下进行灵活的应用.消息中包含了执行任务所需的的参数,用于启动任务执行, suoy所以消息队列也可以称作在web应用开发中, 用户触发的某些事件需要较长事件 ...
C#实现异步消息队列
C#实现异步消息队列原文:C#实现异步消息队列拿到新书<.net框架设计>,到手之后迅速读了好多,虽然这本书不像很多教程一样从头到尾系统的讲明一些知识,但是从项目实战角度告诉我们如何使 ...
redis stream java消息队列_Redis 异步消息队列与延时队列
消息中间件,大家都会想到 Rabbitmq 和 Kafka 作为消息队列中间件,来给应用程序之间增加异步消息传递功能.这两个中间件都是专业的消息队列中间件,特性之多超出了大多数人的理解能力.但是这种属 ...
Redis异步消息队列
一.异步消息队列介绍个人认为消息队列的主要特点是异步处理,主要目的是减少请求响应时间和解耦.所以主要的使用场景就是将比较耗时而且不需要即时(同步)返回结果的操作作为消息放入消息队列.同时由于使用了消 ...
消息队列 - ActiveMQ
消息队列 - ActiveMQ 一.入门概述 1.在什么场景下使用消息中间件,为什么使用 2. 消息队列是什么 3. 去哪下二.ActiveMQ的安装和控制台 1.Linux安装 2.Apache ...
CentOS源码安装消息队列ActiveMQ
消息队列ActiveMQ介绍 JMS全称:Java Message Service中文:Java消息服务.JMS是java的一套API标准,最初的目的是为了是应用程序能够访问现有的MOM系统(MOM是 ...
java分布式面试题之消息队列ActiveMQ部分
java分布式面试题之消息队列ActiveMQ部分 java分布式面试题之消息队列ActiveMQ部分 1.如何使用ActiveMQ解决分布式事务? 在互联网应用中,基本都会有用户注册的功能.在注册的 ...
Spring Boot：使用Rabbit MQ消息队列
综合概述消息队列消息队列就是一个消息的链表,可以把消息看作一个记录,具有特定的格式以及特定的优先级.对消息队列有写权限的进程可以向消息队列中按照一定的规则添加新消息,对消息队列有读权限的进程则可以 ...

Spring framework(10)：集成 JMS 异步消息队列（ActiveMQ）