使用 ActiveMQ 实现JMS 异步调用

目录

• 启动 activemq 服务器
• activemq 的消息通道
  • queue
  • topic

简介

服务之间的同步调用，可以使用 http 或 rpc 来完成，但并非所有的调用都需要同步，有些场景下，当客户端调用服务端时，并不需要等待服务端做出响应，此时就应该使用异步调用。异步调用的常用方式是基于 mq (message queue) 来实现的。下文会以 activemq 为例进行讲解。

activemq 是 java 世界中最为流行的开源消息中间件，它不仅功能强大，而且性能稳定。它可全面支持 jms（java 消息服务）技术规范，为 java 应用程序提供标准的 jms api。

此外 activemq 具备与 spring 框架整合的能力，它一直都是 spring 应用程序的消息中间件标配。同样， spring boot 也提供了 activemq 的开箱即用的插件，只需要几项配置，就能接入 activemq，并轻松使用 jms api 编写异步消息通信程序。

active mq 官网地址如下

http://activemq.apache.org

启动 activemq 服务器

先使用 docker 安装 activemq ，目前 activemq 官方并未提供相应的 docker 镜像，我们选择使用第三方镜像 webcenter/activemq 。

docker pull webcenter/activemq:5.14.3

接下来运行 activemq

docker run  -d -p 8161:8161 -p 61616:61616 -e activemq_admin_login=admin -e activemq_admin_password=admin --name activemq webcenter/activemq:5.14.3

在启动 activemq 容器时，容器对宿主机暴露了两个端口号：

• 8161: 表示 activemq 控制台端口号，可在浏览器中通过控制台来执行 activemq 的相关操作
• 61616: 表示 activemq 所监听的 tcp 端口号，应用程序可通过该端口号与 activemq 建立 tcp 连接，并完成后续的异步消息通信

此外，在启动 activemq 容器时，还提供了两个环境变量

• activemq_admin_login: 用于设置控制台管理员的用户名，默认为 admin
• activemq_admin_password: 用于设置控制台管理员的密码，默认为 admin

查看控制台

webcenter/activemq 镜像拥有一个基于 web 的控制台，可通过浏览器访问。容器启动完毕后，可以打开浏览器，并在地址栏中输入 http://localhost:8161

点击 manage activemq borker 链接，浏览器将弹出一个对话框，此时输入用户名和密码，认证通过后会进入管理界面

在管理界面中，包括 8 个功能菜单

1. home: 基本信息
2. queues： 管理的队列
3. topics： 查看所管理的主题
4. subscribers: 查看相关主题的订阅者
5. connections: 查看客户端的连接信息
6. network: 查看网络相关信息
7. scheduled: 查看 activemq 内部运行的定时任务
8. send： 通过表单方式查看向队列或主题发送具体消息

activemq 的消息通道

activemq 管理了两类消息通道，一类是队列(queue)，另一类叫做主题(topic)。

queue

queue 用于解决消息的 点对点 通信问题，也就是说，消息从生产者(producer) 发出后，首先进入 activemq 某个指定的 queue 中，然后再将消息传送给其中一个消费者（consumer）。

topic

topic 用于解决消息的发布与订阅(publish-subscribe) 通信问题，也就是说，消息从 producer 发出后，首先将其发布到 activemq 某个指定的 topic 上，然后将此消息分发给每个订阅者(subscriber) 。

比较

在具体场合下，灵活使用以上两种通信模式来实现 producer 与 consumer/subscriber 间的异步调用，从而解决调用方的耦合问题。可见，queue 能解决调用缓冲问题，topic 能解决消息广播问题， queue 与 topic 都能解决掉调用耦合问题，这些技术都为一个好的软件架构提供了有效的支撑。

开发生产者和消费者

下面就以 queue 为例，将 activemq 与 spring boot 进行整合，将 producer 作为客户端， consumer 作为服务端，通过 queue 实现客户端与服务端的异步调用

开发服务端（消费者）

