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.