1、applicationContext-base.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 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>
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); } }
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); } } }
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
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())); } }
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; } }
相关推荐
RabbitMq与Spring整合实例,整个工程采用maven,具体过程看博文: http://blog.csdn.net/evankaka/article/details/50495437
spring amqp 配置实现rabbitmq 路由
Spring rabbitmq开发技术,rabbitmq主要配置由fectory- connect ,template声明,exchange交换器,queue队列,,,监听
spring的2个数据源的配置,并且每个数据源都配置了事物管理。还有rabbitMQ的发送端代码。本人项目亲自用到的,可以运行。
RabbitMQ 是一个开源的消息代理中间件,广泛用于构建分布式应用程序中的消息系统。在 Spring Boot 项目中,通过集成 RabbitMQ,可以实现异步消息传递、消息队列等功能,提高系统的可靠性和扩展性。
本篇文章主要介绍了Spring Boot 配置多个RabbitMQ,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
SSM集成rabbitmq的web工程,这个文件是初始的web项目,下载后可以用来练习。具体的配置,请参考本人的博客
这里两个项目,一个发送mq消息列队,另一个介绍mq消息,你需要配置安装rabbitmq
spring3 + rabbitmq 包含配置文件demo,用到了fastjson 。消息生产者使用了消息确认,消费者使用了事务 用到的包基本都是最近的新版本:spring-amqp-1.6.0.RELEASE.jar spring-rabbit-1.3.5.RELEASE.jar spring-retry...
mysql需开启binlog 查看是否开启binlog ...3.3 rabbitmq配置 在virtualHost:/ 下新增Exchanges: canal.exchange 新增队列:test.queue, 绑定canal.queue, RoutingKey:canal.routing.key canal下载及配置 ...
spring集成rabbitmq 通俗易懂的demo,保证可以使用,具体的配置,可以参考我的文章
spring-cloud之spring-cloud-stream,底层为rabbitmq实现,简单栗子,仅供参考。生产者、消费者配置及生产者、消费者代码实现。
rabbitMQ 简单的配置,连接配置&spring; template声明&消息对象json转换类。。。 都比较简但的XML文件配置
实践方志鹏博客搭建Springcloud+RabbitMQ+Config-client+config-server Eureka-server的微服务架构,通过/bus/refresh接口触发所有config-client自动从config-server重新读取配置文件。SpringCloud和SpringBoot版本...
RabbitMQ连接池+SpringBoot实现。通过连接池实现将高效的管理RabbitMQ的Connection,并与springboot进行整合,实现消息发送,获取队列列表等功能。基于此可以进行更多功能的扩充。
Springboot整合RabbitMQ最简单demo,适用于springcloud项目,作为消息总线适用,需要安装RabbitMQ,Mac linux可以使用命令行一键安装,在项目配置文件配置好端口即可(已默认配置),启动项目访问8080端口,参数见controller.
资源包含rabbitmq镜像集群的搭建文档和springboot连接rabbitmq集群的配置方式,供参考
1 SpringBoot整合RabbitMQ实战系列教程-整合配置篇-源码数据库
springboot集成RabbitMQ非常简单,如果只是简单的使用配置非常少,springboot提供了spring-boot-starter-amqp项目对消息各种支持。下面通过本文给大家介绍下spring boot整合RabbitMQ(Direct模式),需要的朋友可以...
spring boot和rabbitMQ最佳实践 1、JSON序列化与反序列化 要实现消息跨平台,需要配置消息JSON序列化(配置见代码实现),就可以实现不同语言之间互相发送/接收消息,还可以直接用RabbitMQ控制台发送消息。 2、...