使用springboot整合rabbitmq
逻辑梳理：
1.创建连接：ConnectionFactory (host、port、username、password)
2.创建队列并绑定 （queue、exchange、routingkey）
3.创建RabbitAdmin类用来管理
4. 生产者：设置消息发送类 RabbitTemplate 重发类 RetryTemplate
消费者：设置消息接收监听类，方法
第一步：创建连接
String host = StringUtils.hasText(env().getProperty(ENV_CONNECTION_HOST)) ?
env().getProperty(ENV_CONNECTION_HOST) : DEFAULT_CONNECTION_HOST;
int port = NumberUtils.toInt(env().getProperty(ENV_CONNECTION_PORT)) >= 1024 &&
NumberUtils.toInt(env().getProperty(ENV_CONNECTION_PORT)) < 65535 ?
NumberUtils.toInt(env().getProperty(ENV_CONNECTION_PORT)) : DEFAULT_CONNECTION_PORT;
String username = StringUtils.hasText(env().getProperty(ENV_CONNECTION_NAME)) ?
env().getProperty(ENV_CONNECTION_NAME) : DEFAULT_CONNECTION_NAME;
String password = StringUtils.hasText(env().getProperty(ENV_CONNECTION_PWD)) ?
env().getProperty(ENV_CONNECTION_PWD) : DEFAULT_CONNECTION_PWD;
int cachesize = NumberUtils.toInt(env().getProperty(ENV_CONNECTION_CACHESIZE)) > 5 &&
NumberUtils.toInt(env().getProperty(ENV_CONNECTION_CACHESIZE)) < 25 ?
NumberUtils.toInt(env().getProperty(ENV_CONNECTION_CACHESIZE)) : DEFAULT_CONNECTION_CACHESIZE;
CachingConnectionFactory factory = new CachingConnectionFactory(host,port);
factory.setUsername(username);
factory.setPassword(password);
//打印连接消息
logger.info("******rabbit connection*******");
logger.info("[主机地址：]"+host);
logger.info("[端口号：]"+port);
logger.info("[登录用户：]"+username);
logger.info("[系统默认连接数：]"+cachesize);
return factory;
}
第二步：创建队列并绑定
//创建队列，并将队列与交换机和路由键绑定
@Bean
public Queue queue(){
return new Queue(StringUtils.hasText(env().getProperty(ENV_QUEUE_NAME)) ?
env().getProperty(ENV_QUEUE_NAME) : DEFAULT_QUEUE_NAME);
}
//创建交换机,根据创建的类来指定交换机类型:TopicExchange、D
@Bean
public TopicExchange topicExchange(){
return new TopicExchange(StringUtils.hasText(env().getProperty(ENV_EXCHANGE_NAME)) ?
env().getProperty(ENV_EXCHANGE_NAME) : DEFAULT_EXCHANGE_NAME);
}
//将创建好的队列和交换机通过指定routingkey绑定
@Bean
public Binding queueBinding(){
return BindingBuilder.bind(queue()).
to(topicExchange()).
with(StringUtils.hasText(env().getProperty(ENV_ROUTINGKEY))?
env().getProperty(ENV_ROUTINGKEY) : DEFAULT_ROUTINGKEY);
}
第三步：创建RabbitAdmin类用来管理
//管理类
@Bean
public AmqpAdmin rabbitAdmin() {
return new RabbitAdmin(connectionFactory());
}
第四步：1.设置生产者消息发送类和断连重连类
//设置断连重发类
@Bean
public RetryTemplate retryTemplate(){
long interval = NumberUtils.toLong(env().getProperty(ENV_BACKOFF4ASYN_INTERVAL)) >=100?
NumberUtils.toLong(env().getProperty(ENV_BACKOFF4ASYN_INTERVAL)) :
DEFAULT_BACKOFF_INTERVAL;
double multiplier = NumberUtils.toDouble(env().getProperty(ENV_BACKOFF4ASYN_MULTIPLIER),0) >0?
NumberUtils.toDouble(env().getProperty(ENV_BACKOFF4ASYN_MULTIPLIER)) :
DEFAULT_BACKOFF_MULTIPLIER;
long max_interval = NumberUtils.toLong(env().getProperty(ENV_BACKOFF4ASYN_MAXINTERVAL),0) >1000?
NumberUtils.toLong(env().getProperty(ENV_BACKOFF4ASYN_MAXINTERVAL)) :
DEFAULT_BACKOFF_MAXINTERVAL;
ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
//initial-interval(>100)
backOffPolicy.setInitialInterval(interval);
backOffPolicy.setMultiplier(multiplier);
//max-interval(>1000)
backOffPolicy.setMaxInterval(max_interval);
RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setBackOffPolicy(backOffPolicy);
logger.info("*******rabbit retry-template******");
logger.info("[interval]{}",interval);
logger.info("[multiplier]{}",multiplier);
logger.info("[max_interval]{}",max_interval);
return retryTemplate;
}
//将生产者发送类和断连重发类绑定在一起的方法
private void setRetryTemplate(RabbitTemplate _template){
if(StringUtils.hasText(env().getProperty(ENV_BACKOFF4ASYN_OPEN)) &&
0 == env().getProperty(ENV_BACKOFF4ASYN_OPEN).compareToIgnoreCase("true")){
_template.setRetryTemplate(retryTemplate());
logger.info("[retry]{}",true);
}else{
logger.info("[retry]{}",false);
}
}
//生产者操作类
@Bean
public RabbitTemplate rabbitTemplate(){
RabbitTemplate template = new RabbitTemplate(connectionFactory());
//向操作类中设置已经和队列绑定好了的交换机
template.setExchange(topicExchange().getName());
//设置发出去的信息转化的方式，特别注意，发送者的消息处理方式必须和接收方一致
template.setMessageConverter(new FastJsonMessageConvert());
//向template中设置如果断连的重发类
setRetryTemplate(template);
logger.info("*******rabbit template*****");
logger.info("[exchange]:",topicExchange().getName());
logger.info("[message.convert]",FastJsonMessageConvert.class.getSimpleName());
return template;
}
2.设置消费者接收消息处理类
//设置消费者[SimpleMessageListenerContainer<--MessageListenerAdapter<--FastJsonMessageConvert]
@Bean
public SimpleMessageListenerContainer messageListenerContainer(){
//设置消费者连接信息
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
container.setQueueNames(queue().getName());
//设置监听模板
container.setMessageListener(messageListenerAdapter());
//设置应答方式
container.setAcknowledgeMode(AcknowledgeMode.NONE);
return container;
}
@Bean
public MessageListenerAdapter messageListenerAdapter() {
DataPackageHandler handler = new DataPackageHandler();
//设置消费者处理信息的方式和转化信息的方式
MessageListenerAdapter adapter = new MessageListenerAdapter(handler,new FastJsonMessageConvert());
//指定是消息处理类中的哪一个方法
adapter.setDefaultListenerMethod("messageHandler");
return adapter;
}
以下是完整代码：MyRabbitConfiguration类
package com.sunland.demo.configuration;
import java.util.Properties;
import org.apache.commons.lang.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.util.StringUtils;
import com.sunland.common.cfg.AtomRoot;
import com.sunland.common.cfg.XmlConfiguration;
/**
* 三个工具类：
* BindingBuilder:用来通过绑定指定的队列和交换机
* StringUtils:用来操作字符串
* NumberUtils:用来操作int
* @author qiu
*
*/
@Configuration
@EnableScheduling
public class MyRabbitConfiguration {
private static final Logger logger = LoggerFactory.getLogger(MyRabbitConfiguration.class.getName());
private Properties prop = new Properties();
private String xmlConfigure = "conf/myRabbit.xml";
private Properties env(){
if(prop.size() <= 0){
AtomRoot root = XmlConfiguration.getConfiguration(xmlConfigure);
if(root != null){
prop.putAll(XmlConfiguration.getAllProperties(root));
}
}
return prop;
}
//[rabbit设置]
//[connection设置]
private static String ENV_CONNECTION_HOST = "mq.host";
private static String ENV_CONNECTION_PORT = "mq.port";
private static String ENV_CONNECTION_NAME = "mq.name";
private static String ENV_CONNECTION_PWD = "mq.password";
private static String ENV_CONNECTION_CACHESIZE = "mq.cachesize";
//[默认连接设置]
private static String DEFAULT_CONNECTION_HOST = "localhost";
private static Integer DEFAULT_CONNECTION_PORT = 5672;
private static String DEFAULT_CONNECTION_NAME = "qiuhangxiang";
private static String DEFAULT_CONNECTION_PWD = "qiu1994825";
private static Integer DEFAULT_CONNECTION_CACHESIZE = 5;
//
private Integer MAX_CACHESIZE = 25;
//[生产者发送消息交换机、队列和路由键设置]
private static String ENV_EXCHANGE_NAME = "mq.exchange";
private static String ENV_QUEUE_NAME = "mq.queue";
private static String ENV_ROUTINGKEY = "mq.routingkey";
//[生产者发送消息默认交换机、队列和路由键设置]
private static String DEFAULT_EXCHANGE_NAME = "i.am.default.exchange";
private static String DEFAULT_QUEUE_NAME = "i.am.default.queue";
private static String DEFAULT_ROUTINGKEY = "i.am.default.routingkey";
//[消费者默认交换机和队列]
private static String DEFAULT_CUSTOMER_QUEUE_NAME = "i.am.default.customer.queue";
private static String DEFAULT_CUSTOMER_ROUTINGKEY = "i.am.default.customer.routingkey";
//异步报文发送重试模板
public static String ENV_BACKOFF4ASYN_OPEN = "mq.retry";
public static String ENV_BACKOFF4ASYN_INTERVAL = "mq.retry.interval";
public static String ENV_BACKOFF4ASYN_MULTIPLIER = "mq.retry.multiplier";
public static String ENV_BACKOFF4ASYN_MAXINTERVAL = "mq.retry.maxinterval";
public static long DEFAULT_BACKOFF_INTERVAL = 500;
public static double DEFAULT_BACKOFF_MULTIPLIER = 10.0;
public static long DEFAULT_BACKOFF_MAXINTERVAL = 10000;
@Bean
public ConnectionFactory connectionFactory(){
String host = StringUtils.hasText(env().getProperty(ENV_CONNECTION_HOST)) ?
env().getProperty(ENV_CONNECTION_HOST) : DEFAULT_CONNECTION_HOST;
int port = NumberUtils.toInt(env().getProperty(ENV_CONNECTION_PORT)) >= 1024 &&
NumberUtils.toInt(env().getProperty(ENV_CONNECTION_PORT)) < 65535 ?
NumberUtils.toInt(env().getProperty(ENV_CONNECTION_PORT)) : DEFAULT_CONNECTION_PORT;
String username = StringUtils.hasText(env().getProperty(ENV_CONNECTION_NAME)) ?
env().getProperty(ENV_CONNECTION_NAME) : DEFAULT_CONNECTION_NAME;
String password = StringUtils.hasText(env().getProperty(ENV_CONNECTION_PWD)) ?
env().getProperty(ENV_CONNECTION_PWD) : DEFAULT_CONNECTION_PWD;
int cachesize = NumberUtils.toInt(env().getProperty(ENV_CONNECTION_CACHESIZE)) > 5 &&
NumberUtils.toInt(env().getProperty(ENV_CONNECTION_CACHESIZE)) < 25 ?
NumberUtils.toInt(env().getProperty(ENV_CONNECTION_CACHESIZE)) : DEFAULT_CONNECTION_CACHESIZE;
CachingConnectionFactory factory = new CachingConnectionFactory(host,port);
factory.setUsername(username);
factory.setPassword(password);
//打印连接消息
logger.info("******rabbit connection*******");
logger.info("[主机地址：]"+host);
logger.info("[端口号：]"+port);
logger.info("[登录用户：]"+username);
logger.info("[系统默认连接数：]"+cachesize);
return factory;
}
//创建队列，并将队列与交换机和路由键绑定
@Bean
public Queue queue(){
return new Queue(StringUtils.hasText(env().getProperty(ENV_QUEUE_NAME)) ?
env().getProperty(ENV_QUEUE_NAME) : DEFAULT_QUEUE_NAME);
}
//创建交换机,根据创建的类来指定交换机类型:TopicExchange、D
@Bean
public TopicExchange topicExchange(){
return new TopicExchange(StringUtils.hasText(env().getProperty(ENV_EXCHANGE_NAME)) ?
env().getProperty(ENV_EXCHANGE_NAME) : DEFAULT_EXCHANGE_NAME);
}
//将创建好的队列和交换机通过指定routingkey绑定
@Bean
public Binding queueBinding(){
return BindingBuilder.bind(queue()).
to(topicExchange()).
with(StringUtils.hasText(env().getProperty(ENV_ROUTINGKEY))?
env().getProperty(ENV_ROUTINGKEY) : DEFAULT_ROUTINGKEY);
}
//设置断连重发类
@Bean
public RetryTemplate retryTemplate(){
long interval = NumberUtils.toLong(env().getProperty(ENV_BACKOFF4ASYN_INTERVAL)) >=100?
NumberUtils.toLong(env().getProperty(ENV_BACKOFF4ASYN_INTERVAL)) :
DEFAULT_BACKOFF_INTERVAL;
double multiplier = NumberUtils.toDouble(env().getProperty(ENV_BACKOFF4ASYN_MULTIPLIER),0) >0?
NumberUtils.toDouble(env().getProperty(ENV_BACKOFF4ASYN_MULTIPLIER)) :
DEFAULT_BACKOFF_MULTIPLIER;
long max_interval = NumberUtils.toLong(env().getProperty(ENV_BACKOFF4ASYN_MAXINTERVAL),0) >1000?
NumberUtils.toLong(env().getProperty(ENV_BACKOFF4ASYN_MAXINTERVAL)) :
DEFAULT_BACKOFF_MAXINTERVAL;
ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
//initial-interval(>100)
backOffPolicy.setInitialInterval(interval);
backOffPolicy.setMultiplier(multiplier);
//max-interval(>1000)
backOffPolicy.setMaxInterval(max_interval);
RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setBackOffPolicy(backOffPolicy);
logger.info("*******rabbit retry-template******");
logger.info("[interval]{}",interval);
logger.info("[multiplier]{}",multiplier);
logger.info("[max_interval]{}",max_interval);
return retryTemplate;
}
//将生产者发送类和断连重发类绑定在一起的方法
private void setRetryTemplate(RabbitTemplate _template){
if(StringUtils.hasText(env().getProperty(ENV_BACKOFF4ASYN_OPEN)) &&
0 == env().getProperty(ENV_BACKOFF4ASYN_OPEN).compareToIgnoreCase("true")){
_template.setRetryTemplate(retryTemplate());
logger.info("[retry]{}",true);
}else{
logger.info("[retry]{}",false);
}
}
//生产者操作类
@Bean
public RabbitTemplate rabbitTemplate(){
RabbitTemplate template = new RabbitTemplate(connectionFactory());
//向操作类中设置已经和队列绑定好了的交换机
template.setExchange(topicExchange().getName());
//设置发出去的信息转化的方式，特别注意，发送者的消息处理方式必须和接收方一致
template.setMessageConverter(new FastJsonMessageConvert());
//向template中设置如果断连的重发类
setRetryTemplate(template);
logger.info("*******rabbit template*****");
logger.info("[exchange]:",topicExchange().getName());
logger.info("[message.convert]",FastJsonMessageConvert.class.getSimpleName());
return template;
}
//管理类
@Bean
public AmqpAdmin rabbitAdmin() {
return new RabbitAdmin(connectionFactory());
}
//设置消费者[SimpleMessageListenerContainer<--MessageListenerAdapter<--FastJsonMessageConvert]
@Bean
public SimpleMessageListenerContainer messageListenerContainer(){
//设置消费者连接信息
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
container.setQueueNames(queue().getName());
//设置监听模板
container.setMessageListener(messageListenerAdapter());
//设置应答方式
container.setAcknowledgeMode(AcknowledgeMode.NONE);
return container;
}
@Bean
public MessageListenerAdapter messageListenerAdapter() {
DataPackageHandler handler = new DataPackageHandler();
//设置消费者处理信息的方式和转化信息的方式
MessageListenerAdapter adapter = new MessageListenerAdapter(handler,new FastJsonMessageConvert());
//指定是消息处理类中的哪一个方法
adapter.setDefaultListenerMethod("messageHandler");
return adapter;
}
}
DataPackageHandler处理消息类
package com.sunland.demo.configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.localhost.demo.PojoTest;
public class DataPackageHandler {
private static Logger logger = LoggerFactory.getLogger(DataPackageHandler.class.getName());
public void messageHandler(PojoTest obj){
try{
System.out.println("DataPackageHandler"+obj.getAge()+":"+obj.getName());
}catch(Exception e){
e.printStackTrace();
}
}
}
生产者发送消息类
package com.sunland.demo.configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitGatewaySupport;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
public class AsyncGateway extends RabbitGatewaySupport {
private Logger log = LoggerFactory.getLogger(AsyncGateway.class);
public void send(final Object obj){
try {
String routingKey;
// if(obj instanceof IllegalVehicle){
// routingKey = "queue.itaxi.illegal";
// }
// //[dispatch]
// else
routingKey = "mq_routingkey";
// rabbitTemplate.convertAndSend(routingKey, obj);
this.getRabbitTemplate().convertAndSend(routingKey, obj,
new MessagePostProcessor() {
public Message postProcessMessage(Message message)
throws AmqpException {
//
// ---set correlation data so that the replies can
// be correlated to the request
message.getMessageProperties().setType(obj.getClass().getName());
return message;
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
}
一个基本的spring整合rabbitmq的例子完成