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

      本实例尝试针对 Direct exchange、Fanout exchange、Topic exchange三种路由形式进行了消费者和生产者的集中实现，若有不对之处烦请读者指出。

一、项目基于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>

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

}

三、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

四、生产者代码

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

五、消费者代码

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

六、消息转换器的配置 

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

}