1. 首页
  2. 文章中心
  3. java

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

68 阅读 0 评论
我是靠谱客的博主 炙热蜜粉，最近开发中收集的这篇文章主要介绍RabbitMQ基于Spring AMQP的Java Config 简单配置实例，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

2019独角兽企业重金招聘Python工程师标准>>> hot3.png

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

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

}

 

转载于:https://my.oschina.net/u/2478308/blog/1519334

最后

以上就是炙热蜜粉为你收集整理的RabbitMQ基于Spring AMQP的Java Config 简单配置实例的全部内容，希望文章能够帮你解决RabbitMQ基于Spring AMQP的Java Config 简单配置实例所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
点赞(45)

相关文章

rocketmq生产者和消费者
rocketmq生产者和消费者
RabbitMQ的生产者与消费者
RabbitMQ的生产者与消费者
Spring Boot 与消息 （JMS、AMQP、RabbitMQ）RabbitMQ教程 - 鸟哥的专栏 - CSDN博客
Spring Boot 与消息 （JMS、AMQP、RabbitMQ）RabbitMQ教程 - 鸟哥的专栏 - CSDN博客
SpringBoot 集成 AMQP 连接 RabbitMQSpringBoot 集成 AMQP 连接 RabbitMQ
SpringBoot 集成 AMQP 连接 RabbitMQSpringBoot 集成 AMQP 连接 RabbitMQ
RabbitMQ基于Spring AMQP的Java Config 简单配置实例
RabbitMQ基于Spring AMQP的Java Config 简单配置实例
为什么使用消息队列,为什么使用RabbitMQ、springAMQP
为什么使用消息队列,为什么使用RabbitMQ、springAMQP
Spring集成RabbitMQ测试
Spring集成RabbitMQ测试
RabbitMQ和Spring AMQP学习二(消息模型详解、Spring AMQP启动流程分析-超详细)RabbitMQ和Spring AMQP学习二
RabbitMQ和Spring AMQP学习二(消息模型详解、Spring AMQP启动流程分析-超详细)RabbitMQ和Spring AMQP学习二

评论列表共有 0 条评论

立即
投稿
返回
顶部