5 RabbitMQ整合Spring

重点

• RabbitAdmin
• Spring AMQP 声明
• RabbitTemplate
• SimpleMessageListenerContainer
• MessageListenerAdaper
• MessageConverter

使用RabbitMQ结合Spring AMQP

RabbitAdmin

• 用于声明RabbitMQ相关配置、操做RabbitMQ
• autoStartup设为true:表示Spring容器启动时自动初始化RabbitAdmin
• 底层实现:从Spring容器中获取Exchange、Binding、RoutingKey以及Queue的@Bean声明
• rabbitTemplate的execute方法执行对应的声明等操做

SpringAMQP声明

• exchange
  • TopicExchange
  • FanoutExchange
  • DirectExchange
• queue
• binding
  • BindingBuilder

spring使用@bean声明exchange queue binding 例子以下:

@Bean  
 public TopicExchange exchange001() {  
        return new TopicExchange("topic001", true, false);  
    }  

    @Bean  
    public Queue queue001() {  
        return new Queue("queue001", true); //队列持久 
    }  
    
    @Bean  
    public Binding binding001() {  
        return BindingBuilder.bind(queue001()).to(exchange001()).with("spring.*");  
    }  
复制代码

RabbitTemplate

• 与SpringAMQP整合的时候进行发送消息的关键类
• 提供丰富的发送消息方法，包括可靠性投递消息方法、回调监听消息接口
• ConfirmCallback
• ReturnCallback
• 与Spring整合时须要实例化，可是与Springboot整合时不须要，在配置文件添加配置便可

如下是RabbitTemplate实例化的例子:

@Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    	RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setConfirmCallback(//);
        rabbitTemplate.setReturnCallback(//);
    	return rabbitTemplate;
    }
复制代码

使用RabbitTemplate sendMessage

• rabbitTemplate.convertAndSend方法是主要的发送消息的方法
• MessageProperties 用于构造消息体
• MessagePostProcessor:消息发送以后对消息进行的设置

例子：

@Test
	public void testSendMessage() throws Exception {
		//1 建立消息
		MessageProperties messageProperties = new MessageProperties();
		messageProperties.getHeaders().put("desc", "信息描述..");
		messageProperties.getHeaders().put("type", "自定义消息类型..");
		Message message = new Message("Hello RabbitMQ".getBytes(), messageProperties);
		
		rabbitTemplate.convertAndSend("topic001", "spring.amqp", message, new MessagePostProcessor() {
			@Override
			public Message postProcessMessage(Message message) throws AmqpException {
				System.err.println("------添加额外的设置---------");
				message.getMessageProperties().getHeaders().put("desc", "额外修改的信息描述");
				message.getMessageProperties().getHeaders().put("attr", "额外新加的属性");
				return message;
			}
		});
	}
复制代码

SimpleMessageListenerContainer

简单消息容器 功能:

• 监听多个队列、自动启动、自动声明
• 设置事务特性、事务管理器、事务属性、事务容器、是否开启事务、回滚消息
• 设置消费者数量、最小最大数量、批量消费
• 设置消息确认和自动确认模式、是否重回队列、异常捕获函数
• 设置消费者标签生成策略、是否独占模式、消费者属性等
• 设置具体的监听器、消息转换器 message convert
• SimpleMessageListenerContainer能够进行动态设置，好比在运行中的应用能够动态的修改其消费者的大小、接收消息的模式等,能够基于此开发rabbitmq自定义后台管控平台

属性

• queues
• concurrentConsumers:当前消费者数
• maxConcurrentConsumers:最大消费者并发
• defaultRequeueRejected: 是否重回队列 false
• acknowledgeMode:消息确认模型 AUTO
• exposeListenerChannel:
• messageListener: 消息监听
• consumerTagStrategy:consumerTag生成策略

@Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
    	SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    	container.setQueues(queue001(), queue002(), queue003(), queue_image(), queue_pdf());
    	container.setConcurrentConsumers(1);
    	container.setMaxConcurrentConsumers(5);
    	container.setDefaultRequeueRejected(false);
    	container.setAcknowledgeMode(AcknowledgeMode.AUTO);
    	container.setExposeListenerChannel(true);
    	container.setConsumerTagStrategy(new ConsumerTagStrategy() {
			@Override
			public String createConsumerTag(String queue) {
				return queue + "_" + UUID.randomUUID().toString();
			}
		});
        }
复制代码

MessageListenerApapter 适配器模式

• extends AbstractAdaptableMessageListener 消息listener
• queueOrTagToMethodName 队列标识与方法名称组成的集合
• defaultListenerMethod 默认监听方法名称
• Delegate 委托对象:实际真实的委托对象

MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate);

//设定queue使用哪一个adapter方法处理
Map<String,String> queueOrTagToMethodName = new HashMap<>();
queueOrTagToMethodName.put("queue001","method1");
queueOrTagToMethodName.put("queue002","method2");
adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
//设置默认处理方法,默认处理方法是handleMessage
adapter.setDefaultListenerMethod("handleMessage");
//设置消息转换方式
adapter.setMessageConverter(new TextMessageConvert());
//消息监听
container.setMessageListener(adapter);
复制代码

MessageConverter 消息转换器

自定义经常使用转换器 都须要实现 MessageConverter接口

