RabbitMQ基于Spring AMQP的Java Config 简单配置实例

1、项目基于Maven，在pom.xml中引入相关依赖。

<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>${rabbitmq.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>${spring-rabbit.version}</version>
</dependency>

2、RabbitMQ的Java Config 部分

package com.my.rabbitmq.appconfig;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.springframework.amqp.core.AbstractExchange;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Binding.DestinationType;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.core.Queue;
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.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.my.rabbitmq.util.FastJsonMessageConverter;
import com.my.rabbitmq.consumer.RabbitMQConsumer;
import com.my.rabbitmq.producer.RabbitMQProducer;

/**
*
* @description rabbitMQ config
*
* @author yuanzi
* @time 2017年8月23日 下午2:46:01
*/
@Configuration
public class RabbitmqConfig {

private static Logger log = LogManager.getLogger(RabbitmqConfig.class);

//读取rabbitmq配置文件
private static InputStream configFile = RabbitmqConfig.class.getClassLoader()
.getResourceAsStream("rabbitmq.properties");
private static Properties rabbitmqProperties = new Properties();

static {
try {
rabbitmqProperties.load(configFile);
} catch (IOException e) {
e.printStackTrace();
}
}

//配置生产者ConnectionFactory，配置基础连接信息
@Bean(name = "ProducerConnectionFactory")
public ConnectionFactory ProducerConnectionFactory() throws IOException {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(
rabbitmqProperties.getProperty("rabbitmq.host"));
connectionFactory.setPort(Integer.parseInt(rabbitmqProperties.getProperty("rabbitmq.port")));
connectionFactory.setUsername(rabbitmqProperties.getProperty("rabbitmq.username"));
connectionFactory.setPassword(rabbitmqProperties.getProperty("rabbitmq.password"));
return connectionFactory;
}

//配置消费者ConnectionFactory，配置基础连接信息
@Bean(name = "ConsumerConnectionFactory")
public ConnectionFactory ConsumerConnectionFactory() throws IOException {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(
rabbitmqProperties.getProperty("rabbitmq.host"));
connectionFactory.setPort(Integer.parseInt(rabbitmqProperties.getProperty("rabbitmq.port")));
connectionFactory.setUsername(rabbitmqProperties.getProperty("rabbitmq.username"));
connectionFactory.setPassword(rabbitmqProperties.getProperty("rabbitmq.password"));
return connectionFactory;
}

//配置RabbitAdmin，在配置了多个ConnectionFactory的情况下，需要配置RabbitAdmin，否则无法自动在rabbitmq服务器中注册交换机队列等。
@Bean
public RabbitAdmin rabbitAdmin() throws IOException {
return new RabbitAdmin(ConsumerConnectionFactory());
}

//配置生产者Template
@Bean(name = "ProducerRabbitTemplate")
public RabbitTemplate ProducerRabbitTemplate() throws IOException {
RabbitTemplate producerRabbitTemplate = new RabbitTemplate();
producerRabbitTemplate.setConnectionFactory(ProducerConnectionFactory());
producerRabbitTemplate.setExchange(rabbitmqProperties.getProperty("rabbitmq.producerExchange"));
//配置生产者信息转换器，封装发送信息的格式
producerRabbitTemplate.setMessageConverter(ProducerMessageConverter());
return producerRabbitTemplate;
}

//配置消费者Template
@Bean(name = "ConsumerRabbitTemplate")
public RabbitTemplate ConsumerRabbitTemplate() throws IOException {
RabbitTemplate consumerRabbitTemplate = new RabbitTemplate(ConsumerConnectionFactory());
consumerRabbitTemplate.setExchange(rabbitmqProperties.getProperty("rabbitmq.consumerExchange"));
return consumerRabbitTemplate;
}

//配置生产者交换机类型
@Bean(name = "ProducerExchange")
public AbstractExchange ProducerExchange() throws IOException {
String exchangeType = rabbitmqProperties.getProperty("rabbitmq.exchangeType");
if (exchangeType.contains("F")) {
FanoutExchange fanoutExchange = new FanoutExchange(
rabbitmqProperties.getProperty("rabbitmq.producerExchange"), true, false);
log.info("ProducerExchange-Fanout 生产者：广播路由");
return fanoutExchange;
} else if (exchangeType.contains("T")) {
TopicExchange topicExchange = new TopicExchange(rabbitmqProperties.getProperty("rabbitmq.producerExchange"),
true, false);
log.info("ProducerExchange-Topic 生产者：多路广播路由");
return topicExchange;
} else {
DirectExchange directExchange = new DirectExchange(
rabbitmqProperties.getProperty("rabbitmq.producerExchange"), true, false);
log.info("ProducerExchange-Direct 生产者：直接路由");
return directExchange;
}
}

//配置消费者交换机类型
@Bean(name = "ConsumerExchange")
public AbstractExchange ConsumerExchange() throws IOException {
String exchangeType = rabbitmqProperties.getProperty("rabbitmq.exchangeType");
if (exchangeType.contains("F")) {
FanoutExchange fanoutExchange = new FanoutExchange(
rabbitmqProperties.getProperty("rabbitmq.consumerExchange"), true, false);
log.info("ProducerExchange-Fanout 消费者：广播路由");
return fanoutExchange;
} else if (exchangeType.contains("T")) {
TopicExchange topicExchange = new TopicExchange(rabbitmqProperties.getProperty("rabbitmq.consumerExchange"),
true, false);
log.info("ProducerExchange-Topic 消费者：多路广播路由");
return topicExchange;
} else {
DirectExchange directExchange = new DirectExchange(
rabbitmqProperties.getProperty("rabbitmq.consumerExchange"), true, false);
log.info("ProducerExchange-Direct 消费者：直接路由");
return directExchange;
}
}

//配置参数：生产者队列名称，重启rabbitmq服务时队列还是否存在，只被一个connection使用并且在connection关闭时queue是否被删除，当最后一个consumer取消订阅时queue是否被删除
@Bean(name = "ProducerQueue")
public Queue ProducerQueue() throws IOException {
Queue producerQueue = new Queue(rabbitmqProperties.getProperty("rabbitmq.producerQueue"), true, false, false);
return producerQueue;
}

//配置参数：消费者队列名称，重启rabbitmq服务时队列还是否存在，只被一个connection使用并且在connection关闭时queue是否被删除，当最后一个consumer取消订阅时queue是否被删除
@Bean(name = "ConsumerQueue")
public Queue ConsumerQueue() throws IOException {
Queue consumerQueue = new Queue(rabbitmqProperties.getProperty("rabbitmq.consumerQueue"), true, false, false);
return consumerQueue;
}

//配置生产者消息转换器
@Bean(name = "jsonMessageConverter")
public AbstractMessageConverter ProducerMessageConverter() {
FastJsonMessageConverter jsonMessageConverter = new FastJsonMessageConverter();
return jsonMessageConverter;
}

//配置生产者队列与交换机之间的绑定关系和routingKey
@Bean(name = "ProducerBinding")
public Binding ProducerBinding() {
Binding binding = new Binding(rabbitmqProperties.getProperty("rabbitmq.producerQueue"), DestinationType.QUEUE,
rabbitmqProperties.getProperty("rabbitmq.producerExchange"),
rabbitmqProperties.getProperty("rabbitmq.producerRoutingKey"), null);
return binding;
}

//配置消费者队列与交换机之间的绑定关系和routingKey
@Bean(name = "ConsumerBinding")
public Binding ConsumerBinding() {
Binding binding = new Binding(rabbitmqProperties.getProperty("rabbitmq.consumerQueue"), DestinationType.QUEUE,
rabbitmqProperties.getProperty("rabbitmq.consumerExchange"),
rabbitmqProperties.getProperty("rabbitmq.consumerRoutingKey"), null);
return binding;
}

//实例化生产者代码，配置生产者要投递信息的RoutingKey
@Bean(name = "RabbitMQProducer")
public RabbitMQProducer RabbitMQProducer() {
RabbitMQProducer rabbitMQProducer = new RabbitMQProducer(
rabbitmqProperties.getProperty("rabbitmq.producerSendRoutingKey"));
return rabbitMQProducer;
}

//实例化消费者代码
@Bean(name = "RabbitMQConsumer")
public RabbitMQConsumer RabbitMQConsumer() {
RabbitMQConsumer rabbitMQConsumer = new RabbitMQConsumer();
return rabbitMQConsumer;
}

//配置监听模式，当有消息到达时会通知监听在对应的队列上的监听对象
@Bean(name = "ListenerContainer")
public AbstractMessageListenerContainer MessageListenerContainer() throws IOException {
SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer();
simpleMessageListenerContainer.setConnectionFactory(ConsumerConnectionFactory());
simpleMessageListenerContainer.setQueueNames(rabbitmqProperties.getProperty("rabbitmq.consumerQueue"));
simpleMessageListenerContainer.setMessageListener(RabbitMQConsumer());
return simpleMessageListenerContainer;
}

}

3、rabbitmq.properties 文件

#单一交换机配置参数
#rabbitmq server地址
rabbitmq.host = 127.0.0.1
#rabbitmq 端口
rabbitmq.port = 5672
#生产者队列名称
rabbitmq.producerQueue = d_queue_producer
#生产者RoutingKey
rabbitmq.producerRoutingKey = d_key_producer
#消费者队列名称
rabbitmq.consumerQueue = d_queue_consumer
#消费者RoutingKey
rabbitmq.consumerRoutingKey = d_key_consumer
#交换机类型 D为单一传播交换机类型，F为广播交换机类型，T为多路广播交换机类型
rabbitmq.exchangeType = D
#生产者需要绑定的交换机
rabbitmq.producerExchange = d_exchange
#消费者需要绑定的交换机
rabbitmq.consumerExchange = d_exchange
#rabbitmq用户名称
rabbitmq.username = guest
#rabbitmq用户密码
rabbitmq.password = guest
#rabbitmq生产者发送消息时的RoutingKey
rabbitmq.producerSendRoutingKey = d_key_consumer
---------------------------------------------分割线----------------------------------------------
#广播交换机配置参数，广播交换机下routingKey失效
#rabbitmq server地址
rabbitmq.host = 127.0.0.1
#rabbitmq 端口
rabbitmq.port = 5672
#生产者队列名称
rabbitmq.producerQueue = f_queue_producer
#生产者RoutingKey
rabbitmq.producerRoutingKey = f_key_producer
#消费者队列名称
rabbitmq.consumerQueue = f_queue_consumer
#消费者RoutingKey
rabbitmq.consumerRoutingKey = f_key_consumer
#交换机类型 D为单一传播交换机类型，F为广播交换机类型，T为多路广播交换机类型
rabbitmq.exchangeType = F
#生产者需要绑定的交换机
rabbitmq.producerExchange = f_exchange
#消费者需要绑定的交换机
rabbitmq.consumerExchange = f_exchange
#rabbitmq用户名称
rabbitmq.username = guest
#rabbitmq用户密码
rabbitmq.password = guest
#rabbitmq生产者发送消息时的RoutingKey
rabbitmq.producerSendRoutingKey = f_test

---------------------------------------------分割线----------------------------------------------
#多路广播交换机配置参数
#配置rootingkey时，“#”表示0个或若干个关键字，“*”表示一个关键字。例如“log.*”能与“log.warn”匹配，无法
#与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配
#rabbitmq server地址
rabbitmq.host = 127.0.0.1
#rabbitmq 端口
rabbitmq.port = 5672
#生产者队列名称
rabbitmq.producerQueue = t_queue_producer
#生产者RoutingKey
rabbitmq.producerRoutingKey = t_key_producer
#消费者队列名称
rabbitmq.consumerQueue = t_queue_consumer
#消费者RoutingKey
rabbitmq.consumerRoutingKey = #.test
#交换机类型 D为单一传播交换机类型，F为广播交换机类型，T为多路广播交换机类型
rabbitmq.exchangeType = T
#生产者需要绑定的交换机
rabbitmq.producerExchange = t_exchange2
#消费者需要绑定的交换机
rabbitmq.consumerExchange = t_exchange2
#rabbitmq用户名称
rabbitmq.username = guest
#rabbitmq用户密码
rabbitmq.password = guest
#rabbitmq生产者发送消息时的RoutingKey
rabbitmq.producerSendRoutingKey = t.test

4、生产者代码

package com.my.rabbitmq.producer;

import javax.annotation.Resource;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
* @description RabbitMQ消息发送
*
* @author yuanzi
* @time 2017年8月1日 下午3:16:08
*/
public class RabbitMQProducer {

private static Logger log = LogManager.getLogger(RabbitMQProducer.class);

@Resource(name = "ProducerRabbitTemplate")
private RabbitTemplate rabbitTemplate;

private String routingKey;

public RabbitMQProducer(String routingKey) {
this.routingKey = routingKey;
}

public void sendDataToQueue(Object message) {

rabbitTemplate.convertAndSend(this.routingKey, message);

log.info("rabbitMQ producer send message:" + message);
}
}

5、消费者代码

package com.my.rabbitmq.consumer;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
/**
*
* @description RabbitMQ消息接收
*
* @author yuanzi
* @time 2017年8月23日 下午3:38:03
*/
public class RabbitMQConsumer implements MessageListener {

private static Logger log = LogManager.getLogger(RabbitMQConsumer.class);

public void onMessage(Message message) {
String data = new String(message.getBody());
log.info("rabbitMQ consumer receive message：" + data);
}
}

6、消息转换器的配置

package com.my.rabbitmq.util;

import java.io.UnsupportedEncodingException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;

import net.sf.json.JSONObject;

/**
*
* @description 对发送的信息进行json格式的封装
*
* @author yuanzi
* @time 2017年8月1日 下午3:13:30
*/
public class FastJsonMessageConverter extends AbstractMessageConverter {

public static final String DEFAULT_CHARSET = "UTF-8";

@SuppressWarnings("static-access")
@Override
protected Message createMessage(Object message, MessageProperties messageProperties) {
byte[] bytes = null;

JSONObject json = new JSONObject();
json.put("data", message);

try {
bytes = json.toString().getBytes(this.DEFAULT_CHARSET);
} catch (UnsupportedEncodingException e) {
throw new MessageConversionException("Failed to convert Message content:" + e);
}

if (bytes != null) {
messageProperties.setContentLength(bytes.length);
}
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setContentEncoding(this.DEFAULT_CHARSET);

return new Message(bytes, messageProperties);
}

@Override
public Object fromMessage(Message message) throws MessageConversionException {
return null;
}

}