首先创建一个名为 acitvemq-hello-server 的 spring boot 项目，如果在 eclipse 中安装了 spring tools ,可以在新建时选择 new spring starter project 选项。或者新建 maven 工程。对应的 maven 依赖如下

    <parent>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-parent</artifactid>
        <version>1.5.19.release</version>
        <relativepath/> <!-- lookup parent from repository -->
    </parent>

    <dependencies>
        <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-starter-activemq</artifactid>
        </dependency>
    </dependencies>

在 spring boot 框架中已经内置了对 activemq 的支持，我们只需要依赖 spring-boot-starter-mq 就能启动 activemq,此时还需要在 application.properties 文件中添加 activemq 配置项

spring.activemq.broker-url=tcp://10.104.10.1:61616
spring.activemq.user=admin
spring.activemq.password=admin

接下来创建 helloserver 的类,封装服务端相关代码

@component
public class helloserver {
    @jmslistener(destination="hello-queue")
    public void receive(string message) {
        system.out.println(message);
    }
}

使用 @component 注解，说明它可被 spring ioc 容器所管理。此时只需要使用 @jmslistener 注解，并将其绑定到 receive() 方法上，就能从 activemq 中接收响应的消息。

@jmslistener 注解中需要添加一个 destination 属性来指定 queue/topic 的名称，该名称具有唯一性。消息将以一个 string 类型参数的形式传入方法体中，也可以接收其他类型的消息，这取决于客户端发送的消息是哪种类型。spring jms 将消息放入 activemq 时会进行序列化，当消息从 activemq 取出时将进行反序列化，应用程序无需关注这些底层细节，只需要将精力放在业务逻辑上。

最后，编写一个 spring boot 应用程序启动类来启动服务端(使用 spring tools 工具会自动生成)

@springbootapplication
public class activemqhelloserverapplication {

    public static void main(string[] args) {
        springapplication.run(activemqhelloserverapplication.class, args);
    }
}

当服务端启动完毕后，将一直监听 activemq 的 hello-queue 队列中即将到来的消息，消息由客户端来发送。

开发客户端（生产者）

创建一个名为 active-mq-client 的 maven 项目, pom.xml 文件内容与服务端相似。application.properties 文件与服务端相同。

接下来创建一个名为 helloclient 的类，将其作为客户端。

@component
public class helloclient {
    
    @autowired
    private jmstemplate jmstemplate;
    
    public void send(string message) {
        jmstemplate.convertandsend("hello-queue", message);
    }
}

这里使用了 @autowired 注解, jmstemplate 对象注入进来，还编写了一个 send() 方法，在该方法中调用 jmstemplate 对象的 convertandsend 来转换并发送消息。

最后使用 spring boot 应用程序启动类来启动客户端

@springbootapplication
public class activemqhelloclientapplication {

    @autowired
    private helloclient helloclient;
    
    @postconstruct
    public void init() {
        helloclient.send("hello world");
    }
    
    public static void main(string[] args) {
        springapplication.run(activemqhelloclientapplication.class, args);
    }

}

需要注意的是， init() 方法带有 @postconstruct 注解，表示 spring ioc 容器实例化 activemqhelloclientapplication 类后将调用该方法。

运行 main() 方法可以启动客户端应用程序，并可以在服务端应用程序控制台中看到 client 发送的消息，也可以在 activemq 控制台中查看队列的当前状态

queue 表格中列明的含义如下

• name 表示队列名称，可在应用程序中自动创建，也可在 activemq 控制台中手动创建
• number of pending messages 表示阻塞在队列中未经消费的消息条数
• number of consumers 表示正在与 activemq 建立连接的消费者数量
• messages enqueued 表示进入队列的消息数量
• messages dequeued 表示离开队列的消息数量

此外，还有下面几种操作

• browser 用于查看当前队列中消息的相关细节
• active consumers 用于查看当前活动消费者的相关信息
• active producers 用于查看当前活动生产者的相关信息
• send to 用于向当前队列中发送具体消息
• purge 用于清空队列中的消息
• delete 用于删除当前队列

参考

• 《架构探险—轻量级微服务架构》