示例应用程序

在 Spring AMQP 示例项目中包含两个示例应用程序。第一个是一个简单的“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` 实例，默认情况下，它会查找任何类型为交换机、队列或绑定的 bean，然后在代理上声明它们。实际上，在 `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()` 方法，因此模板会委托给其 `MessageConverter` 实例。在这种情况下，它使用默认的 `SimpleMessageConverter`，但可以在 `HelloWorldConfiguration` 中定义的 `rabbitTemplate` bean 中提供不同的实现。

现在打开Consumer 类。它实际上共享相同的配置基类，这意味着它共享rabbitTemplate bean。这就是为什么我们使用routingKey（用于发送）和queue（用于接收）来配置该模板。正如我们在AmqpTemplate中描述的那样，您也可以将 'routingKey' 参数传递给发送方法，并将 'queue' 参数传递给接收方法。Consumer 代码基本上是 Producer 的镜像，调用receiveAndConvert() 而不是convertAndSend()。

以下清单显示了Consumer 的主方法

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。事实上，有一个子包提供了 exactly that: org.springframework.amqp.samples.helloworld.async。

同样，我们从发送方开始。打开ProducerConfiguration 类，注意它创建了一个connectionFactory 和一个rabbitTemplate bean。这次，由于配置专门用于消息发送方，我们甚至不需要任何队列定义，并且RabbitTemplate 只有 'routingKey' 属性设置。回想一下，消息被发送到交换机，而不是直接发送到队列。AMQP 默认交换机是一个没有名称的直接交换机。所有队列都使用它们的名称作为路由键绑定到该默认交换机。这就是为什么我们只需要在这里提供路由键。

以下清单显示了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 任务调度支持，您可以了解更多here。简而言之，ProducerConfiguration 中的postProcessor bean 将任务注册到调度程序。

现在我们可以转向接收方。为了强调消息驱动的 POJO 行为，我们从对消息做出反应的组件开始。该类名为HelloWorldHandler，如以下清单所示

public class HelloWorldHandler {

    public void handleMessage(String text) {
        System.out.println("Received: " + text);
    }

}

该类是一个 POJO。它没有扩展任何基类，也没有实现任何接口，甚至不包含任何导入。它通过 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 配置，因此这里我们将重点介绍使此示例与众不同的内容。有一个服务器将市场数据（股票报价）推送到主题交换机。然后，客户端可以通过将队列与路由模式（例如，`app.stock.quotes.nasdaq.*`）绑定来订阅市场数据提要。此演示的另一个主要功能是客户端发起的请求-回复“股票交易”交互，由服务器处理。这涉及一个私有的 `replyTo` 队列，该队列由客户端在订单请求消息本身中发送。

服务器的核心配置位于 `org.springframework.amqp.rabbit.stocks.config.server` 包中的 `RabbitServerConfiguration` 类中。它扩展了 `AbstractStockAppRabbitConfiguration`。这是定义服务器和客户端共有的资源的地方，包括市场数据主题交换机（其名称为 'app.stock.marketdata'）和服务器公开用于股票交易的队列（其名称为 'app.stock.request'）。在该通用配置文件中，您还会看到在 `RabbitTemplate` 上配置了 `Jackson2JsonMessageConverter`。

服务器特定的配置包括两件事。首先，它在 `RabbitTemplate` 上配置市场数据交换机，以便它不需要在每次调用发送 `Message` 时都提供该交换机名称。它是在基配置类中定义的抽象回调方法中完成的。以下清单显示了该方法

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 包。在那里，您可以看到提供 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();
}

这些是私有队列，并且唯一名称是自动生成的。第一个生成的队列由客户端用于绑定到服务器公开的市场数据交换。回想一下，在 AMQP 中，消费者与队列交互，而生产者与交换交互。队列与交换的“绑定”告诉代理将来自给定交换的消息传递（或路由）到队列。由于市场数据交换是主题交换，因此绑定可以用路由模式来表达。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。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 标头设置为完全限定的类名，以帮助接收应用程序将 JSON 转换回 Java 对象。

spring-rabbit-json 示例探讨了将非 Spring 应用程序的 JSON 转换回 Java 对象的几种技术。

另请参阅 [json-message-converter] 以及 DefaultClassMapper 的 Javadoc。