Json convert: Jackson2JsonMessageConverter 能够进行java对象的转换功能 DefaultJackson2JavaTypeMapper映射器: 能够进行java对象的映射关系 二进制 convert: 如Image、PDF、PPT、stream

Springboot整合RabbitMQ

基本概念

@EnableRabbit

@EnableRabbit和@Configuration一块儿使用，能够加在类或者方法上，这个注解开启了容器对注册的bean的@RabbitListener检查。

@RabbitListener

@RabbitListener用于注册Listener时使用的信息：如queue，exchange，key、ListenerContainerFactory和RabbitAdmin的bean name。

@RabbitListener(containerFactory = "rabbitListenerContainerFactory", bindings = @QueueBinding(
        value = @Queue(value = "${mq.config.queue}", durable = "true"),
        exchange = @Exchange(value = "${mq.config.exchange}", type = ExchangeTypes.TOPIC),
        key = "${mq.config.key}"), admin = "rabbitAdmin")
复制代码

扫描到bean带有该注解后，首先会将注解的内容封装到Endpoint对象中并和ListenerContainerFactory的实例一块儿添加到上面的RabbitListenerEndpointRegistry实例中。添加的时候会建立相应的ListenerContainer实例并添加Listener对象。

@RabbitHandler

@RabbitListener 和 @RabbitHandler结合使用，不一样类型的消息使用不一样的方法来处理。

• 不一样的消息类型走不一样的handler

public class CommandListener{

    @RabbitHandler
    public void handler1(ApiMessage msg){
        System.out.println(msg);
    }

    @RabbitHandler
    public void handler2(Map msg){
        System.out.println(msg);
    }
}
复制代码

生产端发送消息

• ConfirmCallback
• ReturnCallback
• CorrelationData
• convertAndSend

//回调函数: confirm确认 message ->send —> broker -> ack
	final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
		@Override
		public void confirm(CorrelationData correlationData, boolean ack, String cause) {
			System.err.println("correlationData: " + correlationData);
			System.err.println("ack: " + ack);
			if(!ack){
				System.err.println("异常处理....");
			}
		}
	};
	
	//回调函数: return返回 
	final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
		@Override
		public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String routingKey) {
			System.err.println("return exchange: " + exchange + ", routingKey: " 
				+ routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
		}
	};

//发送消息方法调用: 构建Message消息
	public void send(Object message, Map<String, Object> properties) throws Exception {
		MessageHeaders mhs = new MessageHeaders(properties);
		Message msg = MessageBuilder.createMessage(message, mhs);
		rabbitTemplate.setConfirmCallback(confirmCallback);
		rabbitTemplate.setReturnCallback(returnCallback);
		//id + 时间戳 全局惟一 
		CorrelationData correlationData = new CorrelationData("1234567890");
		rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);
	}
复制代码

消费端监听方式

• @RabbitListener
  • bindings @QueueBindingvalue
    • @Queue
      • durable 队列是否持久化
      • value 值
      • exchange @Exchange
        • type 交换机的类型
        • value 值
        • ignoreDeclarationExceptions 是否忽略声明异常
      • key routing Key
• @RabbitHandler
• onMessage
  • @Payload: message body
    • deliveryTag
    • basicAck
  • Channel:
  • @Headers: message headers

/** * *spring.rabbitmq.listener.order.queue.name=queue-2 spring.rabbitmq.listener.order.queue.durable=true spring.rabbitmq.listener.order.exchange.name=exchange-1 spring.rabbitmq.listener.order.exchange.durable=true spring.rabbitmq.listener.order.exchange.type=topic spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true spring.rabbitmq.listener.order.key=springboot.* * @param order * @param channel * @param headers * @throws Exception */
	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}", 
			durable="${spring.rabbitmq.listener.order.queue.durable}"),
			exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}", 
			durable="${spring.rabbitmq.listener.order.exchange.durable}", 
			type= "${spring.rabbitmq.listener.order.exchange.type}", 
			ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),
			key = "${spring.rabbitmq.listener.order.key}"
			)
	)
	@RabbitHandler
	public void onOrderMessage(@Payload com.bfxy.springboot.entity.Order order, Channel channel, @Headers Map<String, Object> headers) throws Exception {
		System.err.println("--------------------------------------");
		System.err.println("消费端order: " + order.getId());
		Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
		//手工ACK
		channel.basicAck(deliveryTag, false);
	}
复制代码

Spring cloud stream with RabbitMQ

特色: Spring cloud stream 生产端和消费端可使用不一样的中间件

总体架构核心概念图:

基本概念

Barista

Barista接口: Barista接口是定义来做为后面类的参数. 这一接口定义了通道类型和通道名称,通道名称是做为配置用，通道类型则决定了app会使用这一通道进行发送消息仍是从中接收消息。

• @Output 输入注解
• @Input 输出注解
• @StreamListener 监听消息

使用Spring Cloud Stream很是简单，使用好这3个注解便可，在实现高性能消息的生产和消费场景很是适合,可是使用SpringCloudStream框架有一个很是大的问题就是不能实现可靠性的投递,也就是没法保证消息的100%可靠性,会存在少许消息丢失的问题。缘由是为了兼容Kafka。

使用流程图