zoukankan      html  css  js  c++  java

  • RabbitMD大揭秘

    RabbitMD大揭秘

    欢迎关注H寻梦人公众号

    image-20210120103201067

    通过SpringBoot整合RabbitMQ的案例来说明，RabbitMQ相关的各个属性以及使用方式；并通过相关源码深刻理解。

    Queue(消息队列)

    Queue(消息队列) 用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

    RabbitMQ 中消息只能存储在 队列 中，这一点和 Kafka 这种消息中间件相反。Kafka 将消息存储在 topic（主题） 这个逻辑层面，而相对应的队列逻辑只是topic实际存储文件中的位移标识。 RabbitMQ 的生产者生产消息并最终投递到队列中，消费者可以从队列中获取消息并消费。

    多个消费者可以订阅同一个队列，这时队列中的消息会被平均分摊（Round-Robin，即轮询）给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，这样避免的消息被重复消费。

    RabbitMQ 不支持队列层面的广播消费,如果有广播消费的需求，需要在其上进行二次开发,这样会很麻烦，不建议这样做。

    交换机的类型：

    交换机主要包括如下4种类型：

    1. Direct exchange（直连交换机）
    2. Fanout exchange（扇型交换机）
    3. Topic exchange（主题交换机）
    4. Headers exchange（头交换机）

    1. 基本配置

    1.1 配置文件属性说明

    RabbitMQ最基本的基础配置如下：

    server:
  port: 11000

spring:
  application:
    name: rabbitmq-test

  rabbitmq:
#    单机IP配置
#    host: rabbitmq-host
#    port: 5672

#    集群IP配置
    addresses: rabbitmq01-host:5672,rabbitmq02-host:5672

#    用户名和密码，默认都是guest
    username: xiongmin
    password: xiongmin

#    交换器名可以不设置默认 【"" --> /】 交换器
    virtual-host: /rabbitmq_test
    publisher-returns: true # 发送者开启 return 确认机制
    publisher-confirm-type: correlated # 发送者开启 confirm 确认机制 等价于 spring.rabbitmq.publisher-returns=true
    default-exchange: default_exchange
    listener:
      simple:
        acknowledge-mode: manual # 设置消费端手动 ack
        retry:
          enabled: true # 支持重试
          
          
----------------------------------------------------------------------------------------------

server:
  port: 11000

spring:
  application:
    name: rabbitmq-test

  rabbitmq:
#    单机IP配置
#    host: rabbitmq-host
#    port: 5672

#    集群IP配置
    addresses: rabbitmq01-host:5672,rabbitmq02-host:5672

#    用户名和密码，默认都是guest
    username: xiongmin
    password: xiongmin

#    交换器名可以不设置默认 【"" --> /】 交换器
    virtual-host: /rabbitmq_test
    publisher-returns: true # 发送者开启 return 确认机制
    publisher-confirm-type: correlated # 发送者开启 confirm 确认机制 等价于 spring.rabbitmq.publisher-returns=true
    #连接超时时间
    connection-timeout: 15000
    # 使用return-callback时必须设置mandatory为true
    template:
      mandatory: true
    default-exchange: default_exchange
    # 消费端配置
    listener:
      simple:
        retry:
          enabled: true # 支持重试
        #消费端
        concurrency: 5
        #最大消费端数
        max-concurrency: 10
        #自动签收auto  手动 manual
        acknowledge-mode: manual # 设置消费端手动 ack
        #限流（海量数据，同时只能过来一条）
        prefetch: 1

    1.2 配置类说明

    package com.cli.springboot_rabbitmq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
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.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author xiongmin
 * @Description
 * @Date 2021/2/25 18:31
 * @Version 1.0
 **/
@Configuration
public class RabbitMQConfig {

    @Value("${spring.rabbitmq.addresses}")
    private String addresses;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.default-exchange}")
    private String defaultExchange;

    @Autowired
    private ConfirmCallbackService confirmCallbackService;
    @Autowired
    private ReturnCallbackService returnCallbackService;


    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(addresses);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
