Spring AMQP

Spring AMQP 是基于 Spring 框架的AMQP消息解决方案，提供模板化的发送和接收消息的抽象层，提供基于消息驱动的 POJO的消息监听等，很大方便我们使用RabbitMQ程序的相关开发。

Spring AMQP包含一些模块，如：spring-amqp, spring-rabbit and spring-erlang等，每个模块分别由独立的一些Jar包组成.

Spring AMQP模块主要包含org.springframework.amqp.core这个包中。这个包定义的相关类主要是与前面讲的AMQP模型相对应。Spring AMQP的目的是提供不依赖于任何特定的AMQP代理实现或客户端库通用的抽象。最终用户代码将很容易实现更易替换、添加和删除AMQP，因为它可以只针对抽象层来开发。这可以很方便我们选择和使用哪一个具体的broker实现，如sping-rabbit实现。

Message

Spring AMQP定义了Message类, 它定义了一个更一般的AMQP域模型,它是Spring AMQP的很重要的一部分，Message消息是当前模型中所操纵的基本单位，它由Producer产生，经过Broker被Consumer所消费。它是生产者和消费者发送和处理的对象。Message封装了body及MessageProperties，很方便了API的调用，它的定义如下:

public class Message {

　　private final MessageProperties messageProperties;
　　private final byte[] body;
 
　　public Message(byte[] body, MessageProperties messageProperties) {
　　　　this.body = body;
　　　　this.messageProperties = messageProperties;
　　}
 
　　public byte[] getBody() {
　　　　return this.body;
　　}
 
　　public MessageProperties getMessageProperties() {
　　　　return this.messageProperties;
　　}
}

MessageProperties 定义了多种常用的属性，如：messageId', 'timestamp', 'contentType'等等。

Exchange

Exchange接口代表了AMQP Exchange，它是消息发送的地方。在虚拟主机的消息协商器(Broker)中，每个Exchange都有唯一的名字。

Exchange包含4种类型：Direct, Topic, Fanout, Headers。不同的类型，他们如何处理绑定到队列方面的行为会有所不同。

1)Direct类型: 允许一个队列通过一个固定的Routing-key(通常是队列的名字)进行绑定。 Direct交换器将消息根据其routing-key属性投递到包含对应key属性的绑定器上。

2)Topic类型: 支持消息的Routing-key用*或#的模式，进行绑定。*匹配一个单词，#匹配0个或者多个单词。例如，binding key *.user.# 匹配routing key为 usd.user和eur.user.db，但是不匹配user.hello。

3)Fanout类型：它只是将消息广播到所有绑定到它的队列中,而不考虑routing key的值。

4)Header类型： 它根据应用程序消息的特定属性进行匹配，这些消息可能在binding key中标记为可选或者必选

注意：AMQP规范要求，任何的消息协商器(broker)需要提供一个没有名称的"default" Direct Exchange。

它的相关定义如下：

public interface Exchange {
　　String getName();
　　String getExchangeType();
　　boolean isDurable();
　　boolean isAutoDelete();
　　Map getArguments();
}

Queue

Queue,队列，它代表了Message Consumer接收消息的地方，它用来保存消息直到发送给消费者。Queue有以下一些重要的属性。

1)持久性：如果启用，队列将会在消息协商器(Broker)重启前都有效。

2)自动删除：如果启用，那么队列将会在所有的消费者停止使用之后自动删除掉自身。

3)惰性：如果没有声明队列，那么在执行到使用的时候会导致异常，并不会主动声明。

4)排他性：如果启用，队列只能被声明它的消费者使用。

public class Queue  {
 
    private final String name;
    private volatile boolean durable;
    private volatile boolean exclusive;
    private volatile boolean autoDelete;
    private volatile Map arguments;
 
    public Queue(String name) {
    this(name, true, false, false);
    }
 
