3.2.2 Log4j Appender

3.2.3 Log4j2 Appender

3.2.4 Logback Appender

3.2.5 定制Messages

简化 String 属性

Log4j和logback先进技术

3.3 样例应用程序

3.3.1 介绍

3.3.2 Hello World

介绍

在src/main/java 目录中，导航到org.springframework.amqp.helloworld 包中.

打开HelloWorldConfiguration 类，你可以注意到它包含了@Configuration 类级注解和一些@Bean 方法级注解. 

这是Spring 的基于Java的配置．你可进一步的了解here.

@Bean

public ConnectionFactory connectionFactory() {
    CachingConnectionFactory connectionFactory =null;
    connectionFactory =new CachingConnectionFactory("localhost");
    connectionFactory.setUsername("guest");
    connectionFactory.setPassword("guest");
    return connectionFactory;
}

配置中同样也包含了RabbitAdmin的实例, 它会默认查找类型为Exchange, Queue, 或 Binding的bean并在broker中进行声明.
事实上,"helloWorldQueue" bean是在HelloWorldConfiguration 中生成的，因为它是 Queue的实例.

@Bean

public Queue helloWorldQueue() {
    returnnew Queue(this.helloWorldQueueName);
}

重看"rabbitTemplate"bean配置,你会看到它将helloWorldQueue的名称设成了"queue"属性(用于接收消息) 以及"routingKey" 属性(用于发送消息).

publicstaticvoid main(String[] args) {

    ApplicationContext context =
        new AnnotationConfigApplicationContext(HelloWorldConfiguration.class);
    AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
    amqpTemplate.convertAndSend("Hello World");
    System.out.println("Sent: Hello World");
}

在上面的例子中你可以看到， 取回的AmqpTemplate用来发送消息.因为客户端代码应该尽可能地依赖于接口，因此类型是AmqpTemplate而不是RabbitTemplate.
即使在HelloWorldConfiguration中创建的bean是RabbitTemplate的实例,依赖于接口则意味着这端代码更具有便携性(portable) (配置可以独立于代码进行修改).
因为convertAndSend() 方法是通过模板来调用的，因此模板会将调用委派给它的MessageConverter实例.在这种情况下,它默认使用的是SimpleMessageConverter,但也可以在HelloWorldConfiguration中为"rabbitTemplate"指定其它的实现.

现在打开Consumer类. 它实际上共享了同一个配置基类，这意味着它将共享"rabbitTemplate" bean. 这就是为什么我们要使用"routingKey" (发送) 和"queue" (接收)来配置模板的原因.
正如你在Section 3.1.4, “AmqpTemplate”中看到的,你可以代替在发送方法中传递routingKey参数，代替在接收方法中传递queue 参数．  Consumer 代码基本上是Producer的镜子,只不过调用的是receiveAndConvert()而非convertAndSend()方法.

publicstaticvoid 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 样例就可以可以提供异步接收的示例了,又名 Message-driven POJOs. 事实上,有一个子包明确地提供了这种功能: org.springframework.amqp.samples.helloworld.async.

再一次地我们将从发送端开始. 打开ProducerConfiguration类可注意到它创建了一个"connectionFactory"和"rabbitTemplate" bean.
这次，由于配置是专用于消息发送端，因此我们不需要任何队列定义,RabbitTemplate只须设置routingKey属性.
回想一下，消息是发送到交换器上的而不是直接发到队列上的. AMQP默认交换器是无名称的direct类型交换器.
所有队列都是通过使用它们的名称作为路由键绑定到默认交换器上的.这就是为什么在这里我们只提供路由键的原因．

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

由于这个示例展示的是异步消息处理，生产方设计为连续发送消息(尽管类似于同步版本中的 message-per-execution模型,但不太明显，实际上它是消息驱动消费者)负责连续发送消息的组件是作为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 3.0 任务调度支持,你可从here这里来了解.
简短故事是：在 ProducerConfiguration 中的"postProcessor" bean使用调度器来注册了任务.

现在，让我们转向接收方. 为强调 Message-driven POJO 行为，将从对消息起反应的组件开始. 

此类被称为HelloWorldHandler.

publicclass HelloWorldHandler {

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

}