//        connectionFactory.setPublisherConfirms(rabbitmqProps.isPublisherConfirms()); //消息回调，必须要设置
        return connectionFactory;
    }

    /**
     * 使用的自己的创建的RabbitMQ 完全没有使用到RabbitMQ相关的默认配置，即使在yaml文件中配置了消费者手动确认，使用如下的rabbitMQ 也是无效的，
     * 因为RabbitTemplate相关的ConnectionFactory 没有设置消费者手动确认消息，这里不会使用到yaml的配置
     * @return
     */
    @Bean(value = "rabbitTemplateMessaging")
    public RabbitTemplate rabbitTemplateMessaging() {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        // 消息类型转换器 -- 数据转换为json存入消息队列
        rabbitTemplate.setMessageConverter(jackson2MessageConverter());
        // 设置默认的交换机，如果发送的消息没有指定交换机，则使用默认的交换机
        rabbitTemplate.setExchange(defaultExchange);
        /**
         * mandatory：交换器无法根据自身类型和路由键找到一个符合条件的队列时的处理方式
         * true：RabbitMQ会调用Basic.Return命令将消息返回给生产者
         * false：RabbitMQ会把消息直接丢弃
         */
        rabbitTemplate.setMandatory(true);
        /**
         * 消费者确认收到消息后，手动ack回执回调处理
         */
        rabbitTemplate.setConfirmCallback(confirmCallbackService);
        /**
         * 消息投递到队列失败回调处理
         */
        rabbitTemplate.setReturnsCallback(returnCallbackService);


        // 服务端响应发送到的队列 reply-address 格式: exchange/routingKey
        rabbitTemplate.setReplyAddress("messaging/messaging-response");
        // 设置回复和接受消息的时间，单位为毫秒
        rabbitTemplate.setReplyTimeout(20000);
        rabbitTemplate.setReceiveTimeout(20000);
        return rabbitTemplate;
    }

    /**
     * @param connectionFactory connectionFactory属性信息会直接是用yaml配置文件中配置的
     * @return
     */
    @Bean(value = "rabbitTemplateInit")
    public RabbitTemplate rabbitTemplateInit(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        // 消息类型转换器 -- 数据转换为json存入消息队列
        rabbitTemplate.setMessageConverter(jackson2MessageConverter());
        // 设置默认的交换机，如果发送的消息没有指定交换机，则使用默认的交换机
        rabbitTemplate.setExchange(defaultExchange);
        /**
         * mandatory：交换器无法根据自身类型和路由键找到一个符合条件的队列时的处理方式
         * true：RabbitMQ会调用Basic.Return命令将消息返回给生产者
         * false：RabbitMQ会把消息直接丢弃
         */
        rabbitTemplate.setMandatory(true);
        /**
         * 消费者确认收到消息后，手动ack回执回调处理
         */
        rabbitTemplate.setConfirmCallback(confirmCallbackService);
        /**
         * 消息投递到队列失败回调处理
         */
        rabbitTemplate.setReturnsCallback(returnCallbackService);


        // 服务端响应发送到的队列 reply-address 格式: exchange/routingKey
        rabbitTemplate.setReplyAddress("messaging/messaging-response");
        // 设置回复和接受消息的时间，单位为毫秒
        rabbitTemplate.setReplyTimeout(20000);
        rabbitTemplate.setReceiveTimeout(20000);
        return rabbitTemplate;
    }

    /**
     * 的作用解释如下：
     * 数据转换为json存入消息队列
     * [[RabbitMQ]Jackson2JsonMessageConverter转换实体类常的问题](https://blog.csdn.net/qq_31897023/article/details/103875594)
     * [Springboot Rabbitmq 使用Jackson2JsonMessageConverter 消息传递后转对象](https://www.cnblogs.com/timseng/p/11688019.html)
     * @return
     */
    @Bean
    public Jackson2JsonMessageConverter jackson2MessageConverter() {
        Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
        return converter;
    }