    // Getters and Setters omitted for brevity    

Binding

生产者发送消息到Exchange,接收者从Queue接收消息，而绑定(Binging)是生产者和消费者消息传递的重要连接，它是连接生产者和消费者进行信息交流的关键。

Binging实例本身仅仅是代表持有连接的数据信息。不过它可以被AmqpAdmin这个类用来实际触发broker上的绑定操作。同时它可以在程序启动时，简化 Queues, Exchanges, and Bindings的定义及一些操作。

1、下面我们看一下绑定Queue到Exchange的一些基本选项及例子。

你可以用一个固定的RoutingKey将Queue绑定到DirectExchange 

例如：new Binding(someQueue, someDirectExchange, "foo.bar")

2、也可以用*、#模式匹配的RoutingKey方式,绑定到TopicExchange。

例如:  new Binding(someQueue, someTopicExchange, "foo.*")

3、绑定到一个广播的FanoutExchange,它没有RoutingKey

例如:  new Binding(someQueue, someFanoutExchange)

4、当然你也可以用fluent API的方式。

Binding b = BindingBuilder.bind(someQueue).to(someTopicExchange).with("foo.*");

ConnectionFactory

Spring AMQP中连接和资源的管理，尤其是spring-rabbit这个模块，因为spring-rabbit是 RabbitMQ 的唯一支持的实现。

在spring-rabbit中，管理消息协商器(broker)连接的核心组件是ConnectionFactory这个接口。 ConnectionFactory提供了

org.springframework.amqp.rabbit.connection.Connection(com.rabbitmq.client.Connection的包装类)实例的连接与管理。而CachingConnectionFactory是ConnectionFactory的在Spring AMQP中唯一实现，它创建一个连接代理，使程序可以共享的连接。

Connection 提供一个createChannel的方法。CachingConnectionFactory 的实现能支持channels的缓存,并且能根据区分是事务性或非事务性各自独立。同时，CachingConnectionFactory也提供hostname的构造函数，并且可以设置username、password、setChannelCacheSize等方法。CachingConnectionFactory  默认channel cache 大小为1，如果想改变可以用setChannelCacheSize设置channel cache size的大小。

@Bean
public ConnectionFactory rabbitConnectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory(
            environment.getProperty("rabbitmq.host"),
            environment.getProperty("rabbitmq.port", Integer.class)
    );
    factory.setUsername(environment.getProperty("rabbitmq.username"));
    factory.setPassword(environment.getProperty("rabbitmq.password"));
    return factory;
}
 Connection connection = factory.createConnection();

Routing Connection Factory

从spring-rabbit 1.3版本开始，AbstractRoutingConnectionFactory 被引入进来，它提供了一个这样的途径来配置许多的不同的Connection Factory的映射，并且能够根据运行时的lookupKey(通过绑定线程上下文的方式) 来决定使用哪个具体的Connection Factory。

 为了方便Spring AMQP提供了 AbstractRoutingConnectionFactory 的具体实现SimpleRoutingConnectionFactory。它是从SimpleResourceHolder中获得当前线程绑定的lookupKey。

    Map factories = new HashMap<Object, org.springframework.amqp.rabbit.connection.ConnectionFactory>(2);
    factories.put("foo", connectionFactory1);
    factories.put("bar", connectionFactory2);

    AbstractRoutingConnectionFactory connectionFactory = new SimpleRoutingConnectionFactory();
    connectionFactory.setTargetConnectionFactories(factories);

    final RabbitTemplate template = new RabbitTemplate(connectionFactory);
    Expression expression = new SpelExpressionParser()
        .parseExpression("T(org.springframework.amqp.rabbit.core.RabbitTemplateTests)" +
            ".LOOKUP_KEY_COUNT.getAndIncrement() % 2 == 0 ? 'foo' : 'bar'");

AmqpTemplate

Spring AMQP提供了一个发送和接收消息的操作模板类AmqpTemplate。 AmqpTemplate它定义包含了发送和接收消息等的一些基本的操作功能。RabbitTemplate是AmqpTemplate的一个实现。

RabbitTemplate支持消息的确认与返回，为了返回消息，RabbitTemplate 需要设置mandatory 属性为true,并且CachingConnectionFactory 的publisherReturns属性也需要设置为true。返回的消息会根据它注册的RabbitTemplate.ReturnCallback setReturnCallback 回调发送到给客户端，

一个RabbitTemplate仅能支持一个ReturnCallback 。

为了确认Confirms消息, CachingConnectionFactory 的publisherConfirms 属性也需要设置为true，确认的消息会根据它注册的RabbitTemplate.ConfirmCallback setConfirmCallback回调发送到给客户端。一个RabbitTemplate也仅能支持一个ConfirmCallback.

Sending messages

利用AmqpTemplate模板来发送消息。

AmqpTemplate提供了以下的几个方法来发送消息

void send(Message message) throws AmqpException;

void send(String routingKey, Message message) throws AmqpException;

void send(String exchange, String routingKey, Message message) throws AmqpException;

发送时，需要指定Exchange和routingKey. 如:

amqpTemplate.send("marketData.topic", "quotes.nasdaq.FOO",new Message("12.34".getBytes(), someProperties));

从Spring-Rabbit版本1.3开始，它提供了MessageBuilder 和 MessagePropertiesBuilder两个类，这可以方便我们以fluent流式的方式创建Message及MessageProperties对象。