相当明显的, 这是一个POJO. 它没有继承任何基类,它没有实现任何接口,它甚至不包含任何导入. 它将通过Spring AMQP MessageListenerAdapter来适配MessageListener接口.然后适配器可配置在SimpleMessageListenerContainer上.
在这个例子中，容器是在ConsumerConfiguration类中创建的.你可以看到POJO是包装在适配器中的.

@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,你会看每秒就会发送消息和接收到消息．

Stock Trading 示例演示了比Hello World示例更高级的消息场景.然而，配置却是很相似的 - 只是有一点复杂.
由于我们已经详细讲解了Hello World配置,因此在这里我们将重点关注不一样的东西. 有一个服务器发送市场数据(股票报价)到Topic交换器中.
然后，客户端可订阅市场数据，即通过使用路由模式(如. "app.stock.quotes.nasdaq.*")来绑定队列(e.g. "app.stock.quotes.nasdaq.*").
这个例子的其它主要功能是 有一个请求回复“股票交易”的互动，它是由客户发起并由服务器来处理的. 这涉及到一个私有的“回复(replyTo)”队列，发送客户端的信息在请求消息中。

服务器的核心配置在RabbitServerConfiguration类中(位于 org.springframework.amqp.rabbit.stocks.config.server 包中).
它继承了 AbstractStockAppRabbitConfiguration. 这是服务器和客户端定义常用资源的地方，包括市场数据Topic交换器(其名称为app.stock.marketdata) 以及服务器公开股票交易的队列(其名称为app.stock.request).
在那个公共配置文件中,你会看到在RabbitTemplate上配置了一个JsonMessageConverter．

服务器特有配置由２部分组成.首先，它在RabbitTemplate上配置了市场数据交换器，这样在发送消息时，就不必提供交换器名称．它是通过基础配置类中的抽象回调方法中定义做到这一点的．

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

其次, 声明了股票请求队列.在这里，它不需要任何明确的绑定，因为它将以它自己的名称作为路由键来绑定到无名称的默认交换器上.正如先前提到的，AMQP规范定义了此种行为.

@Beanpublic Queue stockRequestQueue() {
    returnnew Queue(STOCK_REQUEST_QUEUE_NAME);
}

现在你已经看过了服务器的AMQP资源配置，导航到src/test/java目录下的org.springframework.amqp.rabbit.stocks包.在那里你会实际的 提供了main()方法的Server类．
它基于server-bootstrap.xml 创建了一个ApplicationContext.在那里，你会看到发布虚假市场数据的调度任务.

那个配置依赖于Spring 3.0的"task"命名空间支持.bootstrap配置文件也导入了其它一些文件.最令人关注的是位于src/main/resources目录下的server-messaging.xml.在那里，你会看到"messageListenerContainer" bean，它负责处理股票交易请求.
最后在看一下定义在src/main/resources目录下的server-handlers.xml，其中定义了一个 "serverHandler" bean.这个bean是ServerHandler类的实例，它是Message-driven POJO 的好例子，它也有发送回复消息的能力.
注意，它自身并没有与框架或任何AMQP概念耦合.它只是简单地接受TradeRequest并返回一个TradeResponse.

public TradeResponse handleMessage(TradeRequest tradeRequest) { ...
}

现在我们已经看了服务端的重要配置和代码，让我们转向客户端．最佳起点是从 org.springframework.amqp.rabbit.stocks.config.client 包下的RabbitClientConfiguration开始. 

注意，它声明了两个不带明确参数的队列．

@Bean

public Queue marketDataQueue() {
    return amqpAdmin().declareQueue();
}

@Bean

public Queue traderJoeQueue() {
    return amqpAdmin().declareQueue();
}

那些是私有队列, 唯一名称会自动自成.客户端会用第一个生成的队列来绑定由服务端公开的市场交换器.
记住在AMQP中,消费者与队列交互，而生产者与交换器交互. 队列和交换器之间的绑定指示broker从给定的交换器中投递或路由什么消息给队列.
由于市场交换器是一个Topic交换器,绑定可通过路由正则表达式来表达. 

RabbitClientConfiguration声明了一个Binding对象,其对象是通过BindingBuilder的便利API来生成的.

@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 注解来注入值.这通常是一个好主意，否则值就会硬编码在类中，没有修改就没有重新编译.

在这种情况下，通过修改绑定中的路由正则表达式，可很容易地运行多个版本的Client．让我们立即尝试．

