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 列出了最有用的属性。
属性 | 描述 |
---|---|
spring.artemis.host | broker 主机 |
spring.artemis.port | broker 端口 |
spring.artemis.user | 用于访问 broker 的用户(可选) |
spring.artemis.password | 用于访问 broker 的密码(可选) |
例如,考虑应用程序中的以下条目。可能用于非开发设置的 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 特定的属性。
属性 | 描述 |
---|---|
spring.activemq.broker-url | Broker 的 URL |
spring.activemq.user | 用于访问 Broker 的用户(可选) |
spring.activemq.password | 用于访问 Broker 的密码(可选) |
spring.activemq.in-memory | 是否启动内存 Broker(默认:true) |
请注意,不是为 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 中描述的那样。
消息转换器 | 做了什么 |
---|---|
MappingJackson2MessageConverter | 使用 Jackson 2 JSON 库对消息进行与 JSON 的转换 |
MarshallingMessageConverter | 使用 JAXB 对消息进行与 XML 的转换 |
MessagingMessageConverter | 使用底层 MessageConverter(用于有效负载)和JmsHeaderMapper(用于将 Jms 信息头映射到标准消息标头)将 Message 从消息传递抽象转换为 Message,并从 Message 转换为 Message |
SimpleMessageConverter | 将 String 转换为 TextMessage,将字节数组转换为 BytesMessage,将 Map 转换为 MapMessage,将Serializable 转换为 ObjectMessage |
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 订单消息传递。