//--------------------------配置相关的Exchange和相关的Queue----------------------------------
    
    /**
     * 创建topic模式的交换器
     * @return
     */
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topicExchange",true,false);
    }

    /**
     * 创建fanout模式的交换器
     * 发布订阅模式
     * 发布订阅是交换机针对队列来说的，一个消息可投入一个或多个队列
     * 注意：多个消费者可以订阅同一个队列，这时队列中的消息会被平均分摊（Round-Robin，即轮询）给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，这样避免的消息被重复消费。
     * @return
     */
    @Bean
    FanoutExchange fanoutExchange() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效; 如果持久性，则RabbitMQ重启后，交换机还存在
        // autoDelete:是否自动删除，当所有与之绑定的消息队列都完成了对此交换机的使用后，删掉它
        return new FanoutExchange("fanoutExchange",true,false);
    }

    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        return new DirectExchange("TestDirectExchange",true,false);
    }

    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);

        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("TestDirectQueue",true);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }


    // 默认的 Direct交换机 起名：DefaultDirectExchange
    @Bean
    DirectExchange DefaultDirectExchange() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        return new DirectExchange(defaultExchange,true,false);
    }

    //队列 起名：DefaultDirectQueue
    @Bean
    public Queue DefaultDirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);

        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("DefaultDirectQueue",true);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：DefaultDirectRouting
    @Bean
    Binding bindingDefaultDirect() {
        return BindingBuilder.bind(DefaultDirectQueue()).to(DefaultDirectExchange()).with("DefaultDirectRouting");
    }


    /**
     *  创建三个队列 ：fanout.A   fanout.B  fanout.C
     *  将三个队列都绑定在交换机 fanoutExchange 上
     *  因为是扇型交换机, 路由键无需配置,配置也不起作用
     */
    @Bean
    public Queue queueA() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue queueB() {
        return new Queue("fanout.B");
    }

    @Bean
    public Queue queueC() {
        return new Queue("fanout.C");
    }

    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }

    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }

    @Bean
    Binding bindingExchangeC() {
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }



    //绑定键
    public final static String MAN = "topic.MAN";
    public final static String WOMAN = "topic.WOMAN";

    @Bean
    public Queue firstQueue() {
        return new Queue(RabbitMQConfig.MAN);
    }

    @Bean
    public Queue secondQueue() {
        return new Queue(RabbitMQConfig.WOMAN);
    }



    //将firstQueue和topicExchange绑定,而且绑定的键值为topic.MAN
    //这样只要是消息携带的路由键是topic.MAN,才会分发到该队列
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(MAN);
    }

    //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
    // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }



}
    交换机的属性

    除交换机类型外，在声明交换机时还可以附带许多其他的属性，其中最重要的几个分别是：

    • Name：交换机名称
    • Durability：是否持久化。如果持久性，则RabbitMQ重启后，交换机还存在
    • Auto-delete：当所有与之绑定的消息队列都完成了对此交换机的使用后，删掉它
    • Arguments：扩展参数

    2. 消息转换器

    可以注意到的是上面的配置中RabbitTemplate设置的消息转换器是Jackson2JsonMessageConverter；下面将对消息转换器说明

    消息转换器接口源码：

    /*
 * Copyright 2002-2019 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.amqp.support.converter;

import java.lang.reflect.Type;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.lang.Nullable;

/**
 * Message converter interface.
 *
 * @author Mark Fisher
 * @author Mark Pollack
 * @author Gary Russell
 */
public interface MessageConverter {

	/**
	 * Convert a Java object to a Message. 将消息对象转换成java对象。
	 * @param object the object to convert
	 * @param messageProperties The message properties.
	 * @return the Message
	 * @throws MessageConversionException in case of conversion failure
	 */
	Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException;

