8.1 使用 JMS 发送消息

8.1 使用 JMS 发送消息

JMS 是一个 Java 标准，它定义了一个用于使用消息代理的公共 API。自 2001 年首次引入以来，JMS 一直是 Java 中异步消息传递的首选方法。在 JMS 之前，每个消息代理都有一个专用 API，这使得应用程序的消息代码在代理之间的可移植性更差。但是有了 JMS，所有兼容的实现都可以通过公共接口进行处理，这与 JDBC 为关系数据库操作提供公共接口的方式非常相似。

Spring 通过称为 JmsTemplate 的基于模板的抽象来支持 JMS。使用 JmsTemplate，很容易从生产者端跨队列和主题发送消息，并在消费者端接收这些消息。Spring 还支持消息驱动 POJO 的概念：简单的 Java 对象以异步方式对队列或主题上到达的消息做出响应。

我们将探讨 Spring 的 JMS 支持，包括 JmsTemplate 和消息驱动 POJO。但是在可以发送和接收消息之前，需要一个消息代理，它可以在生产者和消费者之间传递这些消息。让我们通过在 Spring 中设置消息代理来开始对 Spring JMS 的探索。

8.1.1 设置 JMS

在使用 JMS 之前，必须将 JMS 客户端添加到项目的构建中。使用 Spring Boot，这个过程简单的不能再简单了，需要做的仅仅是将 starter 依赖添加到构建中。但是，首先必须决定是使用 Apache ActiveMQ，还是使用较新的 Apache ActiveMQ Artemis Broker。

如果使用 ActiveMQ，需要添加以下依赖到项目的 pom.xml 文件中：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

如果选择 ActiveMQ Artemis，starter 如下所示：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-artemis</artifactId>
</dependency>

Artemis 是 ActiveMQ 的下一代重新实现，实际上这让 ActiveMQ 成为一个遗留选项。因此，对于 Taco Cloud，将选择 Artemis。但是，这种选择最终对如何编写发送和接收消息的代码几乎没有影响。唯一显著的区别在于如何配置 Spring 来创建与 Broker 的连接。

默认情况下，Spring 假设 Artemis Broker 正在监听 localhost 的 61616 端口。对于开发目的，这是可以的，但是一旦准备好将应用程序发送到生产环境中，就需要设置一些属性来告诉 Spring 如何访问代理。表 8.1 列出了最有用的属性。

例如，考虑应用程序中的以下条目。可能用于非开发设置的 yml 文件：

spring:
  artemis:
    host: artemis.tacocloud.com
    port: 61617
    user: tacoweb
    password: 13tm31n

这将设置 Spring，以创建到监听 artemis.tacocloud.com（端口 61617）的 Artemis Broker 的 broker 连接。它还设置将与该 broker 交互的应用程序的凭据，凭据是可选的，但建议用于生产部署。

如果要使用 ActiveMQ 而不是 Artemis，则需要使用表 8.2 中列出的 ActiveMQ 特定的属性。

请注意，不是为 Broker 的主机名和端口提供单独的属性，而是使用单个属性 spring.activemq.broker-url 指定 ActiveMQ Broker 的地址。URL 应该是 tcp:// URL，如下面的 YAML 片段所示：

spring:
  activemq:
    broker-url: tcp://activemq.tacocloud.com
    user: tacoweb
    password: 13tm31n

无论选择 Artemis 还是 ActiveMQ，当 Broker 在本地运行时，都不需要为开发环境配置这些属性。

但是，如果使用 ActiveMQ，则需要设置 spring.activemq.in-memory 属性为 false，以防止 Spring 启动内存中的 Broker。内存中的 Broker 可能看起来很有用，但它只在发布和消费同一个应用的消息时有用（这一点用处有限）。

在继续之前，将希望安装并启动一个 Artemis（或 ActiveMQ）Broker，而不是使用嵌入式 Broker。与其在这里重复安装说明，我建议你参考 Broker 文档了解详细信息：