Message message = MessageBuilder.withBody("foo".getBytes())

.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)

.setMessageId("123")

.setHeader("bar", "baz")

.build();

Receiving messages

接收消息可以有两种方式。比较简单的一种方式是，直接去查询获取消息，即调用receive方法,如果该方法没有获得消息，receive方法不阻塞，直接返回null。另一种方式是注册一个Listener监听器,一旦有消息到时来，异步接收。

注意: 接收消息，都需要从queue中获得,接收消息时，RabbitTemplate需要指定queue，或者设置默认的queue。

API提供的直接获得消息方法,如下

Message receive() throws AmqpException; 

Message receive(String queueName) throws AmqpException;

AmqpTemplate也提供了方法来直接接收POJOs对象(代替Message对象)，同时也提供了各种的MessageConverter用来处理返回的Object对象。

Object receiveAndConvert() throws AmqpException;

Object receiveAndConvert(String queueName) throws AmqpException;

AmqpTemplate 从Spring-Rabbit 1.3版本开始也提供了 receiveAndReply 方法来异步接收、处理及回复消息。

AmqpTemplate 要注意receive和reply阶段，大数时情况下，你需要提供仅仅一个ReceiveAndReplyCallback 的实现，用它来处理接收到消息和回复(对象)消息的业务逻辑。同时也要注意，ReceiveAndReplyCallback 可能返回null,在这种情况下，receiveAndReply 就相当于receive 方法。

boolean receiveAndReply(ReceiveAndReplyCallback callback) throws AmqpException;

boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback) throws AmqpException;

boolean receiveAndReply(ReceiveAndReplyCallback callback, String replyExchange, String replyRoutingKey) throws AmqpException;

boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback,String replyExchange, String replyRoutingKey) throws AmqpException;

Asynchronous Consumer异步接受消息

为了异步接收消息，Spring AMQP也提供了多种不同的实现方式。如:通过实现MessageListener的方式，或者通过注解@RabbitListener的方式等，来实现异步接受消息及处理。

MessageListener 异步接收消息，MessageListener 定义比较简单。

public interface MessageListener {

void onMessage(Message message);

}

如果你的程序需要依赖Channel,你需要用 ChannelAwareMessageListener这个接口。

public interface ChannelAwareMessageListener {

void onMessage(Message message, Channel channel) throws Exception;

}

如果你想将你的程序及Message API严格分开的话，可以用MessageListenerAdapter这个适配器类

MessageListenerAdapter listener = new MessageListenerAdapter(somePojo);

listener.setDefaultListenerMethod("myMethod");

看一下Container 容器。一般地，容器是处于“主动”的责任，以使侦听器回调可以保持被动触发处理。容器是生命周期的组件，它提供了开始和停止容器的方法。在配置容器时，我们需要配置AMQP Queue和MessageListener的对应连接,需要配置ConnectionFactory 的一个引用以及Listener、以及能够从该Queue中接收消息的Queue的引用,这样的Container才能通知到对应的Listener。

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();

container.setConnectionFactory(rabbitConnectionFactory);

container.setQueueNames("some.queue");

container.setMessageListener(new MessageListenerAdapter(somePojo));

从RabbitMQ 的3.2版本开始， broker支持消费消息的priority 优先级。只需要在SimpleMessageListenerContainer 添加priority属性的设置。

<rabbit:listener-container connection-factory="rabbitConnectionFactory">

<rabbit:listener queues="some.queue" ref="somePojo" method="handle" priority="10" />

利用注解@RabbitListener实现异步接受消息

从spring-rabbit 1.4版本开始，可心利用注解的@RabbitListener来异步接收消息，它是更为简便的方式。

@Component

public class MyService {

@RabbitListener(queues = "myQueue")

public void processOrder(String data) { ... }

}

在容器中，需要配置@EnableRabbit，来支持@RabbitListener起作用。

@Configuration

@EnableRabbit

public class AppConfig { 

　　@Bean

　　public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {

　　　　SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

　　　　factory.setConnectionFactory(connectionFactory());

　　　　factory.setConcurrentConsumers(3);

　　　　factory.setMaxConcurrentConsumers(10);

　　　　return factory;

　　}

}

通过上面的配置，就配置好Listener和Container了，就可以异步接收消息了。

Message Converters

AmqpTemplate 定义提供了各种发送和接收委拖给MessageConverter转化对象消息的方法。MessageConverter 本身比较简单，它提供了消息对象的转化,可将object转化成Message 对象，或者将Message 对象转化成Object对象。它提供了默认的SimpleMessageConverter实现，以及第三方的MessageConverter，如Jackson2JsonMessageConverter，MarshallingMessageConverter等，来处理消息与对象之间的转换。 