	/**
	 * Convert a Java object to a Message. 将java对象和属性对象转换成Message对象。
	 * The default implementation calls {@link #toMessage(Object, MessageProperties)}.
	 * @param object the object to convert
	 * @param messageProperties The message properties.
	 * @param genericType the type to use to populate type headers.
	 * @return the Message
	 * @throws MessageConversionException in case of conversion failure
	 * @since 2.1
	 */
	default Message toMessage(Object object, MessageProperties messageProperties, @Nullable Type genericType)
			throws MessageConversionException {

			return toMessage(object, messageProperties);
	}

	/**
	 * Convert from a Message to a Java object.
	 * @param message the message to convert
	 * @return the converted Java object
	 * @throws MessageConversionException in case of conversion failure
	 */
	Object fromMessage(Message message) throws MessageConversionException;

}

    可以通过实现MessageConverter接口，实现自定义的消息转换器，如下：

    import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;

public class TestMessageConverter implements MessageConverter {


    @Override
    public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        System.out.println("=======toMessage=========");
        return new Message(object.toString().getBytes(),messageProperties);
    }

    //消息类型转换器中fromMessage方法返回的类型就是消费端处理器接收的类型
    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        System.out.println("=======fromMessage=========");
        return new String(message.getBody());
    }
}

    简介Jackson2JsonMessageConverter消息转换器：

    使用Jackson2JsonMessageConverter后，反序列化时要求发送的类和接受的类完全一样（字段，类名，包路径）。【也就是消息的生产的消息类型和消息的消费方法的消息参数类型一致】

    3. 生产者消费者使用案例

    3.1 相关注解

    3.2 消息的发送处理

    package com.cli.springboot_rabbitmq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.UUID;

/**
 * @Author xiongmin
 * @Description
 * @Date 2021/2/26 10:35
 * @Version 1.0
 **/
@Slf4j
@Service
public class RabbitMQService {

//    @Resource(name = "rabbitTemplateMessaging")
//    private RabbitTemplate rabbitTemplate;

    /**
     * 换成这个RabbitTemplate 之后，所有的消费者都要手动Ack消息确认被消费
     */
    @Resource(name = "rabbitTemplateInit")
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RabbitMQConfig amqpConfig;


    // 发送消息的后置处理器，MessagePostProcessor类的postProcessMessage方法得到的Message就是将参数Object内容转换成Message对象
    // 没有指定具体的Exchange就会使用默认的Exchange
    public void messageDeliver(String routineKey, Object o) {
        rabbitTemplate.convertAndSend(routineKey, o, message -> {
            System.out.println("-------处理前message-------------");
            System.out.println(message);
            // 设置message的一些头部信息
            message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
            message.getMessageProperties().setCorrelationId(UUID.randomUUID().toString());
            return message;
        });
    }

    // 发送消息的后置处理器，MessagePostProcessor类的postProcessMessage方法得到的Message就是将参数Object内容转换成Message对象
    public void messageDeliver(String exchange, String routineKey, Object o) {
        rabbitTemplate.convertAndSend(exchange, routineKey, o, message -> {
            System.out.println("-------处理前message-------------");
            System.out.println(message);
            // 设置message的一些头部信息
            message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
            message.getMessageProperties().setCorrelationId(UUID.randomUUID().toString());
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            System.out.println("-------处理后message-------------");
            System.out.println(message);
            return message;
        },new CorrelationData(UUID.randomUUID().toString()));
    }
    