• Artemis —— https://activemq.apache.org/artemis/docs/latest/using-server.html

• ActiveMQ —— http://activemq.apache.org/getting-started.html#GettingStarted-PreInstallationRequirements

有了构建中的 JMS starter 和等待将消息从一个应用程序传递到另一个应用程序的 Broker，就可以开始发送消息了。

8.1.2 使用 JmsTemplate 发送消息

在构建中有 JMS starter 依赖（无论 Artemis 还是 ActiveMQ），Spring Boot 将会自动配置 JmsTemplate，这样就可以将其注入并使用它发送和接收消息了。

JmsTemplate 是 Spring JMS 集成支持的核心。与 Spring 的其他面向模板的组件非常相似，JmsTemplate 消除了大量与 JMS 协同工作所需的样板代码。如果没有 JmsTemplate，将需要编写代码来创建与消息代理的连接和会话，并编写更多代码来处理在发送消息过程中可能抛出的任何异常。JmsTemplate 专注于真正想做的事情：发送消息。

JmsTemplate 有几个发送消息的有用方法，包括：

// 发送原始消息
void send(MessageCreator messageCreator) throws JmsException;
void send(Destination destination, MessageCreator messageCreator) throws JmsException;
void send(String destinationName, MessageCreator messageCreator) throws JmsException;
// 发送转换自对象的消息
void convertAndSend(Object message) throws JmsException;
void convertAndSend(Destination destination, Object message) throws JmsException;
void convertAndSend(String destinationName, Object message) throws JmsException;
// 发送经过处理后从对象转换而来的消息
void convertAndSend(Object message, MessagePostProcessor postProcessor) throws JmsException;
void convertAndSend(Destination destination, Object message, MessagePostProcessor postProcessor) throws JmsException;
void convertAndSend(String destinationName, Object message, MessagePostProcessor postProcessor) throws JmsException;

实际上只有两个方法，send() 和 convertAndSend()，每个方法都被重载以支持不同的参数。如果仔细观察，会发现 convertAndSend() 的各种形式可以分为两个子类。在试图理解所有这些方法的作用时，请考虑以下细分：

• send() 方法需要一个 MessageCreator 来制造一个 Message 对象。
• convertAndSend() 方法接受一个 Object，并在后台自动将该 Object 转换为一条 Message。
• 三种 convertAndSend() 方法会自动将一个 Object 转换成一条 Message，但也会接受一个 MessagePostProcessor，以便在 Message 发送前对其进行定制。

此外，这三个方法类别中的每一个都由三个重载的方法组成，它们是通过指定 JMS 目的地（队列或主题）的方式来区分的：

• 一个方法不接受目的地参数，并将消息发送到默认目的地。
• 一个方法接受指定消息目的地的目标对象。
• 一个方法接受一个 String，该 String 通过名称指定消息的目的地。

要使这些方法工作起来，请考虑下面程序清单中的 JmsOrderMessagingService，它使用 send() 方法的最基本形式。程序清单 8.1 使用 send() 发送订到到默认目的地

package tacos.messaging;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;

@Service
public class JmsOrderMessagingService implements OrderMessagingService {

    private JmsTemplate jms;

    @Autowired
    public JmsOrderMessagingService(JmsTemplate jms) {
        this.jms = jms;
    }

    @Override
    public void sendOrder(Order order) {
        jms.send(new MessageCreator() {
            @Override
            public Message createMessage(Session session)
                throws JMSException {
                return session.createObjectMessage(order);
            }
        });
    }
}

sendOrder() 方法调用 jms.send()，传递 MessageCreator 的匿名内部类实现。该实现重写 createMessage() 以从给定的 Order 对象创建新的对象消息。

我认为程序清单 8.1 中的代码虽然简单，但是有点笨拙。声明匿名内部类所涉及的过程会使简单的方法调用变得复杂。意识到 MessageCreator 是一个功能接口，这时可以用一个 lambda 表达式稍微调整一下 sendOrder() 方法：

