spring的rabbitmq配置

1、applicationContext-base.xml

Xml代码


<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

xmlns:aop="http://www.springframework.org/schema/aop" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:p="http://www.springframework.org/schema/p" xmlns:mvc="http://www.springframework.org/schema/mvc"

xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"

xmlns:websocket="http://www.springframework.org/schema/websocket"

xmlns:jpa="http://www.springframework.org/schema/data/jpa" xmlns:cache="http://www.springframework.org/schema/cache"

xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:rabbit="http://www.springframework.org/schema/rabbit"

xmlns:task="http://www.springframework.org/schema/task"

xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd http://www.springframework.org/schema/cache http://www.springframework.org/schema/cache/spring-cache-4.1.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd http://www.springframework.org/schema/websocket http://www.springframework.org/schema/websocket/spring-websocket-4.1.xsd http://www.springframework.org/schema/data/jpa http://www.springframework.org/schema/data/jpa/spring-jpa-1.3.xsd http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.1.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.1.xsd">
<!-- 自动扫描包，可以写多个 -->

<context:component-scan base-package="com.test.**">

<context:exclude-filter type="annotation"

expression="org.springframework.stereotype.Controller" />

</context:component-scan>

<!-- 开启注解事务只对当前配置文件有效 -->

<tx:annotation-driven transaction-manager="transactionManager"

proxy-target-class="true" />

<jpa:repositories base-package="com.test.

repository-impl-postfix="Impl" entity-manager-factory-ref="entityManagerFactory"

transaction-manager-ref="transactionManager" />

<bean id="entityManagerFactory"

class="org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean">

<property name="dataSource" ref="dataSource" />

<property name="packagesToScan" value="com.test. />

<property name="persistenceProvider">

<bean class="org.hibernate.ejb.HibernatePersistence" />

</property>

<property name="jpaVendorAdapter">

<bean class="org.springframework.orm.jpa.vendor.HibernateJpaVendorAdapter">

<property name="generateDdl" value="true" />

<property name="databasePlatform" value="${hibernate.dialect}" />

<property name="showSql" value="${hibernate.show_sql}" />

</bean>

</property>

<property name="jpaDialect">

<bean class="org.springframework.orm.jpa.vendor.HibernateJpaDialect" />

</property>

<property name="jpaPropertyMap">

<map>

<entry key="hibernate.query.substitutions" value="true 1, false 0" />

<entry key="hibernate.default_batch_fetch_size" value="16" />

<entry key="hibernate.max_fetch_depth" value="2" />

<entry key="hibernate.generate_statistics" value="true" />

<entry key="hibernate.bytecode.use_reflection_optimizer"

value="true" />

<entry key="hibernate.cache.use_second_level_cache" value="${hibernate.cache.use_second_level_cache}" />

<entry key="hibernate.cache.use_query_cache" value="${hibernate.cache.use_query_cache}" />

<entry key="hibernate.hbm2ddl.auto" value="${hibernate.hbm2ddl.auto}" />

</map>

</property>

</bean>

<!--事务管理器配置 -->

<bean id="transactionManager" class="org.springframework.orm.jpa.JpaTransactionManager">

<property name="entityManagerFactory" ref="entityManagerFactory" />

</bean>

<!-- 数据源 -->

<bean name="dataSource"

class="org.springframework.jdbc.datasource.DriverManagerDataSource">

<property name="driverClassName" value="${hibernate.connection.driver_class}" />

<property name="url" value="${hibernate.connection.url}" />

<property name="username" value="${hibernate.connection.username}" />

<property name="password" value="${hibernate.connection.password}" />

</bean>

<bean id="objectMapper" class="com.test.core.utils.JsonObjectMapper" />

<!-- 初始化数据库记录 -->

<jdbc:initialize-database data-source="dataSource"

ignore-failures="ALL">

<jdbc:script location="classpath:*.sql" encoding="UTF-8" />

</jdbc:initialize-database>

<!-- 异步的线程池，线程池的最在数不能设定太小，不然<rabbit:listener/>/@RabbitListener太多的话，会出现发无法正常消费问题 -->

<task:executor id="taskExecutor" pool-size="4-256" queue-capacity="128" />

<!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 -->

