使用 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 用于删除当前队列

参考

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