@Override
public void sendOrder(Order order) {
    jms.send(session -> session.createObjectMessage(order));
}

但是请注意，对 jms.send() 的调用没有指定目的地。为了实现这一点，还必须使用 spring.jms.template.default-destination 属性指定一个默认的目的地名称。例如，可以在 application.yml 中设置属性：

spring:
  jms:
    template:
      default-destination: tacocloud.order.queue

在许多情况下，使用缺省目的地是最简单的选择。它让你指定一次目的地名称，允许代码只关心发送消息，而不关心消息被发送到哪里。但是，如果需要将消息发送到缺省目的地之外的目的地，则需要将该目的地指定为 send() 方法的参数。

一种方法是传递目标对象作为 send() 的第一个参数。最简单的方法是声明一个 Destination bean，然后将其注入执行消息传递的 bean。例如，下面的 bean 声明了 Taco Cloud 订单队列 Destination：

public Destination orderQueue() {
    return new ActiveMQQueue("tacocloud.order.queue");
}

需要注意的是，这里使用的 ActiveMQQueue 实际上来自于 Artemis（来自 org.apache.activemq.artemis.jms.client 包)。如果正在使用 ActiveMQ（而不是 Artemis），那么还有一个名为 ActiveMQQueue 的类（来自 org.apache.activemq.command 包）。

如果这个 Destination bean 被注入到 JmsOrderMessagingService 中，那么可以在调用 send() 时使用它来指定目的地：

private Destination orderQueue;

@Autowired
public JmsOrderMessagingService(JmsTemplate jms, Destination orderQueue) {
    this.jms = jms;
    this.orderQueue = orderQueue;
}

@Override
public void sendOrder(Order order) {
    jms.send(
        orderQueue,
        session -> session.createObjectMessage(order));
}

使用类似这样的 Destination 对象指定目的地，使你有机会配置 Destination，而不仅仅是目的地的名称。但是在实践中，几乎只指定了目的地名称，将名称作为 send() 的第一个参数通常更简单：

@Override
public void sendOrder(Order order) {
    jms.send(
        "tacocloud.order.com",
        session -> session.createObjectMessage(order));
}

虽然 send() 方法并不是特别难以使用（特别是当 MessageCreator 以 lambda 形式给出时），但是提供 MessageCreator 还是会增加一些复杂性。如果只需要指定要发送的对象（以及可选的目的地），不是会更简单吗？这简要地描述了 convertAndSend() 的工作方式，让我们来看看。

在发送前转换消息

JmsTemplates 的 convertAndSend() 方法不需要提供 MessageCreator，从而简化了消息发布。相反，将要直接发送的对象传递给 convertAndSend()，在发送之前会将该对象转换为消息。

例如，sendOrder() 的以下重新实现使用 convertAndSend() 将 Order 发送到指定的目的地：

@Override
public void sendOrder(Order order) {
    jms.convertAndSend("tacocloud.order.queue", order);
}

与 send() 方法一样，convertAndSend() 将接受 Destination 或 String 值来指定目的地，或者可以完全忽略目的地来将消息发送到默认目的地。

无论选择哪种形式的 convertAndSend()，传递给 convertAndSend() 的 Order 都会在发送之前转换为消息。实际上，这是通过 MessageConverter 实现的，它完成了将对象转换为消息的复杂工作。

配置消息转换器

MessageConverter 是 Spring 定义的接口，它只有两个用于实现的方法：

public interface MessageConverter {
    Message toMessage(Object object, Session session)
        throws JMSException, MessageConversionException;

    Object fromMessage(Message message);
}

这个接口的实现很简单，都不需要创建自定义实现。Spring 已经提供了一些有用的实现，就像表 8.3 中描述的那样。

SimpleMessageConverter 是默认的消息转换器，但是它要求发送的对象实现 Serializable 接口。这样要求可能还不错，但是可能更喜欢使用其他的消息转换器，如 MappingJackson2MessageConverter，来避免上述限制。

