spring的rabbitmq配置_xml增加spring.rabbitmq.connection-timeout 的值

作者：盐析白兔 | 2024-02-25 06:07:29

踩

xml增加spring.rabbitmq.connection-timeout 的值

来源：http://conkeyn.iteye.com/blog/2197486

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;  
     }  
   
 }  

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/139217?site