    // 发送消息的后置处理器，MessagePostProcessor类的postProcessMessage方法得到的Message就是将参数Object内容转换成Message对象
    public void messageDeliver(String exchange, String routineKey, User user) {
        rabbitTemplate.convertAndSend(exchange, routineKey, user, message -> {
            System.out.println("-------处理前message-------------");
            System.out.println(message);
            // 设置message的一些头部信息
            /**
             * 消息在消费时，先根据消息投中给的messageId找到对饮给的User, 在判断User的状态是否已经被消费过
             * 感觉这也是一种防止消息重复消费的方式，即使同一个消息投递多次，也你能防止消息重复消费
             */
            message.getMessageProperties().setMessageId(user.getName());
            message.getMessageProperties().setCorrelationId(UUID.randomUUID().toString());
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            System.out.println("-------处理后message-------------");
            System.out.println(message);
            return message;
        },new CorrelationData(UUID.randomUUID().toString()));
    }
    
    
    /**
     * 防止同一时间段有好多个同步配置的消息发送，避免多个重复消息
     * @param deviceId
     */
    public void deliveConfigSyncMsgBeforeCheck(String deviceId) {
        if (connect == null) {
            connect = rabbitTemplate.getConnectionFactory().createConnection();
            // connect = amqpConfig.connectionFactory().createConnection();
        }
        if (channel == null) {
            try {
                channel = connect.createChannel(false);
            } catch (Exception e) {
                connect = amqpConfig.connectionFactory().createConnection();
                channel = connect.createChannel(false);
            }
        }
        try {
            long messageCount = channel.messageCount("f5-config");
            // 如果f5-ltm-state为空那我就发一条消息，否者就不发，防止MQ消息堆积
            if (messageCount == 0) {
                messageDeliver("f5.config", deviceId);
            } else {
                boolean a = false;
                int i;
                for(i=0; i < messageCount; i++) {
                    channel.basicQos(1);
                    GetResponse response = channel.basicGet("f5-config", true);
                    if (response == null) {
                        continue;
                    }
                    AMQP.BasicProperties props = response.getProps();
                    byte[] body = response.getBody();
                    String message = new String(body);
                    channel.basicPublish("messaging","f5.config", props, message.getBytes("UTF-8"));
                    if (message.contains(deviceId)) {
                        a =  true;
                        break;
                    }
                }
                if (!a && i >= messageCount) {
                    messageDeliver("f5.config", deviceId);
                }
            }
        } catch (Exception e) {
            logger.info("Retrieve Message fail " + e.getMessage());
        }
    }
    
}

    发送消息的后置处理器，MessagePostProcessor类的postProcessMessage方法得到的Message就是将参数Object内容转换成Message对象

    rabbitTemplate.convertAndSend(exchange, routineKey, o, message -> {
            System.out.println("-------处理前message-------------");
            System.out.println(message);
            // 设置message的一些头部信息
            message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
            message.getMessageProperties().setCorrelationId(UUID.randomUUID().toString());
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            System.out.println("-------处理后message-------------");
            System.out.println(message);
            return message;
        },new CorrelationData(UUID.randomUUID().toString()));

rabbitTemplate的convertAndSend的相关源码：
    @Override
	public void convertAndSend(String exchange, String routingKey, final Object message,
			final MessagePostProcessor messagePostProcessor,
			@Nullable CorrelationData correlationData) throws AmqpException {
		Message messageToSend = convertMessageIfNecessary(message);
		messageToSend = messagePostProcessor.postProcessMessage(messageToSend, correlationData,
				nullSafeExchange(exchange), nullSafeRoutingKey(routingKey));
		send(exchange, routingKey, messageToSend, correlationData);
	}

    image-20210227213546475

    3.3 消息的消费处理

    @RabbitListener和@RabbitHandler搭配使用

    @RabbitListener可以标注在类上面，当使用在类上面的时候，需要配合@RabbitHandler注解一起使用，@RabbitListener标注在类上面表示当有收到消息的时候，就交给带有@RabbitHandler的方法处理，具体找哪个方法处理，需要跟进MessageConverter转换后的java对象。

    package com.cli.springboot_rabbitmq.consumer;

import com.cli.springboot_rabbitmq.model.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @Author xiongmin
 * @Description
 * @Date 2021/2/26 11:54
 * @Version 1.0
 **/
@Component
@Slf4j
@RabbitListener(queues = "topic.WOMAN") //监听的队列名称 topic.WOMAN
public class TopicConsumerListenerTwo {


//    @RabbitHandler
//    public void receive(@Payload String message, @Headers Map<String, Object> headers) {
//        try {
//            log.info("TopicReceiver消费者收到消息  : ");
//            log.info("message = " + message);
//
//            log.info("TopicReceiver消费者收到消息头部信息  : ");
//            if (null != headers && !headers.isEmpty()) {
//                headers.forEach((key, value) -> {
//                    log.info(key + ": " + value + "
");
//                });
//            } else {
//                log.info("headers is empty");
//            }
//        } catch (Exception e) {
//            log.error(e.getMessage(), e);
//        }
//    }