为了应用不同的消息转换器，需要做的是将选择的转换器声明为一个 bean。例如，下面这个 bean 声明将会使用 MappingJackson2MessageConverter 而不是 SimpleMessageConverter：

@Bean
public MappingJackson2MessageConverter messageConverter() {
    MappingJackson2MessageConverter messageConverter =
        new MappingJackson2MessageConverter();
    messageConverter.setTypeIdPropertyName("_typeId");
    return messageConverter;
}

注意一下，你在返回 MappingJackson2MessageConverter 之前调用了 setTypeIdPropertyName()。这是非常重要的，因为它使接收者知道要将传入消息转换成什么类型。默认情况下，它将包含被转换类型的完全限定类名。但这有点不灵活，要求接收方也具有相同的类型，具有相同的完全限定类名。

为了实现更大的灵活性，可以通过调用消息转换器上的 setTypeIdMappings() 将合成类型名称映射到实际类型。例如，对消息转换器 bean 方法的以下更改将合成订单类型 ID 映射到 Order 类：

@Bean
public MappingJackson2MessageConverter messageConverter() {
    MappingJackson2MessageConverter messageConverter =
        new MappingJackson2MessageConverter();
    messageConverter.setTypeIdPropertyName("_typeId");

    Map<String, Class<?>> typeIdMappings = new HashMap<String, Class<?>>();
    typeIdMappings.put("order", Order.class);
    messageConverter.setTypeIdMappings(typeIdMappings);

    messageConverter;
}

与在消息的 _typeId 属性中发送完全限定的类名不同，将发送值 order。在接收应用程序中，将配置类似的消息转换器，将 order 映射到它自己对 order 的理解。订单的实现可能在不同的包中，有不同的名称，甚至有发送者 Order 属性的一个子集。

后期处理消息

让我们假设，除了利润丰厚的网络业务，Taco Cloud 还决定开几家实体 Taco 连锁店。考虑到他们的任何一家餐馆也可以成为 web 业务的执行中心，他们需要一种方法来将订单的来源传达给餐馆的厨房。这将使厨房工作人员能够对商店订单采用与网络订单不同的流程。

在 Order 对象中添加一个新的 source 属性来携带此信息是合理的，可以用 WEB 来填充在线订单，用 STORE 来填充商店中的订单。但这将需要更改网站的 Order 类和厨房应用程序的 Order 类，而实际上，这些信息只需要为 taco 准备人员提供。

更简单的解决方案是在消息中添加一个自定义头信息，以承载订单的源。如果正在使用 send() 方法发送 taco 订单，这可以通过调用消息对象上的 setStringProperty() 轻松实现：

jms.send("tacocloud.order.queue",
        session -> {
            Message message = session.createObjectMessage(order);
            message.setStringProperty("X_ORDER_SOURCE", "WEB");
        });

这里的问题是没有使用 send()。通过选择使用 convertAndSend()，Message 对象是在幕后创建的，并且不能访问它。

幸运的是，有一种方法可以在发送消息之前调整在幕后创建的 Message。通过将 MessagePostProcessor 作为最后一个参数传递给 convertAndSend()，可以在消息创建之后对其进行任何操作。下面的代码仍然使用 convertAndSend()，但是它也使用 MessagePostProcessor 在消息发送之前添加 X_ORDER_SOURCE 头信息：

jms.convertAndSend("tacocloud.order.queue", order,
    new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message)
            throws JMSException {
            message.setStringProperty("X_ORDER_SOURCE", "WEB");
            return message;
        }
});

可能注意到了 MessagePostProcessor 是一个函数接口，这意味着可以使用 lambda 将其简化为匿名内部类：

jms.convertAndSend("tacocloud.order.queue", order,
    message -> {
        message.setStringProperty("X_ORDER_SOURCE", "WEB");
        return message;
    });

尽管只需要这个特定的 MessagePostProcessor 来处理对 convertAndSend() 的调用，但是可能会发现自己使用同一个 MessagePostProcessor 来处理对 convertAndSend() 的几个不同调用。在这些情况下，也许方法引用是比 lambda 更好的选择，避免了不必要的代码重复：