启动运行org.springframework.amqp.rabbit.stocks.Server然后再运行 org.springframework.amqp.rabbit.stocks.Client.你将会看到NASDAQ股票的交易报价，因为关联stocks.quote.pattern 键的值在client.properties中是app.stock.quotes.nasdaq.
现在，保持现有Server 和Client 运行，将其属性值修改为app.stock.quotes.nyse.再启动第二个Client实例.你会看到第一个client仍然接收NASDAQ 报价，而第二个client接收的NYSE报价. 你可以改变模式，获取所有的股票报价或个别股票的报价。

最后一个我们将暴露的特性是从客户端的角度来看待请求－回复交互.记住我们已经看了ServerHandler，它会接受TradeRequest对象并返回TradeResponse对象. 客户端相应的代码是 RabbitStockServiceGateway（位于org.springframework.amqp.rabbit.stocks.gateway 包).为发送消息，它会委派给RabbitTemplate.

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) {
                thrownew 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的格式来发送请求．经过一个简短的人工延迟来模拟“处理”请求，你应该看到一个确认消息出现在客户端上。

3.4 测试支持

3.4.1 介绍

第一个, LatchCountDownAndCallRealMethodAnswer 提供了返回null和计数下一个锁存器的Answer<Void>.

LatchCountDownAndCallRealMethodAnswer answer = new LatchCountDownAndCallRealMethodAnswer(2);
doAnswer(answer)
    .when(listener).foo(anyString(), anyString());

...

assertTrue(answer.getLatch().await(10, TimeUnit.SECONDS));

第二个， LambdaAnswer<T> 提供了一种调用真正方法的机制，并提供机会来返回定制结果（基于InvocationOnMock和结果）.

public　class Foo {

    public String foo(String foo) {
        return foo.toUpperCase();
    }

}

Foo foo = spy(new Foo());

doAnswer(new LambdaAnswer<String>(true, (i, r) -> r + r))
    .when(foo).foo(anyString());
assertEquals("FOOFOO", foo.foo("foo"));

doAnswer(new LambdaAnswer<String>(true, (i, r) -> r + i.getArguments()[0]))
    .when(foo).foo(anyString());
assertEquals("FOOfoo", foo.foo("foo"));

doAnswer(new LambdaAnswer<String>(false, (i, r) ->
    "" + i.getArguments()[0] + i.getArguments()[0])).when(foo).foo(anyString());
assertEquals("foofoo", foo.foo("foo"));

When using Java 7 or earlier:

doAnswer(new LambdaAnswer<String>(true, new ValueToReturn<String>() {
    @Overridepublic String apply(InvocationOnMock i, String r) {
        return r + r;
    }
})).when(foo).foo(anyString());

在你的@Configuration 类中使用 @RabbitListenerTest (它也会通过@EnableRabbit来启用@RabbitListener 探测).注解会导致框架使用子类RabbitListenerTestHarness来代替标准RabbitListenerAnnotationBeanPostProcessor．

RabbitListenerTestHarness 通过两种方式来增强监听器 - 将其包装进Mockito Spy, 启用了Mockito 存根和验证操作．也可在监听器添加Advice 来启用对参数，结果或异常的访问.
你可以控制哪一个(或两个)来在@RabbitListenerTest上启用属性. 后者用于访问调用中更为低级数据- 它也支持测试线程阻塞，直到异步监听器被调用.

重要

final @RabbitListener 不能被发现或通知 ，同时，只有带id属性的监听器才能发现或通知．

让我们看一些例子.

使用spy:

@Configuration

@RabbitListenerTest

public　class Config {

    @Bean

　public Listener listener() {
        returnnew Listener();
    }

    ...

}

public　class Listener {

    @RabbitListener(id="foo", queues="#{queue1.name}")

　public String foo(String foo) {
        return foo.toUpperCase();
    }

    @RabbitListener(id="bar", queues="#{queue2.name}")

　public　void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
        ...
    }

}

public　class MyTests {

    @Autowired

    private RabbitListenerTestHarness harness;

@Test

　public　void testTwoWay() throws Exception {
        assertEquals("FOO", this.rabbitTemplate.convertSendAndReceive(this.queue1.getName(), "foo"    ));

        Listener listener = this.harness.getSpy("foo"); 
        assertNotNull(listener);
        verify(listener).foo("foo");
    }