SimpleMessageConverter

SimpleMessageConverter是Spring AMQP中MessageConverter的一个默认实现。

public interface MessageConverter {

　　Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException; 

　　Object fromMessage(Message message) throws MessageConversionException; 

}

为了自定义转化对象，你也可以第三方的MessageConverter,如使用  Jackson2JsonMessageConverter 或者MarshallingMessageConverter,其中Jackson2JsonMessageConverter提供了Json对象的转化，MarshallingMessageConverter对象则提供了对象的Marshaller和 Unmarshaller转化。 

ContentTypeDelegatingMessageConverter

从Spring-Rabbit 1.4.2开始,它提供了ContentTypeDelegatingMessageConverter,它能根据不同的MessageProperties属性(contentType)决定来委托给具体的哪一个MessageConverter。

代码示例

import org.junit.Test;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

public class CodeTest {
    @Test
    public void testSendAndReceive() {
　　　　//创建ConnectionFactory //注意: guest的用户只能够在localhost 127.0.0.1进行测试
        String hostname = "localhost";
        String username = "mytest";
        String password = "mytest";
        String virtualHost = "/";
        CachingConnectionFactory cf = new CachingConnectionFactory(hostname);
        cf.setUsername(username);
        cf.setPassword(password);
        cf.setVirtualHost(virtualHost);

        RabbitAdmin admin = new RabbitAdmin(cf);

　　　　//创建Exchange
        String exchangeName = "direct.test.exchange";
        DirectExchange exchange = new DirectExchange(exchangeName);
        admin.declareExchange(exchange);

　　　　//创建Queue
        String queueName = "direct.test.queue";
        Queue queue = new Queue(queueName, true, false, false, null);
        admin.declareQueue(queue);

　　　　//创建Binding
        String routingKey = "direct.test.queue";
        admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(routingKey));

　　　　//创建RabbitTemplate
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cf);
        rabbitTemplate.setExchange(exchangeName);
        rabbitTemplate.setQueue(queueName);

　　　　//创建Message
        String messageStr = "this is direct message";
        Message message = MessageBuilder.withBody(messageStr.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).build();
　　　　//根据routingKey发送消息
        System.out.println("message=" + message);
        rabbitTemplate.send(routingKey, message);

　　　　//接收消息
        Message resultMessage = rabbitTemplate.receive();
        System.out.println("resultMessage=" + resultMessage);
        if (resultMessage != null) {
            System.out.println("receive massage=" + new String(resultMessage.getBody()));
        }
    }
}

 异步接收消息实例

<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory" exchange="direct.test.exchange" queue="direct.test.queue"/>
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:direct-exchange name="direct.test.exchange" durable="true">
　　<rabbit:bindings>
　　　　<rabbit:binding queue="direct.test.queue" key="direct.test.queue">rabbit:binding>
　　<rabbit:bindings>
<rabbit:direct-exchange/>
<rabbit:queue name="direct.test.queue" durable="true" auto-delete="false" exclusive="false"/>
<rabbit:connection-factory id="connectionFactory" username="mytest" password="mytest" host="127.0.0.1" port="5672"/>

<bean id="myMessageListener" class="com.company.rabbitmqpro.service.MyMessageListener">bean>
<bean id="simpleMessageListenerContainerFactory" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
　　<property name="connectionFactory" ref="connectionFactory"/>
　　<property name="queueNames" >
　　　　<list>
　　　　　　<value>direct.test.queuevalue>
　　　　<list>
　　<property/>
　　<property name="messageListener" ref="myMessageListener">property>
　　<property name="concurrentConsumers" value="3"/>
　　<property name="maxConcurrentConsumers" value="10"/>
<bean/>

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
public class MyMessageListener implements MessageListener{
　　@Override
　　public void onMessage(Message message) {
　　　　System.out.println("received: " + message);
　　}
}

import javax.annotation.Resource;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "classpath*:applicationContext.xml" })
public class ListenerTest {
　　@Resource
　　private RabbitTemplate rabbitTemplate;
　　@Test
　　public void testSendAsynListener() {
　　　　String sendMsg = "this is direct message";
　　　　Message message = MessageBuilder.withBody(sendMsg.getBytes())
　　　　　　　　　　.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
　　　　　　　　　　.build();
　　　　String routingKey = "direct.test.queue";
　　　　rabbitTemplate.send(routingKey, message);
　　　　System.out.println("send ok");
　　}
}