<rabbit:annotation-driven />

<bean id="rabbitListenerContainerFactory" class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">

<property name="connectionFactory" ref="rabbitConnFactory" />

<property name="transactionManager" ref="transactionManager" />

<property name="concurrentConsumers" value="1" />

<property name="maxConcurrentConsumers" value="10" />

<property name="messageConverter" ref="jsonMessageConverter" />

<property name="taskExecutor" ref="taskExecutor" />

<property name="channelTransacted" value="true" />

<property name="adviceChain">

<array>

<ref bean="retryInterceptor" />

</array>

</property>

</bean>

<!-- rabbit:admin用于管理（创建和删除） exchanges, queues and bindings等 -->

<bean id="rabbitConnectionFactory" class="com.rabbitmq.client.ConnectionFactory">

<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.vhost}" />

<property name="connectionTimeout" value="${rabbitmq.connection.timeout}" />

</bean>

<bean id="rabbitConnFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">

<constructor-arg ref="rabbitConnectionFactory" />

<property name="channelCacheSize" value="25" />

<property name="executor" ref="taskExecutor" />

</bean>

<rabbit:admin connection-factory="rabbitConnFactory" id="rabbitAdmin" />

<!-- 180秒 -->

<rabbit:template id="amqpTemplate" reply-timeout="1000" connection-factory="rabbitConnFactory" message-converter="jsonMessageConverter" />

<!-- 定义接收异常消息的exchange和queue -->

<util:map id="dlxNaming" key-type="java.lang.String" value-type="java.lang.String">

<entry key="zkcloud.subsystem.dlx.queue" value="#{'$dlx_queue_'+(T(com.zkteco.timecube.zkcloud.core.utils.PropertiesUtil).getValue('zkcloud.subsystem.code'))}" />

<entry key="zkcloud.subsystem.dlx.exchange" value="#{'$dlx_ex_'+(T(com.zkteco.timecube.zkcloud.core.utils.PropertiesUtil).getValue('zkcloud.subsystem.code'))}" />

</util:map>

<rabbit:queue id="zkcloud.subsystem.dlx.queue" name="#{dlxNaming['zkcloud.subsystem.dlx.queue']}">

<rabbit:queue-arguments>

<entry key="x-message-ttl">

<value type="java.lang.Long">86400000</value>

</entry>

<entry key="x-max-length">

<value type="java.lang.Long">100</value>

</entry>

</rabbit:queue-arguments>

</rabbit:queue>

<rabbit:fanout-exchange id="zkcloud.subsystem.dlx.exchange" name="#{dlxNaming['zkcloud.subsystem.dlx.exchange']}">

<rabbit:bindings>

<rabbit:binding queue="zkcloud.subsystem.dlx.queue" />

</rabbit:bindings>

</rabbit:fanout-exchange>

<bean id="retryInterceptor" class="org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean">

<property name="messageRecoverer" ref="messageRecoverer" />

<property name="retryOperations" ref="retryTemplate" />

</bean>

<!-- <bean id="messageRecoverer" class="org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer" /> -->

<!-- 拒绝请求消息，并回复该请求者的请求被服务端拒绝-->

<bean id="messageRecoverer" class="com.test.retry.RejectAndRplyToRequeueRecoverer">

<property name="replyToTemplate" ref="amqpTemplate"/>

</bean>

<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">

<property name="backOffPolicy">

<bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy">

<property name="initialInterval" value="1000" />

<property name="maxInterval" value="10000" />

</bean>

</property>

<property name="retryPolicy">

<bean class="org.springframework.retry.policy.SimpleRetryPolicy">

<property name="maxAttempts" value="1" />

</bean>

</property>

</bean>

<bean id="jsonMessageConverter"

class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter"></bean>

<!-- quartz配置 -->

<bean class="com.zkteco.timecube.quartz.QuartJobSchedulingListener" />

<bean class="org.springframework.scheduling.quartz.SchedulerFactoryBean">

<property name="jobFactory">

<bean class="com.zkteco.timecube.quartz.SpringQuartzJobFactory"></bean>

</property>

<property name="dataSource" ref="dataSource" />

<!-- 要记得要指定配置文件的位置 -->

<property name="configLocation" value="classpath:config/quartz.properties" />

