spring-amqp 配置实现rabbitmq 路由

示例说明

将主题交换机应用到我们的日志系统中。在开始工作前，我们假设日志的路由键由两个单词组成，路由键看起来是这样的：service.severity代码跟上一篇教程差不多

假设有 order,user,email 服务，日志级别有error,info,warning。

生产者：服务和日志级别的排列组合作为routing key(如 order.email,order.info,order.warning,user.error,user.info .....),分别发送3条日志。

消费者：定义消费者，A、接收所有error 级别日志（ *.error）；B、接收 order 服务的日志（order.*），C、接收user服务和info级别的日志（user.* *.info）,D、接收所有的日志（#）

请参考 “rabbitmq 主题交换机java 实现”的示例说明（ http://blog.csdn.net/convict_eva/article/details/52315474）
项目依赖：

<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.6.1.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-amqp</artifactId>
<version>1.6.1.RELEASE</version>
</dependency>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.6.1</version>
</dependency>

目录结构



MsgHandler  消息处理方法实现

Recv 启动消息接收

Send 消息发送

applicationContext-consumer-sync.xml 消息接收者配置

applicationContext-producer-sync.xml 消息发送者配置

rabbitmq.properties rabbitmq 的属性配置

代码说明

配置文件

rabbitmq.host=192.168.174.128
rabbitmq.port=5672
rabbitmq.username=convict_eva
rabbitmq.password=convict_eva
rabbitmq.virtualHost=/convict_eva
#队列名
rabbitmq.queue.a=spring_queue_a
rabbitmq.queue.b=spring_queue_b
rabbitmq.queue.c=spring_queue_c
rabbitmq.queue.d=spring_queue_d
#队列对应的routing key
rabbitmq.queue.a.routing=*.error
rabbitmq.queue.b.routing=order.*
rabbitmq.queue.c.routing=user.*
rabbitmq.queue.c.routing_2=*.info
rabbitmq.queue.d.routing=#
#发送消息的路由
rabbitmq.msg.routing=user.error
#交换机名称
rabbitmq.exchangeName=topic_exchage

消息发送者

加载配置文件

<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE" />
<property name="ignoreResourceNotFound" value="true" />
<property name="locations">
<list>
<!-- rabbitmq配置 -->
<value>classpath*:/rabbitmq.properties</value>
</list>
</property>
</bean>

定义连接工厂

<bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
<property name="host" value="${rabbitmq.host}" />
<property name="port" value="${rabbitmq.port}" />
<property name="username" value="${rabbitmq.username}" />
<property name="password" value="${rabbitmq.password}" />
<property name="virtualHost" value="${rabbitmq.virtualHost}" />
<!-- 缓存模式   CONNECTION CHANNEL，默认的缓存模式是CHANNEL。
当缓存模式是 CONNECTION时, 队列的自动声明等等 (参考 the section called “Automatic Declaration of Exchanges, Queues and Bindings”) 将不再支持。
在框架(如． RabbitTemplate) 中使用的通道将会可靠地返回到缓存中.如果在框架外创建了通道 (如.直接访问connection(s)并调用 createChannel() ),
你必须可靠地返回它们(通过关闭),也许需要在 finally 块中以防止耗尽通道.
-->
<property name="cacheMode" value="CHANNEL"/>
<!-- 默认通道缓存25,多线程环境中，较小的缓存意味着通道的创建和关闭将以很高的速率运行.加大默认缓存大小可避免这种开销
如果达到了限制,调用线程将会阻塞，直到某个通道可用或者超时, 在后者的情况中，将抛出 AmqpTimeoutException异常.-->
<property name="channelCacheSize" value="10"/>
<!-- channelCheckoutTimeout属性. 当此属性的值大于０时， channelCacheSize会变成连接上创建通道数目的限制. -->
<property name="channelCheckoutTimeout" value="200"/>
<!-- connectionLimit 1.5.5 新增，会限制连接的总数目,当达到限制值时，将 channelCheckoutTimeLimit  来等待空闲连接.如果时间超时了，将抛出 AmqpTimeoutException -->
<property name="connectionLimit" value="50"/>
</bean>

定义队列

<bean id="springQueueA" class="org.springframework.amqp.core.Queue">
<constructor-arg  name="name" value="${rabbitmq.queue.a}"/>
<property name="shouldDeclare" value="false"/>
</bean>
<bean id="springQueueB" class="org.springframework.amqp.core.Queue">
<constructor-arg  name="name" value="${rabbitmq.queue.b}"/>
<property name="shouldDeclare" value="false"/>
</bean>
<bean id="springQueueC" class="org.springframework.amqp.core.Queue">
<constructor-arg  name="name" value="${rabbitmq.queue.c}"/>
<property name="shouldDeclare" value="false"/>
</bean>
<bean id="springQueueD" class="org.springframework.amqp.core.Queue">
<constructor-arg  name="name" value="${rabbitmq.queue.d}"/>
<property name="shouldDeclare" value="false"/>
</bean>

