示例应用
Spring AMQP Samples 项目包含两个示例应用。第一个是简单的“Hello World”示例，演示了同步和异步消息接收。它为理解基本组件提供了一个很好的起点。第二个示例基于一个股票交易用例，用于演示现实世界应用中常见的交互类型。在本章中，我们将快速介绍每个示例，以便您可以重点关注最重要的组件。这两个示例都是基于 Maven 的，因此您可以直接将它们导入到任何支持 Maven 的 IDE（例如 SpringSource Tool Suite）。
“Hello World”示例
“Hello World”示例演示了同步和异步消息接收。您可以将 spring-rabbit-helloworld 示例导入到 IDE 中，然后按照下面的讨论进行操作。
同步示例
在 src/main/java 目录中，导航到 org.springframework.amqp.helloworld 包。打开 HelloWorldConfiguration 类，注意它在类级别包含 @Configuration 注解，并在方法级别包含一些 @Bean 注解。这是 Spring 基于 Java 配置的一个示例。您可以在此处阅读更多相关内容。
以下列表显示了连接工厂是如何创建的
@Bean
public CachingConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory =
new CachingConnectionFactory("localhost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
return connectionFactory;
}该配置还包含一个 RabbitAdmin 实例，它默认会查找类型为 exchange、queue 或 binding 的任何 bean，然后将它们声明在 broker 上。实际上，在 HelloWorldConfiguration 中生成的 helloWorldQueue bean 就是一个例子，因为它是一个 Queue 实例。
以下列表显示了 helloWorldQueue bean 的定义
@Bean
public Queue helloWorldQueue() {
return new Queue(this.helloWorldQueueName);
}回顾 rabbitTemplate bean 的配置，您可以看到它将 helloWorldQueue 的名称设置为其 queue 属性（用于接收消息）和 routingKey 属性（用于发送消息）。
现在我们已经探索了配置，接下来可以看看实际使用这些组件的代码。首先，打开同一个包中的 Producer 类。它包含一个 main() 方法，在该方法中创建 Spring ApplicationContext。
以下列表显示了 main 方法
public static void main(String[] args) {
ApplicationContext context =
new AnnotationConfigApplicationContext(RabbitConfiguration.class);
AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
amqpTemplate.convertAndSend("Hello World");
System.out.println("Sent: Hello World");
}在前面的示例中，检索并使用 AmqpTemplate bean 发送消息。由于客户端代码应尽可能依赖接口，因此类型是 AmqpTemplate 而不是 RabbitTemplate。即使在 HelloWorldConfiguration 中创建的 bean 是 RabbitTemplate 的实例，依赖接口意味着此代码更具可移植性（您可以在不更改代码的情况下独立更改配置）。由于调用了 convertAndSend() 方法，模板会委托给其 MessageConverter 实例。在本例中，它使用了默认的 SimpleMessageConverter，但可以向 HelloWorldConfiguration 中定义的 rabbitTemplate bean 提供不同的实现。
现在打开 Consumer 类。它实际上共享相同的配置基类，这意味着它共享 rabbitTemplate bean。这就是为什么我们将该模板同时配置了 routingKey（用于发送）和 queue（用于接收）。正如我们在 AmqpTemplate 中所述，您也可以将“routingKey”参数传递给发送方法，将“queue”参数传递给接收方法。Consumer 代码基本上是 Producer 的镜像，调用 receiveAndConvert() 而不是 convertAndSend()。
以下列表显示了 Consumer 的 main 方法
public static void main(String[] args) {
ApplicationContext context =
new AnnotationConfigApplicationContext(RabbitConfiguration.class);
AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
System.out.println("Received: " + amqpTemplate.receiveAndConvert());
}如果您运行 Producer，然后运行 Consumer，您应该在控制台输出中看到 Received: Hello World。
异步示例
同步示例 部分详细介绍了同步的 Hello World 示例。本节将介绍一个稍微更高级但功能强大得多的选项。通过一些修改，Hello World 示例可以提供一个异步接收的示例，也称为消息驱动的 POJO。事实上，有一个子包正好提供了这个功能：org.springframework.amqp.samples.helloworld.async。
同样，我们从发送端开始。打开 ProducerConfiguration 类，注意它创建了一个 connectionFactory 和一个 rabbitTemplate bean。这次，由于配置专用于消息发送端，我们甚至不需要任何队列定义，并且 RabbitTemplate 只设置了 routingKey 属性。回想一下，消息是发送到交换机而不是直接发送到队列。AMQP 默认交换机是一个没有名称的 direct exchange。所有队列都绑定到该默认交换机，并以其名称作为路由键。这就是为什么我们只需要在此处提供路由键的原因。
以下列表显示了 rabbitTemplate 的定义
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
template.setRoutingKey(this.helloWorldQueueName);
return template;
}由于此示例演示了异步消息接收，发送端被设计为持续发送消息（如果它像同步版本那样是每执行一次发送一条消息的模型，那么它实际上是一个消息驱动消费者的事实就不会那么明显了）。负责持续发送消息的组件在 ProducerConfiguration 中定义为一个内部类。它被配置为每三秒运行一次。
以下列表显示了该组件
static class ScheduledProducer {
@Autowired
private volatile RabbitTemplate rabbitTemplate;
private final AtomicInteger counter = new AtomicInteger();
@Scheduled(fixedRate = 3000)
public void sendMessage() {
rabbitTemplate.convertAndSend("Hello World " + counter.incrementAndGet());
}
}您不需要理解所有细节，因为真正的重点应该放在接收端（我们将在下一节介绍）。但是，如果您还不熟悉 Spring 任务调度支持，可以在此处了解更多信息。简而言之，ProducerConfiguration 中的 postProcessor bean 会将任务注册到调度器中。
现在我们可以转向接收端。为了强调消息驱动的 POJO 行为，我们从响应消息的组件开始。该类名为 HelloWorldHandler，如下所示
public class HelloWorldHandler {
public void handleMessage(String text) {
System.out.println("Received: " + text);
}
}该类是一个 POJO。它不继承任何基类，不实现任何接口，甚至不包含任何 import。它通过 Spring AMQP 的 MessageListenerAdapter 被“适配”到 MessageListener 接口。然后，您可以在 SimpleMessageListenerContainer 上配置该适配器。对于此示例，容器是在 ConsumerConfiguration 类中创建的。您可以在那里看到 POJO 被封装在适配器中。
以下列表显示了 listenerContainer 的定义方式
@Bean
public SimpleMessageListenerContainer listenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueueName(this.helloWorldQueueName);
container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler()));
return container;
}SimpleMessageListenerContainer 是一个 Spring 生命周期组件，默认情况下会自动启动。如果您查看 Consumer 类，可以看到其 main() 方法只不过是一行用于创建 ApplicationContext 的引导代码。Producer 的 main() 方法也是一行引导代码，因为其方法带有 @Scheduled 注解的组件也会自动启动。您可以按任意顺序启动 Producer 和 Consumer，并且应该会看到每三秒发送和接收一次消息。
股票交易
股票交易示例演示了比Hello World 示例 更高级的消息场景。然而，配置非常相似，只是稍微复杂一些。由于我们已经详细介绍了 Hello World 配置，这里我们将重点关注此示例的不同之处。有一个服务器将市场数据（股票报价）推送到一个 topic exchange。然后，客户端可以通过绑定一个带有路由模式（例如，app.stock.quotes.nasdaq.*）的队列来订阅市场数据馈送。此演示的另一个主要特性是由客户端发起并由服务器处理的请求-回复“股票交易”交互。这涉及客户端在订单请求消息本身中发送的一个私有 replyTo 队列。
服务器的核心配置位于 org.springframework.amqp.rabbit.stocks.config.server 包中的 RabbitServerConfiguration 类中。它继承了 AbstractStockAppRabbitConfiguration。这里定义了服务器和客户端共有的资源，包括市场数据 topic exchange（其名称为 app.stock.marketdata）以及服务器为股票交易暴露的队列（其名称为 app.stock.request）。在该通用配置文件中，您还可以看到在 RabbitTemplate 上配置了一个 Jackson2JsonMessageConverter。
服务器特定的配置包含两部分。首先，它在 RabbitTemplate 上配置市场数据交换机，以便在每次调用发送消息时无需提供该交换机名称。它在基配置类中定义的一个抽象回调方法中完成此操作。以下列表显示了该方法
public void configureRabbitTemplate(RabbitTemplate rabbitTemplate) {
rabbitTemplate.setExchange(MARKET_DATA_EXCHANGE_NAME);
}其次，声明了股票请求队列。在这种情况下，它不需要任何显式绑定，因为它绑定到默认的无名交换机，并以其自己的名称作为路由键。如前所述，AMQP 规范定义了此行为。以下列表显示了 stockRequestQueue bean 的定义
@Bean
public Queue stockRequestQueue() {
return new Queue(STOCK_REQUEST_QUEUE_NAME);
}现在您已经看到了服务器 AMQP 资源的配置，请导航到 src/test/java 目录下的 org.springframework.amqp.rabbit.stocks 包。在那里，您可以看到实际的 Server 类，它提供了一个 main() 方法。它基于 server-bootstrap.xml 配置文件创建 ApplicationContext。在那里，您可以看到发布模拟市场数据的计划任务。该配置依赖于 Spring 的 task 命名空间支持。引导配置文件还导入了其他一些文件。最有趣的是 server-messaging.xml，它直接位于 src/main/resources 下。在那里，您可以看到负责处理股票交易请求的 messageListenerContainer bean。最后，看看在 server-handlers.xml（也在 src/main/resources 中）中定义的 serverHandler bean。该 bean 是 ServerHandler 类的一个实例，是消息驱动 POJO 的一个很好的例子，它也可以发送回复消息。请注意，它本身并不与框架或任何 AMQP 概念耦合。它接受一个 TradeRequest 并返回一个 TradeResponse。以下列表显示了 handleMessage 方法的定义
public TradeResponse handleMessage(TradeRequest tradeRequest) { ...
}现在我们已经看到了服务器最重要的配置和代码，可以转向客户端了。最好的起点可能是在 org.springframework.amqp.rabbit.stocks.config.client 包中的 RabbitClientConfiguration。注意它声明了两个队列，但没有提供显式名称。以下列表显示了这两个队列的 bean 定义
@Bean
public Queue marketDataQueue() {
return amqpAdmin().declareQueue();
}
@Bean
public Queue traderJoeQueue() {
return amqpAdmin().declareQueue();
}这些是私有队列，会自动生成唯一的名称。第一个生成的队列被客户端用来绑定到服务器暴露的市场数据交换机。回想一下，在 AMQP 中，消费者与队列交互，而生产者与交换机交互。队列与交换机的“绑定”就是告诉 broker 将来自给定交换机的消息发送（或路由）到队列。由于市场数据交换机是一个 topic exchange，绑定可以用路由模式表示。RabbitClientConfiguration 使用 Binding 对象完成了这一点，并且该对象是使用 BindingBuilder 流式 API 生成的。以下列表显示了 Binding
@Value("${stocks.quote.pattern}")
private String marketDataRoutingKey;
@Bean
public Binding marketDataBinding() {
return BindingBuilder.bind(
marketDataQueue()).to(marketDataExchange()).with(marketDataRoutingKey);
}注意，实际值已外部化到属性文件（src/main/resources 下的 client.properties）中，并且我们使用 Spring 的 @Value 注解来注入该值。这通常是一个好主意。否则，该值将被硬编码在类中，并且在不重新编译的情况下无法修改。在这种情况下，更容易在更改用于绑定的路由模式时运行客户端的多个版本。我们现在可以尝试一下。
首先运行 org.springframework.amqp.rabbit.stocks.Server，然后运行 org.springframework.amqp.rabbit.stocks.Client。您应该会看到 NASDAQ 股票的模拟报价，因为 client.properties 中与 stocks.quote.pattern 键关联的当前值是 app.stock.quotes.nasdaq.*。现在，在保持现有 Server 和 Client 运行的同时，将该属性值更改为 app.stock.quotes.nyse.* 并启动第二个 Client 实例。您应该看到第一个客户端仍在接收 NASDAQ 报价，而第二个客户端接收 NYSE 报价。您也可以更改模式以获取所有股票，甚至是单个股票代码。
我们将探讨的最后一个特性是客户端角度的请求-回复交互。回想一下，我们已经看到了接受 TradeRequest 对象并返回 TradeResponse 对象的 ServerHandler。客户端对应的代码是 org.springframework.amqp.rabbit.stocks.gateway 包中的 RabbitStockServiceGateway。它委托给 RabbitTemplate 来发送消息。以下列表显示了 send 方法
public void send(TradeRequest tradeRequest) {
getRabbitTemplate().convertAndSend(tradeRequest, new MessagePostProcessor() {
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setReplyTo(new Address(defaultReplyToQueue));
try {
message.getMessageProperties().setCorrelationId(
UUID.randomUUID().toString().getBytes("UTF-8"));
}
catch (UnsupportedEncodingException e) {
throw new AmqpException(e);
}
return message;
}
});
}注意，在发送消息之前，它设置了 replyTo 地址。它提供了由 traderJoeQueue bean 定义（前面已介绍）生成的队列。以下列表显示了 StockServiceGateway 类本身的 @Bean 定义
@Bean
public StockServiceGateway stockServiceGateway() {
RabbitStockServiceGateway gateway = new RabbitStockServiceGateway();
gateway.setRabbitTemplate(rabbitTemplate());
gateway.setDefaultReplyToQueue(traderJoeQueue());
return gateway;
}如果您没有运行服务器和客户端，请立即启动它们。尝试发送格式为“100 TCKR”的请求。在短暂的模拟请求“处理”的人工延迟之后，您应该在客户端看到确认消息。