</bean>

<!-- quartz配置 -->

<beans profile="develop">

<bean id="propertyConfigurer" class="com.test.core.utils.PropertiesUtil"

lazy-init="false">

<property name="locations">

<list>

<value>classpath*:config/*.properties</value>

</list>

</property>

<property name="fileEncoding" value="utf-8" />

</bean>

<!-- 连接rabbitmq -->

<rabbit:connection-factory id="rabbitConnFactory"

host="localhost" username="guest" password="guest" port="5672"

virtual-host="/" connection-timeout="30000" executor="taskExecutor" />

</beans>

<beans profile="test">

<bean id="propertyConfigurer" class="com.test.core.utils.PropertiesUtil"

lazy-init="false">

<property name="locations">

<list>

<value>classpath*:config/*.properties</value>

<value>classpath*:config/test/*.properties</value>

</list>

</property>

<property name="fileEncoding" value="utf-8" />

</bean>

<!-- 连接rabbitmq -->

<rabbit:connection-factory id="rabbitConnFactory"

host="192.168.0.179" username="guest" password="timeucbe" port="5672"

virtual-host="/" connection-timeout="30000" executor="taskExecutor" />

</beans>

<beans profile="production">

<bean id="propertyConfigurer" class="com.test.core.utils.PropertiesUtil"

lazy-init="false">

<property name="locations">

<list>

<value>classpath*:config/*.properties</value>

<value>classpath*:config/production/*.properties</value>

</list>

</property>

<property name="fileEncoding" value="utf-8" />

<!-- 连接rabbitmq -->

<rabbit:connection-factory id="rabbitConnFactory"

host="114.215.82.3" username="guest" password="timecube" port="5672"

virtual-host="/" connection-timeout="30000" executor="taskExecutor" />

</bean>

</beans>

</beans>

2、Exchanges、routing keys、binding keys的配置

Xml代码


<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

xmlns:aop="http://www.springframework.org/schema/aop" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:p="http://www.springframework.org/schema/p" xmlns:mvc="http://www.springframework.org/schema/mvc"

xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"

xmlns:websocket="http://www.springframework.org/schema/websocket"

xmlns:jpa="http://www.springframework.org/schema/data/jpa" xmlns:cache="http://www.springframework.org/schema/cache"

xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:rabbit="http://www.springframework.org/schema/rabbit"

xmlns:task="http://www.springframework.org/schema/task"

xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd http://www.springframework.org/schema/cache http://www.springframework.org/schema/cache/spring-cache-4.1.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd http://www.springframework.org/schema/websocket http://www.springframework.org/schema/websocket/spring-websocket-4.1.xsd http://www.springframework.org/schema/data/jpa http://www.springframework.org/schema/data/jpa/spring-jpa-1.3.xsd http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.1.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.1.xsd">
<rabbit:queue id="queue_one" durable="true" auto-delete="false"

name="queue_one">

<!-- <rabbit:queue-arguments>

<entry key="x-message-ttl">

<value type="java.lang.Long">100</value>

</entry>

<entry key="x-ha-policy" value="all" />

</rabbit:queue-arguments> -->

</rabbit:queue>

<rabbit:direct-exchange name="my-mq-exchange"

durable="true" auto-delete="false" id="my-mq-exchange">

<rabbit:bindings>

<rabbit:binding queue="queue_one" key="queue_one_key" />

</rabbit:bindings>

</rabbit:direct-exchange>

<rabbit:queue id="queue_two" durable="true" auto-delete="false"

exclusive="false" name="queue_two" />

<rabbit:direct-exchange name="my-mq-exchange1"

durable="true" auto-delete="false" id="my-mq-exchange1">

<rabbit:bindings>

<rabbit:binding queue="queue_two" key="queue_two_key" />

</rabbit:bindings>

</rabbit:direct-exchange>

</beans>

Java代码


import javax.annotation.Resource;

import org.springframework.amqp.core.AmqpTemplate;

import org.springframework.stereotype.Service;

import org.springframework.transaction.annotation.Transactional;

/**

* 查

* @version 0.0.0.1

* @since 2015年3月30日 下午3:22:49

*/

@Service("producerMq")

@Transactional

public class ProducerMq