@GetMapping("/convertAndSend/order")
public String convertAndSendOrder() {
    Order order = buildOrder();
    jms.convertAndSend("tacocloud.order.queue", order, this::addOrderSource);
    return "Convert and sent order";
}

private Message addOrderSource(Message message) throws JMSException {
    message.setStringProperty("X_ORDER_SOURCE", "WEB");
    return message;
}

已经看到了几种发送消息的方法。但是，如果没有人收到信息，就没有什么用处。让我们看看如何使用 Spring 和 JMS 接收消息。

8.1.3 接收 JMS 消息

在消费消息时，可以选择 拉模型（代码请求消息并等待消息到达）或 推模型（消息可用时将消息传递给代码）。

JmsTemplate 提供了几种接收消息的方法，但它们都使用拉模型。调用其中一个方法来请求消息，线程会发生阻塞，直到消息可用为止（可能是立即可用，也可能需要一段时间）。

另一方面，还可以选择使用推模型，在该模型中，定义了一个消息监听器，它在消息可用时被调用。

这两个选项都适用于各种用例。人们普遍认为推模型是最佳选择，因为它不会阻塞线程。但是在某些用例中，如果消息到达得太快，侦听器可能会负担过重。拉模型允许使用者声明他们已经准备好处理新消息。

让我们看看接收消息的两种方式。我们将从 JmsTemplate 提供的拉模型开始。

使用 JmsTemplate 接收

JmsTemplate 提供多个用于拉模式的方法，包括以下这些：

Message receive() throws JmsException;
Message receive(Destination destination) throws JmsException;
Message receive(String destinationName) throws JmsException;

Object receiveAndConvert() throws JmsException;
Object receiveAndConvert(Destination destination) throws JmsException;
Object receiveAndConvert(String destinationName) throws JmsException;

可以看到，这 6 个方法是 JmsTemplate 中的 send() 和 convertAndSend() 方法的镜像。receive() 方法接收原始消息，而 receiveAndConvert() 方法使用配置的消息转换器将消息转换为域类型。对于其中的每一个，可以指定 Destination 或包含目的地名称的 String，也可以从缺省目的地获取一条消息。

要查看这些操作，需要编写一些代码来从 tacocloud.order.queue 的目的地拉取 Order。下面的程序清单显示了 OrderReceiver，这是一个使用 JmsTemplate.receive() 接收 Order 数据的服务组件。程序清单 8.2 从队列中拉取订单

package tacos.kitchen.messaging.jms;

import javax.jms.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.stereotype.Component;

@Component
public class JmsOrderReceiver implements OrderReceiver {

    private JmsTemplate jms;
    private MessageConverter converter;

    @Autowired
    public JmsOrderReceiver(JmsTemplate jms, MessageConverter converter) {
        this.jms = jms;
        this.converter = converter;
    }

    public Order receiveOrder() {
        Message message = jms.receive("tacocloud.order.queue");
        return (Order) converter.fromMessage(message);
    }
}

这里，使用了一个 String 来指定从何处拉取订单。receive() 方法返回一个未转换的 Message。但是真正需要的是 Message 中的 Order，所以接下来要做的就是使用注入的消息转换器来转换消息。消息中的类型 ID 属性将指导转换器将其转换为 Order，但是它是作为一个 Object 返回的，在返回它之前需要进行转换。

接收原始 Message 对象在某些需要检查消息属性和标题的情况下可能很有用，但是通常只需要有效载荷。将有效负载转换为域类型需要两个步骤，需要将消息转换器注入组件。当只关心消息的有效负载时，receiveAndConvert() 要简单得多。下面的程序清单显示了如何修改 JmsOrderReceiver 来使用 receiveAndConvert() 而不是 receive()。程序清单 8.3 接收已经转换的 Order 对象

package tacos.kitchen.messaging.jms;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