    @Test

public　void testOneWay() throws Exception {
        Listener listener = this.harness.getSpy("bar");
        assertNotNull(listener);

        LatchCountDownAndCallRealMethodAnswer answer = new LatchCountDownAndCallRealMethodAnswer(2); 
        doAnswer(answer).when(listener).foo(anyString(), anyString());

　　this.rabbitTemplate.convertAndSend(this.queue2.getName(), "bar");

　　this.rabbitTemplate.convertAndSend(this.queue2.getName(), "baz");

        assertTrue(answer.getLatch().await(10, TimeUnit.SECONDS));
        verify(listener).foo("bar", this.queue2.getName());
        verify(listener).foo("baz", this.queue2.getName());
    }

}

@Configuration

@ComponentScan

@RabbitListenerTest(spy = false, capture = true)

public　class Config {

}

@Service

public　class Listener {

    private　boolean failed;

    @RabbitListener(id="foo", queues="#{queue1.name}")

　public String foo(String foo) {
        return foo.toUpperCase();
    }

    @RabbitListener(id="bar", queues="#{queue2.name}")

public　void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
        if (!failed && foo.equals("ex")) {
            failed = true;
            thrownew RuntimeException(foo);
        }
        failed = false;
    }

}

public　class MyTests {

    　@Autowired

　　private RabbitListenerTestHarness harness; 

　　@Test

　　public　void testTwoWay() throws Exception {
        assertEquals("FOO", this.rabbitTemplate.convertSendAndReceive(this.queue1.getName(), "foo"　　));

        InvocationData invocationData =
            this.harness.getNextInvocationDataFor("foo", 0, TimeUnit.SECONDS); 
        assertThat(invocationData.getArguments()[0], equalTo("foo"));     
        assertThat((String) invocationData.getResult(), equalTo("FOO"));
    }

    @Test

　　public　void testOneWay() throws Exception {
        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "bar");
        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "baz");
        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "ex");

        InvocationData invocationData =
            this.harness.getNextInvocationDataFor("bar", 10, TimeUnit.SECONDS); 
        Object[] args = invocationData.getArguments();
        assertThat((String) args[0], equalTo("bar"));
        assertThat((String) args[1], equalTo(queue2.getName()));

        invocationData = this.harness.getNextInvocationDataFor("bar", 10, TimeUnit.SECONDS);
        args = invocationData.getArguments();
        assertThat((String) args[0], equalTo("baz"));

        invocationData = this.harness.getNextInvocationDataFor("bar", 10, TimeUnit.SECONDS);
        args = invocationData.getArguments();
        assertThat((String) args[0], equalTo("ex"));
        assertEquals("ex", invocationData.getThrowable().getMessage()); 
    }

}

这部分参考文档提供了在Spring集成项目中提供AMQP支持的快速介绍．

Spring Integration 项目包含了构建于Spring AMQP项目之上的AMQP 通道适配器(Channel Adapters)和网关(Gateways). 那些适配器是在Spring集成项目中开发和发布的.在Spring 集成中, "通道适配器" 是单向的,而网关是双向的(请求－响应).
我们提供了入站通道适配器(inbound-channel-adapter),出站通道适配器( outbound-channel-adapter), 入站网关(inbound-gateway),以及出站网关(outbound-gateway).

由于AMQP 适配器只是Spring集成版本的一部分，因为文档也只针对Spring集成发行版本部分可用. 

<amqp:inbound-channel-adapter　channel="fromAMQP"　queue-names="some.queue"　connection-factory="rabbitConnectionFactory"/>

<amqp:outbound-channel-adapter　channel="toAMQP"　exchange-name="some.exchange"　　　　routing-key="foo"　amqp-template="rabbitTemplate"/>

<amqp:inbound-gateway　request-channel="fromAMQP"　reply-channel="toAMQP"　　　　　　　　　queue-names="some.queue"　connection-factory="rabbitConnectionFactory"/>

<amqp:outbound-gateway　request-channel="toAMQP"　reply-channel="fromAMQP"　　　　　　　exchange-name="some.exchange"　routing-key="foo"　amqp-template="rabbitTemplate"/>

除了这份参考文档，还有其它资源可帮助你了解AMQP.