定义exchange

<bean id="exchange" class="org.springframework.amqp.core.TopicExchange">
<constructor-arg name="name" value="${rabbitmq.exchangeName}"/>
<constructor-arg name="durable" value="true"/>
<constructor-arg name="autoDelete" value="false"/>
</bean>

exchange 和队列的绑定

<!-- exchange 和 队列绑定  开始 -->
<bean id="arguments" class="java.util.HashMap"/>
<!-- 绑定 -->
<bean id="bindingA" class="org.springframework.amqp.core.Binding">
<!-- 消息发送到目标 -->
<constructor-arg name="destination" value="${rabbitmq.queue.a}"/>
<!-- 消息发送到目标的类型，这里使用队列，也可以是exchange -->
<constructor-arg name="destinationType" value="QUEUE"/>
<!-- 使用的队列名称 -->
<constructor-arg name="exchange" value="${rabbitmq.exchangeName}"/>
<!-- 绑定的routing key -->
<constructor-arg name="routingKey" value="${rabbitmq.queue.a.routing}"/>
<constructor-arg name="arguments" ref="arguments"/>
</bean>

rabbitmq 同步访问模板定义，用于发送和接收消息

<!-- 同步访问rabbitmq-->
<bean id="rabbitTemplate"  class="org.springframework.amqp.rabbit.core.RabbitTemplate">
<constructor-arg name="connectionFactory" ref="connectionFactory"/>
<property name="exchange" value="${rabbitmq.exchangeName}"/>
<!-- 也可以在发送的时候手动设置routing key -->
<property name="routingKey" value="${rabbitmq.msg.routing}"/>
</bean>

发送消息代码实现（Send）

public static void main(String[] args) throws Exception{
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext-producer-sync.xml");
RabbitTemplate template = ctx.getBean(RabbitTemplate.class);
//可以设置发送消息的routing key
//        template.setRoutingKey("user.error");
String msg;
for(int i=0;i<3;i++){
msg = "Hello World!"+i;
template.convertAndSend(msg);
System.out.println(msg);
}
ctx.destroy();
}

消费者

加载配置文件和定义连接工厂就不用说了，和生产者是一样的

<!-- 声明消息转换器为SimpleMessageConverter, -->
<bean id="messageConverter"
class="org.springframework.amqp.support.converter.SimpleMessageConverter">
</bean>

<!-- 用于接收消息的处理类 -->
<bean id="receiveHandler" class="com.convict.rabbitmq.spring.sync.MsgHandler"/>

<!-- 用于消息的监听的代理类MessageListenerAdapter,指定接收消息的类和方法及转换消息的bean -->
<bean id="receiveListenerAdapter"
class="org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="receiveHandler" />
<!-- MsgHandler 不实现 ChannelAwareMessageListener 接口，可以自定义消息处理方法  -->
<!--  <property name="defaultListenerMethod" value="handleMassage"/>-->
<!-- 也可以不设置，默认为SimpleMessageConverter  -->
<property name="messageConverter" ref="messageConverter"/>
</bean>

<!-- 用于消息的监听的容器类SimpleMessageListenerContainer,queueNames 可以配置多个队列名称用逗号隔开 -->
<bean id="listenerContainer"
class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
<!-- 监听多个队列，使用逗号分开 -->
<property name="queueNames" value="${rabbitmq.queue.a},${rabbitmq.queue.c}"></property>
<property name="connectionFactory" ref="connectionFactory"></property>
<property name="messageListener" ref="receiveListenerAdapter"></property>
</bean>

消息处理方法实现

@Override
public void onMessage(Message message, Channel channel) throws Exception {
System.out.println("consumer receive massage:" + new String(message.getBody())+"______routing key : "+message.getMessageProperties().getReceivedRoutingKey());
//channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}
/**
* MessageListenerAdapter：
*
* 如果返回的不是null，就把返回的值当着一个消息发送到接收到消息的exchange，并使用消息的replyTo属性作为routing key。
* 原文如下：
* If a target listener method returns a non-null object (typically of a message content type such as
* <code>String</code> or byte array), it will get wrapped in a Rabbit <code>Message</code> and sent to the exchange of
* the incoming message with the routingKey that comes from the Rabbit ReplyTo property or via
* {setResponseRoutingKey(String) specified routingKey}).
*
*/
public void handleMassage(String msg){
System.out.println("consumer receive massage:" + msg);
}

完整代码：http://download.csdn.net/detail/convict_eva/9620260

注：在路由绑定队列时，不能同时绑定多个routing key