    /**
     * 如果生产者生产的消息类型为String,那么就会执行该方法处理消息
     * @param message
     */
    @RabbitHandler
    public void receive(@Payload String message) {
        try {
            log.info("TopicReceiver消费者收到消息  : ");
            log.info("message = " + message);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


//    @RabbitHandler
//    public void receive(@Payload User user, @Headers Map<String, Object> headers) {
//        try {
//            log.info("TopicReceiver消费者收到消息  : ");
//            log.info("user = " + user);
//
//            log.info("TopicReceiver消费者收到消息头部信息  : ");
//            if (null != headers && !headers.isEmpty()) {
//                headers.forEach((key, value) -> {
//                    log.info(key + ": " + value + "
");
//                });
//            } else {
//                log.info("headers is empty");
//            }
//        } catch (Exception e) {
//            log.error(e.getMessage(), e);
//        }
//    }

    /**
     * 如果生产者生产的消息类型为User,那么就会执行该方法处理消息
     * @param user
     */
    @RabbitHandler
    public void receive(@Payload User user) {
        try {
            log.info("TopicReceiver消费者收到消息  : ");
            log.info("user = " + user);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 如果生产者生产的消息类型为Map,那么就会执行该方法处理消息
     * @param message
     */
    @RabbitHandler
    public void receive(@Payload Map message) {
        log.info("TopicReceiver {topic.WOMAN}消费者收到消息  : ");
        if (null != message && !message.isEmpty()) {
            message.forEach((key, value) -> {
                log.info(key + ": " + value + "
");
            });
        } else {
            log.info("message is empty");
        }
    }


    /**
     * @DO 总结
     * @RabbitListener标注在类上面表示当有收到消息的时候，就交给带有@RabbitHandler的方法处理，具体找哪个方法处理，需要更具MessageConverter转换后的java对象。
     * 注意，如果需要消息的头部信息，由于头部信息是一个MAP数据结构，那么Payload的数据类型不能为MAP类型，否者会报错，且即使其他不是Map类型的Payload，要获取消息的头部信息也会报错
     * 因为消费者消息Payload是MAP的类型的消息时，会查看那个方法中有MAP, 当有多个方法中具有MAP参数时，此时程序也不知道该使用哪个方法来处理这个消息，就会抛出异常
     */

}

    @RabbitListener标注在类上面表示当有收到消息的时候，就交给带有@RabbitHandler的方法处理，具体找哪个方法处理，需要更具MessageConverter转换后的java对象。
    注意: 如果需要消息的头部信息，由于头部信息是一个MAP数据结构，那么Payload的数据类型不能为MAP类型，否者会报错，且即使其他不是Map类型的Payload，要获取消息的头部信息也会报错;
    因为消费者消息Payload是MAP的类型的消息时，会查看那个方法中有MAP, 当有多个方法中具有MAP参数时，此时程序也不知道该使用哪个方法来处理这个消息，就会抛出异常

    4. 消息确认机制

    yaml配置打开confirmCallback 和returnCallback

    server:
  port: 11000

spring:
  application:
    name: rabbitmq-test

  rabbitmq:
#    单机IP配置
#    host: rabbitmq-host
#    port: 5672

#    集群IP配置
    addresses: rabbitmq01-host:5672,rabbitmq02-host:5672

#    用户名和密码，默认都是guest
    username: xiongmin
    password: xiongmin

#    交换器名可以不设置默认 【"" --> /】 交换器
    virtual-host: /rabbitmq_test
    publisher-returns: true # 发送者开启 return 确认机制
    publisher-confirm-type: correlated # 发送者开启 confirm 确认机制 等价于 spring.rabbitmq.publisher-returns=true
    #连接超时时间
    connection-timeout: 15000
    # 使用return-callback时必须设置mandatory为true
    template:
      mandatory: true
    default-exchange: default_exchange
    # 消费端配置
    listener:
      simple:
        retry:
          enabled: true # 支持重试
        #消费端
        concurrency: 5
        #最大消费端数
        max-concurrency: 10
        #自动签收auto  手动 manual
        acknowledge-mode: manual # 设置消费端手动 ack
        #限流（海量数据，同时只能过来一条）
        prefetch: 1

    分别实现confirmCallback和returnCallback回调的类接口

    ConfirmCallbackService.java

    package com.cli.springboot_rabbitmq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

/**
 * @Author xiongmin
 * @Description 监听消息是否发送交换机回调  只有投递失败的时候才会执行
 * @Date 2021/2/27 11:16
 * @Version 1.0
 **/
@Component
@Slf4j
public class ConfirmCallbackService implements RabbitTemplate.ConfirmCallback {

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            log.error("消息发送异常!");
        } else {
            log.info("发送者爸爸已经收到确认，correlationData={} ,ack={}, cause={}", correlationData.getId(), ack, cause);
        }
    }
}

    ReturnCallbackService.java

    package com.cli.springboot_rabbitmq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

/**
 * @Author xiongmin
 * @Description 消息为路由到队列监听类 只有投递失败的时候才会执行
 * @Date 2021/2/27 11:20
 * @Version 1.0
 * 如果消息未能投递到目标 queue 里将触发回调 returnCallback ，一旦向 queue 投递消息未成功，这里一般会记录下当前消息的详细投递数据，方便后续做重发或者补偿等操作。
 **/
@Component
@Slf4j
public class ReturnCallbackService implements RabbitTemplate.ReturnsCallback {

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.error("Fail... message:{},从交换机exchange:{},以路由键routingKey:{}," +
                        "未找到匹配队列，replyCode:{},replyText:{}",
                returned.getMessage(), returned.getExchange(), returned.getRoutingKey(), returned.getReplyCode(), returned.getReplyText());
    }
}

    设置RabbitTemplate

        /**
     * @param connectionFactory connectionFactory属性信息会直接是用yaml配置文件中配置的
     * @return
     */
    @Bean(value = "rabbitTemplateInit")
    public RabbitTemplate rabbitTemplateInit(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        // 消息类型转换器 -- 数据转换为json存入消息队列
        rabbitTemplate.setMessageConverter(jackson2MessageConverter());
        // 设置默认的交换机，如果发送的消息没有指定交换机，则使用默认的交换机
        rabbitTemplate.setExchange(defaultExchange);
        /**
         * mandatory：交换器无法根据自身类型和路由键找到一个符合条件的队列时的处理方式
         * true：RabbitMQ会调用Basic.Return命令将消息返回给生产者
         * false：RabbitMQ会把消息直接丢弃
         */
        rabbitTemplate.setMandatory(true);
        /**
         * 消费者确认收到消息后，手动ack回执回调处理
         */
        rabbitTemplate.setConfirmCallback(confirmCallbackService);
        /**
         * 消息投递到队列失败回调处理
         */
        rabbitTemplate.setReturnsCallback(returnCallbackService);


        // 服务端响应发送到的队列 reply-address 格式: exchange/routingKey
        rabbitTemplate.setReplyAddress("messaging/messaging-response");
        // 设置回复和接受消息的时间，单位为毫秒
        rabbitTemplate.setReplyTimeout(20000);
        rabbitTemplate.setReceiveTimeout(20000);
        return rabbitTemplate;
    }

    消费者消费消息

    package com.cli.springboot_rabbitmq.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;

/**
 * @Author xiongmin
 * @Description
 * @Date 2021/2/26 10:51
 * @Version 1.0
 **/
@Component
@Slf4j
@RabbitListener(queues = "TestDirectQueue") //监听的队列名称 TestDirectQueue，监听多个队列需要用单号分隔
public class DirectConsumerListener {

    @RabbitHandler
    public void receive(@Payload HashMap msg, Channel channel, Message message) throws IOException {
        try {
            log.info("DirectReceiver消费者收到消息  : ");
            if (null != msg && !msg.isEmpty()) {
                msg.forEach((key, value) -> {
                    log.info(key + ": " + value + "
");
                });
            } else {
                log.info("msg is empty");
            }


            // 消费者手动ACK确认消息被消费, 生产者会执行ConfirmCallback回调
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            try {
                if (message.getMessageProperties().getRedelivered()) {
                    log.error("消息已重复处理失败,拒绝再次接收...");
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
                } else  {
                    log.error("消息即将返回队列再次处理");
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
                }
            } catch (Exception ex) {
                log.error("消息消费异常时处理失败",ex.getMessage(),ex);
            }

        }

    }
}

    消费者回执方法

    消费消息有三种回执方法，我们来分析一下每种方法的含义。

    1、basicAck

    basicAck：表示成功确认，使用此回执方法后，消息会被rabbitmq broker 删除。

    void basicAck(long deliveryTag, boolean multiple) 
复制代码

    deliveryTag：表示消息投递序号，每次消费消息或者消息重新投递后，deliveryTag都会增加。手动消息确认模式下，我们可以对指定deliveryTag的消息进行ack、nack、reject等操作。

    multiple：是否批量确认，值为 true 则会一次性 ack所有小于当前消息 deliveryTag 的消息。

    举个栗子： 假设我先发送三条消息deliveryTag分别是5、6、7，可它们都没有被确认，当我发第四条消息此时deliveryTag为8，multiple设置为 true，会将5、6、7、8的消息全部进行确认。

    2、basicNack

    basicNack ：表示失败确认，一般在消费消息业务异常时用到此方法，可以将消息重新投递入队列。

    void basicNack(long deliveryTag, boolean multiple, boolean requeue)
复制代码

    deliveryTag：表示消息投递序号。

    multiple：是否批量确认。

    requeue：值为 true 消息将重新入队列。

    3、basicReject

    basicReject：拒绝消息，与basicNack区别在于不能进行批量操作，其他用法很相似。

    void basicReject(long deliveryTag, boolean requeue)
复制代码

    deliveryTag：表示消息投递序号。

    requeue：值为 true 消息将重新入队列。

    image-20210227211707050

    ConfirmCallBack和returnBackCall回调执行时机如下：

    先从总体的情况分析，推送消息存在四种情况：

    ①消息推送到server，但是在server里找不到交换机
    ②消息推送到server，找到交换机了，但是没找到队列
    ③消息推送到sever，交换机和队列啥都没找到
    ④消息推送成功

    那么我先写几个接口来分别测试和认证下以上4种情况，消息确认触发回调函数的情况：

    ①消息推送到server，但是在server里找不到交换机

    结论： ①这种情况触发的是 ConfirmCallback 回调函数。

    ②消息推送到server，找到交换机了，但是没找到队列

    结论：②这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。

    ③消息推送到sever，交换机和队列啥都没找到
    这种情况其实一看就觉得跟①很像，没错 ，③和①情况回调是一致的，所以不做结果说明了。
    结论： ③这种情况触发的是 ConfirmCallback 回调函数。

    ④消息推送成功

    结论： ④这种情况触发的是 ConfirmCallback 回调函数。

    相关链接
  • 相关阅读:
    MySQL MHA高可用方案
    微服务架构的基础框架选择：Spring Cloud还是Dubbo？
    elasticsearch
    spring Cloud构建微服务架构
    SpringBoot-Learning
    JetBrains激活
    正则表达式(java)规则大全
    Android调用Webservice
    UCML针对数据表的修改自定义维护数据变更记录
    SQL游标模板
  • 原文地址：https://www.cnblogs.com/dream-it-possible/p/14457326.html
Copyright © 2011-2022 走看看