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订单消息传递。