示例应用

Doubao 中英对照 Sample Applications

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、队列或 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 并用于发送一条 Message。由于客户端代码应尽可能依赖接口，类型是 AmqpTemplate 而非 RabbitTemplate。即使在 HelloWorldConfiguration 中创建的 bean 是 RabbitTemplate 的实例，依赖接口也意味着这段代码更可移植（你可以独立于代码更改配置）。由于调用了 convertAndSend() 方法，template 委托给其 MessageConverter 实例。这里使用默认的 SimpleMessageConverter，但也可以向 HelloWorldConfiguration 中定义的 rabbitTemplate bean 提供不同的实现。

现在打开 Consumer 类。它实际上共享同一个配置基类，这意味着它共享 rabbitTemplate bean。这就是为什么我们为该 template 同时配置了 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' 属性。回想一下，消息是发送到 exchange 而非直接发送到队列。AMQP 默认 exchange 是一个没有名称的 direct exchange。所有队列都以其名称作为 routing key 绑定到该默认 exchange。这就是为什么我们这里只需提供 routing key。

以下清单展示 rabbitTemplate 定义：

public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory());
    template.setRoutingKey(this.helloWorldQueueName);
    return template;
}

由于本示例演示异步消息接收，producer 端被设计为持续发送消息（如果它像同步版本那样是每次执行一条消息的模型，就不会那么明显地体现出它实际上是一个消息驱动 consumer）。负责持续发送消息的组件被定义为 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 上配置该 adapter。对于本示例，容器在 ConsumerConfiguration 类中创建。你可以在那里看到被包装在 adapter 中的 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。然后客户端可以用 routing pattern 绑定一个队列来订阅行情数据流（例如 app.stock.quotes.nasdaq.*）。本演示的另一个主要特性是由客户端发起、服务器处理的请求-回复“股票交易”交互。这涉及一个私有的 replyTo 队列，由客户端在订单请求消息本身中发送。

服务器的核心配置在 org.springframework.amqp.rabbit.stocks.config.server 包中的 RabbitServerConfiguration 类里。它扩展了 AbstractStockAppRabbitConfiguration。服务器和客户端共用的资源就在那里定义，包括行情 topic exchange（名称为 'app.stock.marketdata'）以及服务器为股票交易暴露的队列（名称为 'app.stock.request'）。在那个公共配置文件中，你还会看到在 RabbitTemplate 上配置了 JacksonJsonMessageConverter。

服务器专属配置由两部分组成。首先，它在 RabbitTemplate 上配置行情 exchange，这样就不必在每次发送 Message 的调用时都提供该 exchange 名称。它在基配置类中定义的一个抽象回调方法内完成此配置。以下清单展示该方法：

public void configureRabbitTemplate(RabbitTemplate rabbitTemplate) {
    rabbitTemplate.setExchange(MARKET_DATA_EXCHANGE_NAME);
}

其次，声明股票请求队列。这种情况下它不需要任何显式 binding，因为它以自己的名称作为 routing key 绑定到默认的无名称 exchange。如前所述，AMQP 规范定义了该行为。以下清单展示 stockRequestQueue bean 的定义：

@Bean
public Queue stockRequestQueue() {
    return new Queue(STOCK_REQUEST_QUEUE_NAME);
}

看过服务器 AMQP 资源的配置后，导航到 src/test/java 目录下的 org.springframework.amqp.rabbit.stocks 包。在那里你可以看到提供 main() 方法的实际 Server 类。它基于 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();
}

这些是私有队列，会自动生成唯一名称。第一个生成的队列由客户端用于绑定到服务器暴露的行情 exchange。回想一下，在 AMQP 中，consumer 与队列交互，而 producer 与 exchange 交互。队列到 exchange 的“绑定”告诉 broker 把来自某个 exchange 的消息投递（或路由）到队列。由于行情 exchange 是 topic exchange，绑定可以用 routing pattern 表达。RabbitClientConfiguration 用一个 Binding 对象来做到这一点，该对象由 BindingBuilder fluent 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 注解注入该值。这通常是个好主意。否则，该值会硬编码在类中，不重新编译就无法修改。这样一来，我们就可以在更改绑定所用 routing pattern 的同时运行多个客户端版本。现在我们可以试一下。

先运行 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 报价。你也可以改为更改 pattern 以获取所有股票，甚至某个单独的股票代码。

我们探索的最后一个特性是从客户端角度看的请求-回复交互。回想一下，我们已经见过接收 TradeRequest 对象、返回 TradeResponse 对象的 ServerHandler。Client 端相应的代码是 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' 的请求。在模拟请求“处理”的短暂人为延迟之后，你应该会在客户端看到一条确认消息。

从非 Spring 应用接收 JSON

Spring 应用在发送 JSON 时，会把 _TypeId_ header 设为全限定类名，以帮助接收应用把 JSON 转换回 Java 对象。

spring-rabbit-json 示例探索了几种转换来自非 Spring 应用的 JSON 的技术。

另见 Jackson2JsonMessageConverter 以及 DefaultClassMapper 的 Javadoc。