@Component
public class JmsOrderReceiver implements OrderReceiver {
    private JmsTemplate jms;

    @Autowired
    public JmsOrderReceiver(JmsTemplate jms) {
        this.jms = jms;
    }

    public Order receiveOrder() {
        return (Order) jms.receiveAndConvert("tacocloud.order.queue");
    }
}

这个新版本的 JmsOrderReceiver 有一个 receieveOrder() 方法，该方法已经减少到只有一行。不再需要注入 MessageConverter，因为所有的消息转换都将在 receiveAndConvert() 中完成。

在继续之前，让我们考虑一下如何在 Taco Cloud 厨房应用程序中使用 receiveOrder()。在 Taco Cloud 的一个厨房里，一名食品加工人员可能会按下一个按钮或采取一些行动，表示他们已经准备好开始制作 tacos 了。

此时，receiveOrder() 将被调用，而对 receive() 或 receiveAndConvert() 的调用将被阻塞。在订单消息准备好之前，不会发生任何其他事情。一旦订单到达，它将从 receiveOrder() 中返回，其信息用于显示订单的详细信息，以便食品加工人员开始工作。这似乎是拉模型的自然选择。

现在，让我们通过声明 JMS 监听器来了解推模型是如何工作的。

声明消息监听器

在拉模型中，接收消息需要显式调用 receive() 或 receiveAndConvert() 方法，与拉模型不同，消息监听器是一个被动组件，在消息到达之前是空闲的。

要创建对 JMS 消息作出响应的消息监听器，只需使用 @JmsListener 对组件中的方法进行注解。下面程序清单显示了一个新的 OrderListener 组件，它被动地监听消息，而不是主动地请求消息。程序清单 8.4 监听订单的 OrderListener 组件

package tacos.kitchen.messaging.jms.listener;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class OrderListener {

    private KitchenUI ui;

    @Autowired
    public OrderListener(KitchenUI ui) {
        this.ui = ui;
    }

    @JmsListener(destination = "tacocloud.order.queue")
    public void receiveOrder(Order order) {
        ui.displayOrder(order);
    }
}

receiveOrder() 方法由 JmsListener 注解，以监听 tacocloud.order.queue 目的地的消息。它不处理 JmsTemplate，也不被应用程序代码显式地调用。相反，Spring 中的框架代码将等待消息到达指定的目的地，当消息到达时，receiveOrder() 方法将自动调用，并将消息的 Order 有效负载作为参数。

在许多方面，@JmsListener 注解类似于 Spring MVC 的请求映射注释之一，比如 @GetMapping 或 @PostMapping。在 Spring MVC 中，用一个请求映射方法注解的方法对指定路径的请求做出响应。类似地，使用 @JmsListener 注解的方法对到达目的地的消息做出响应。

消息监听器通常被吹捧为最佳的选择，因为它们不会阻塞，并且能够快速处理多个消息。然而，在 Taco Cloud 应用程序的上下文中，它们可能不是最佳选择。食品加工是系统中的一个重要瓶颈，可能无法在接到订单时快速准备 taco。当一个新的订单显示在屏幕上时，食品加工者可能已经完成了一半的订单。厨房用户界面需要在订单到达时对其进行缓冲，以避免给厨房员工带来过重的负担。

这并不是说消息监听器不好。相反，当消息可以快速处理时，它们是完美的选择。但是，当消息处理程序需要能够根据自己的时间请求更多消息时，JmsTemplate 提供的拉模型似乎更合适。

因为 JMS 是由标准 Java 规范定义的，并且受到许多消息 Broker 的支持，所以它是 Java 中消息传递的常用选择。但是 JMS 有一些缺点，尤其是作为 Java 规范，它的使用仅限于 Java 应用程序。RabbitMQ 和 Kafka 等较新的消息传递选项解决了这些缺点，并且适用于 JVM 之外的其他语言和平台。让我们把 JMS 放在一边，看看如何使用 RabbitMQ 进行 taco 订单消息传递。