{

@Resource

private AmqpTemplate amqpTemplate;

//同步示例

public void sendDataToCrQueue(Object obj)

{

amqpTemplate.convertAndSend("my-mq-exchange", "queue_one_key", obj);

}

}

Java代码


import javax.annotation.Resource;

import org.springframework.stereotype.Controller;

import org.springframework.web.bind.annotation.RequestMapping;

/**

* 大

* @version 0.0.0.1

* @since 2015年3月30日 下午3:23:12

*/

@Controller

public class MessageController

{

@Resource

private ProducerMq producer;

@RequestMapping("/producer")

public void producer() throws Exception

{

for (int i = 0; i < 100; i++)

{

producer.sendDataToCrQueue("data" + i);

}

}

}

Java代码


import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.messaging.handler.annotation.SendTo;

import org.springframework.stereotype.Component;

/**

* 队列监听器

*

* @author <a href="mailto:zhongqing.lin@zkteco.com">zhongqing.lin</a>

* @version 0.0.0.1

* @since 2015年3月30日 下午7:02:59

*/

@Component

public class QueueOneLitener

{

@RabbitListener(queues = "queue_one", exclusive = false,containerFactory="rabbitListenerContainerFactory",admin="rabbitAdmin")

//参数中使用@Header获取mesage

@SendTo("my-mq-exchange1/queue_two_key")

public org.springframework.messaging.Message<String> data1(Message message)

{

System.out.println("headers:" + message.getMessageProperties().toString());

String data = new String(message.getBody());

System.out.println("queue_one data:" + data);

return org.springframework.messaging.support.MessageBuilder.withPayload(data).build();

}

}

注意：

@SendTo的value填入的值应该是“exchange/routingKey”格式。

foo/bar - the replyTo exchange and routingKey.
foo/ - the replyTo exchange and default (empty) routingKey.
bar or /bar - the replyTo routingKey and default (empty) exchange.
/ or empty - the replyTo default exchange and default routingKey.

参考地址：http://docs.spring.io/spring-amqp/reference/htmlsingle/#async-annotation-driven-reply

Java代码


import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

/**

* 队列监听器

*

* @author <a href="mailto:zhongqing.lin@zkteco.com">zhongqing.lin</a>

* @version 0.0.0.1

* @since 2015年3月30日 下午7:02:59

*/

@Component

public class QueueTwoLitener

{

@RabbitListener(queues = "queue_two", exclusive = false)

//参数中使用@Header获取mesage

public void onMessage(Message message)

{

System.out.println("queue_two data:" + new String(message.getBody()));

}

}

Java代码


package com.test.rabbit.retry;

import java.util.Map;

import org.apache.commons.lang3.StringUtils;

import org.springframework.amqp.core.Address;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.core.MessageProperties;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;

import com.test.utils.MessageUtil;

import com.test.utils.PropKeys;

/**

* 拒绝消息，并回复

*

* @version 0.0.0.1

* @since 2015年4月21日 下午5:05:35

*/

public class ZkRejectAndRplyToRequeueRecoverer extends RejectAndDontRequeueRecoverer

{

/** 用于发送拒绝消息状态给请求者 */

RabbitTemplate replyToTemplate;

@Override

public void recover(Message message, Throwable cause)

{

MessageProperties mp = message.getMessageProperties();

if (mp != null && StringUtils.isNotBlank(mp.getReplyTo()) && replyToTemplate != null)

{

Map<String, Object> headers = mp.getHeaders();

System.err.println(headers.toString());

Object vLang = headers.get(PropKeys.LANG);

String lang = "en";

if (vLang != null)

{

lang = (String) vLang;

}

com.test.utils.Message rejectRespMsg = new com.test.utils.Message(false);

rejectRespMsg.setPayload(null);

MessageUtil.changeResult(rejectRespMsg, "test.rabbit.replyto.interceptor.illegal.request", lang);

Address address = new Address(mp.getReplyTo());

replyToTemplate.convertAndSend(address.getExchangeName(), address.getRoutingKey(), rejectRespMsg);

}

super.recover(message, cause);

}

public void setReplyToTemplate(RabbitTemplate replyToTemplate)

{

this.replyToTemplate = replyToTemplate;

}

}