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; } }
相关推荐
接下来是`pom.xml`,这是一个Maven项目配置文件,包含了Spring Boot和Spring AMQP(包含RabbitMQ支持)的依赖。例如: ```xml <groupId>org.springframework.boot <artifactId>spring-boot-starter-amqp ...
2. **创建配置**:在Spring的配置文件(如`applicationContext.xml`)中,定义RabbitMQ的相关配置,包括连接工厂、交换机、队列和绑定关系。例如: ```xml <bean id="rabbitConnectionFactory" class="org.spring...
Spring RabbitMQ AMQP 是一个基于Java的开源框架,它整合了RabbitMQ消息中间件,实现了高级消息队列协议(AMQP)。这个框架是Spring生态的一部分,为Spring Boot应用程序提供了强大的消息处理能力。AMQP是一种标准的...
首先,我们要理解Spring对RabbitMQ的支持主要体现在Spring AMQP项目中,它为RabbitMQ提供了一套高级抽象层,使得开发者能够更加便捷地使用RabbitMQ。在整合Spring和RabbitMQ时,我们需要引入相应的依赖。在Maven工程...
在本文中,我们将深入探讨如何使用Spring AMQP配置来实现RabbitMQ的路由。Spring AMQP是Spring框架的一个模块,它提供了与RabbitMQ消息中间件集成的能力,使得在Java应用中处理AMQP(Advanced Message Queuing ...
1. **spring-rabbit-1.5.1.RELEASE.jar**:这是Spring对RabbitMQ的官方支持模块,它提供了与RabbitMQ集成的API和配置。这个库包含了Spring AMQP项目的核心功能,如连接工厂、模板类、监听容器等,使得开发者能够方便...
本项目是一个基于SSM的Web工程,已经集成了RabbitMQ的基础配置,适合初学者用来实践和学习如何在SSM项目中应用RabbitMQ。 **1. Spring与RabbitMQ的整合** Spring框架提供了一个方便的API来与RabbitMQ进行集成。...
接下来,我们将详细讨论如何在SpringCloud项目中配置和使用RabbitMQ。 首先,你需要在本地或者服务器上安装RabbitMQ。可以从官方网站下载对应操作系统的安装包,按照指南完成安装。安装完成后,确保RabbitMQ服务...
本示例聚焦于Spring的事务管理和RabbitMQ的使用,这都是分布式系统中不可或缺的组件。 首先,让我们深入了解Spring的事务管理。在Java环境中,事务管理是确保数据一致性的重要手段。Spring提供了一种声明式事务管理...
Spring Boot 配置多个 RabbitMQ Spring Boot 是一个流行的 Java 框架,用于快速构建生产级别的应用程序。RabbitMQ 是一个消息队列中间件,用于实现异步消息处理和队列管理。配置多个 RabbitMQ 实例是为了提高系统的...
以上是RabbitMQ配置的基本流程和关键知识点,实际操作中可能还需要根据具体需求进行更复杂的配置和优化。记住,理解RabbitMQ的核心概念——如交换器(exchanges)、队列(queues)和绑定(bindings)——对于有效...
1. **ApplicationContext-rabbitmq.xml**: 这个文件通常包含 Spring 的配置,用于设置 RabbitMQ 相关的 Bean。例如,配置 RabbitTemplate、ConnectionFactory 和 AmqpAdmin 等。 2. **ConnectionFactory**: 这是...
接着,配置RabbitMQ的相关参数。在Spring Boot项目中,可以在application.yml或application.properties文件中设置: ```yaml spring: rabbitmq: host: localhost port: 5672 username: guest password: guest ...
Spring Cloud是微服务架构中的一个组件集合,提供了服务发现、配置管理、断路器等核心功能,而RabbitMQ则是一个流行的开源消息队列系统,用于实现应用间的异步通信和解耦。 描述虽然为空,但我们可以从标签“源码”...
Spring 框架与 RabbitMQ 的集成是现代 Java 应用程序中常见的一种消息传递解决方案。RabbitMQ 是一个开源的消息代理和队列服务器,它使用 AMQP(Advanced Message Queuing Protocol)协议来处理应用程序之间的消息...
RabbitMQ作为一款流行的开源消息中间件,广泛应用于Spring Boot项目中。本教程将详细介绍如何在Spring Boot应用中结合RabbitMQ实现消息确认、消息持久化、备用交换机以及死信交换机等功能。 首先,让我们理解这些...
2. 配置RabbitMQ:在Spring的配置文件中,定义连接工厂、信道配置以及RabbitMQ服务器的相关属性。 3. 创建消息模板:使用`RabbitTemplate`作为发送和接收消息的工具类,配置在Spring容器中。 4. 定义消息监听器:...
在Spring中使用RabbitMQ时,我们首先需要配置RabbitMQ连接工厂(ConnectionFactory),然后创建一个RabbitTemplate实例,这是Spring AMQP提供的核心工具类,用于发送和接收消息。通过RabbitTemplate,我们可以使用...
RabbitMQ 是一个开源的消息代理中间件,广泛用于构建分布式应用程序中的消息系统。在 Spring Boot 项目中,通过集成 RabbitMQ,可以实现异步消息传递、消息队列等功能,提高系统的可靠性和扩展性。