springboot整合rabbitmq

古城微笑少年丶 2022-01-30 05:55 322阅读 0赞

# **RabbitMQ概念** #

## **ConnectionFactory、Connection、Channel** ##

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。ConnectionFactory为Connection的制造工厂。  
Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

## **Queue** ##

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。  
![这里写图片描述][SouthEast]  
RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。  
![这里写图片描述][SouthEast 1]  
多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。  
![这里写图片描述][SouthEast 2]

## **Message acknowledgment** ##

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。

这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

--------------------

## **Message durability** ##

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

--------------------

## **Prefetch count** ##

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。  
![这里写图片描述][SouthEast 3]

--------------------

## **Exchange** ##

在上一节我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。  
![这里写图片描述][SouthEast 4]  
Exchange是按照什么逻辑将消息路由到Queue的？这个将在Binding一节介绍。  
RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在Exchange Types一节介绍。

--------------------

## **routing key** ##

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。  
在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。  
RabbitMQ为routing key设定的长度限制为255 bytes。

--------------------

## **Binding** ##

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。  
![这里写图片描述][SouthEast 5]

--------------------

## **Binding key** ##

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。  
在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。  
binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

--------------------

## **Exchange Types** ##

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

**fanout**

fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。  
![这里写图片描述][SouthEast 6]  
上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

**direct**

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。  
![这里写图片描述][SouthEast 7]  
以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

**topic**

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

*  routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
 *  binding key与routing key一样也是句点号“. ”分隔的字符串
 *  binding key中可以存在两种特殊字符“\*”与“\#”，用于做模糊匹配，其中“\*”用于匹配一个单词，“\#”用于匹配多个单词（可以是零个）  
    ![这里写图片描述][SouthEast 8]

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

**headers**

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。  
在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。  
该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

## **RPC** ##

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。  
但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。  
![这里写图片描述][SouthEast 9]

RabbitMQ中实现RPC的机制是：

*  客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14种properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
 *  服务器端收到消息并处理
 *  服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
 *  客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

下面附上项目中的相关代码：

rabbitmq相关配置（生产者和消费者用的是不同的rabbitmq配置）：

@Configuration
    public class RabbitMqConfig
    {
        @Bean(name = "consumeDataConnectionFactory")
        public ConnectionFactory consumeDataConnectionFactory(
            @Value("${spring.rabbitmq.consumeData.host}") String host,
            @Value("${spring.rabbitmq.consumeData.port}") int port,
            @Value("${spring.rabbitmq.consumeData.username}") String username,
            @Value("${spring.rabbitmq.consumeData.password}") String password,
            @Value("${spring.rabbitmq.consumeData.virtualHost}") String virtualHost)
        {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(host);
            connectionFactory.setPort(port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost(virtualHost);
    
            return connectionFactory;
        }
    
        @Bean(name = "sendMailConnectionFactory")
        public ConnectionFactory sendMailConnectionFactory(
            @Value("${spring.rabbitmq.sendMail.host}") String host,
            @Value("${spring.rabbitmq.sendMail.port}") int port,
            @Value("${spring.rabbitmq.sendMail.username}") String username,
            @Value("${spring.rabbitmq.sendMail.password}") String password,
            @Value("${spring.rabbitmq.sendMail.virtualHost}") String virtualHost)
        {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(host);
            connectionFactory.setPort(port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost(virtualHost);
    
            return connectionFactory;
        }
    }

## 消费者 ##

项目中用到的是主动拉取rabbitmq中数据的方式，并且是批量拉取，当队列中有数据的时候，就将所有数据消费，消费完之后手动提交，这边的exchange和queue是事先已经在rabbitmq管理页创建好了的，故代码里直接绑定exchange、topic和routingkey

@Component("monitorQueueConfigConsumeMqTask")
    public class MonitorQueueConfigConsumeMqTask
    {
        /**
         * 日志
         */
        private static final Logger LOGGER = LoggerFactory.getLogger(
            MonitorQueueConfigConsumeMqTask.class);
    
        @Autowired
        private ConnectionFactory connectionFactory;
    
        @Autowired
        private IMonitorQueueConfigService monitorQueueConfigService;
    
        /**
         * <从rabbitmq中消费机场数据>
         *
         * @param monitorQueueConfigId monitorQueueConfigId
         * @throws
         */
        @DataSource(value = DataSourceType.datamonitor)
        public void monitorQueueConfigConsumeMqTask(String monitorQueueConfigId)
        {
            LOGGER.info(
                "[MonitorQueueConfigConsumeMqTask] start to monitorQueueConfigConsumeMqTask. monitorQueueConfigId is {}.",
                monitorQueueConfigId);
            //根据监控队列配置id获取监控队列相关信息
            MonitorQueueConfig monitorQueueConfig = new MonitorQueueConfig();
            monitorQueueConfig.setId(monitorQueueConfigId);
            List<MonitorQueueConfig> monitorQueueConfigList = monitorQueueConfigService.queryMonitorQueueConfigList(
                monitorQueueConfig);
            if (CollectionUtils.isEmpty(monitorQueueConfigList))
            {
                LOGGER.error("[MonitorQueueConfigConsumeMqTask] monitorQueueConfigId is invalid.");
                return;
            }
    
            monitorQueueConfig = monitorQueueConfigList.get(0);
            String queueName = monitorQueueConfig.getQueue();
            String exchange = monitorQueueConfig.getExchange();
            Connection connection = null;
            Channel channel = null;
            long tag = 0;
    
            try
            {
                connection = connectionFactory.newConnection();
                channel = connection.createChannel();
    
                //声明一个exchange
    //            channel.exchangeDeclare(monitorQueueConfig.getExchange(), "topic");
                //声明一个随机队列
    //            String queueName = channel.queueDeclare().getQueue();
                //直接绑定现有的exchange、topic及routingkey（#表名匹配所有）
                channel.queueBind(queueName, exchange, "#");
    
                System.out.println("[MonitorQueueConfigConsumeMqTask] Waiting for messages.");
    
                //采用主动拉取数据的方式消费数据（暂定一次消费mq中所有的数据）
    //            List<GetResponse> responseList = Lists.newArrayListWithCapacity(1000);
                List<GetResponse> responseList = new ArrayList<>();
                List<Date> dataList = new ArrayList<>();
                GetResponse getResponse = null;
                while (true)
                {
                    getResponse = channel.basicGet(queueName, false);
                    if (null == getResponse)
                    {
                        LOGGER.info("[MonitorQueueConfigConsumeMqTask] consume of this time is end.");
                        break;
                    }
                    responseList.add(getResponse);
                    dataList.add(new Date());
                    tag = getResponse.getEnvelope().getDeliveryTag();
                }
                LOGGER.info("Get <{}> responses this batch", responseList.size());
                if (!CollectionUtils.isEmpty(responseList))
                {
                    for (GetResponse getResponseTemp : responseList)
                    {
                        String message = HandleStringUtil.unicodeStr2String(
                            new String(getResponseTemp.getBody(), "utf-8"));
                        LOGGER.info("[MonitorQueueConfigConsumeMqTask] message is {}.", message);
    
                        getResponseTemp.getProps().getTimestamp();
                        getResponseTemp.getMessageCount();
                    }
                    //commit
                    channel.basicAck(tag, true);
                }
    
    //            GetResponse response = channel.basicGet(monitorQueueConfig.getQueue(), false);
    //            String messageStr = new String(response.getBody());
    //            LOGGER.info("[MonitorQueueConsumer] message is {}.",
    //                messageStr.substring(2, messageStr.length() - 2));
    //            channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
    
                //创建队列消费者
    //            Consumer consumer = new MonitorQueueConsumer(channel);
    //            channel.basicConsume(queueName, true, consumer);
    //            channel.basicConsume(monitorQueueConfig.getQueue(), true, consumer);
    //            channel.basicQos(100);
    //            channel.basicConsume(monitorQueueConfig.getQueue(), true, consumer);
    
                LOGGER.info(
                    "[MonitorQueueConfigConsumeMqTask] End to monitorQueueConfigConsumeMqTask.");
            }
            catch (Exception e)
            {
                LOGGER.error(
                    "[MonitorQueueConfigConsumeMqTask] consume data from mq catch exception. {}",
                    e.getMessage());
                LOGGER.error("", e);
    
                try
                {
                    //ack返回false，并重新回到队列，api里面解释得很清楚
                    channel.basicNack(tag, true, true);
                }
                catch (IOException e1)
                {
                    LOGGER.error(
                        "[MonitorQueueConfigConsumeMqTask] basicNack catch IOException. {}",
                        e.getMessage());
                    LOGGER.error("", e);
                }
                return;
            }
            finally
            {
                try
                {
                    channel.close();
                    connection.close();
                    LOGGER.info(
                        "[MonitorQueueConfigConsumeMqTask finally] End to monitorQueueConfigConsumeMqTask.");
                }
                catch (IOException e)
                {
                    LOGGER.error(
                        "[MonitorQueueConfigConsumeMqTask finally] channel or connect catch IOException. {}",
                        e.getMessage());
                    LOGGER.error("", e);
                    return;
                }
                catch (TimeoutException e)
                {
                    LOGGER.error(
                        "[MonitorQueueConfigConsumeMqTask finally] channel or connect catch TimeoutException. {}",
                        e.getMessage());
                    LOGGER.error("", e);
                    return;
                }
            }
        }
    }

注释部分为消费的另外一种方式：定义消费者，由队列推送消息：

public class MonitorQueueConsumer extends DefaultConsumer
    {
        /**
         * 日志
         */
        private static final Logger LOGGER = LoggerFactory.getLogger(MonitorQueueConsumer.class);
    
        /**
         * <构造函数>
         *
         * @param channel channel
         */
        public MonitorQueueConsumer(Channel channel)
        {
            super(channel);
        }
    
        /**
         * <消费到数据后进行后续的操作>
         *
         * @param consumerTag     consumerTag
         * @param envelope        envelope
         * @param basicProperties basicProperties
         * @param message         message
         * @throws
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                                   AMQP.BasicProperties basicProperties, byte[] message)
            throws IOException
        {
            String messageStr = new String(message, "UTF-8");
            LOGGER.info("[MonitorQueueConsumer] message is {}.",
                HandleStringUtil.unicodeStr2String(messageStr));
        }
    
    }

生产者：

@Component
    @Order(2)
    public class MailSendTask implements ApplicationRunner
    {
        /**
         * 日志
         */
        private static final Logger LOGGER = LoggerFactory.getLogger(MailSendTask.class);
    
        /**
         * 定义定时任务线程
         */
        private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(
            1);
    
        @Autowired
        private IAlarmMailSendLogService alarmMailSendLogService;
    
        @Autowired
        @Qualifier("sendMailConnectionFactory")
        private ConnectionFactory connectionFactory;
    
        @Override
        public void run(ApplicationArguments applicationArguments)
            throws Exception
        {
            LOGGER.info("[MailSendTask] start to MailSendTask.");
    
            // 参数：1、任务体 2、首次执行的延时时间
            //      3、任务执行间隔 4、间隔时间单位
            scheduledExecutorService.scheduleAtFixedRate(new Runnable()
            {
                @Override
                public void run()
                {
                    sendMails();
                }
            }, 1, 30, TimeUnit.MINUTES);
    
            LOGGER.info("[MailSendTask] end to MailSendTask.");
        }
    
        /**
         * <发送预警邮件s>
         *
         * @throws
         */
        private void sendMails()
        {
            LOGGER.info("[MailSendTask] start to sendMails");
            //查询所有未发送邮件
            AlarmMailSendLog alarmMailSendLog = new AlarmMailSendLog();
            alarmMailSendLog.setIsSend("N");
            alarmMailSendLog.setIsDel("0");
            List<AlarmMailSendLog> alarmMailSendLogList = alarmMailSendLogService.queryAlarmMailSendLogList(
                alarmMailSendLog);
    
            if (!CollectionUtils.isEmpty(alarmMailSendLogList))
            {
                for (AlarmMailSendLog alarmMailSendLogTemp : alarmMailSendLogList)
                {
                    //调用邮件发送接口发送邮件
                    sendMail(alarmMailSendLogTemp);
    
                    //更新邮件发送日志中状态字段
                    Date currentTime = new Date();
                    alarmMailSendLogTemp.setIsSend("Y");
                    alarmMailSendLogTemp.setSendTime(currentTime);
                    alarmMailSendLogTemp.setModifyTime(currentTime);
    
                    alarmMailSendLogService.updateAlarmMailSendLog(alarmMailSendLogTemp);
                }
            }
    
            LOGGER.info("[MailSendTask] end to sendMails");
        }
    
        /**
         * <发送预警邮件>
         *
         * @param alarmMailSendLog alarmMailSendLog
         * @throws
         */
        private void sendMail(AlarmMailSendLog alarmMailSendLog)
        {
            LOGGER.info("[MailSendTask] start to sendMail");
    
            MailInfoEntity mailInfoEntity = new MailInfoEntity();
            mailInfoEntity.setSubject(alarmMailSendLog.getMailTitle());
            mailInfoEntity.setToUser(Arrays.asList(alarmMailSendLog.getMailAddress().split(",")));
            mailInfoEntity.setContent(alarmMailSendLog.getMailContent());
    
            // 创建一个连接
            Connection connection = null;
            Channel channel = null;
            try
            {
                connection = connectionFactory.newConnection();
                channel = connection.createChannel();
    
                channel.basicPublish("EXCHANGE_AIRPORT_MAIL_INC_V1", "SJZXMONITOR.ALARM.MAIL",
                    new AMQP.BasicProperties(), JSON.toJSONString(mailInfoEntity).getBytes());
            }
            catch (Exception e)
            {
                LOGGER.error(
                    "[MailSendTask] product data to mq catch Exception. {}",
                    e.getMessage());
                LOGGER.error("", e);
                return;
            }
            finally
            {
                try
                {
                    channel.close();
                    connection.close();
                    LOGGER.info(
                        "[MailSendTask finally] End to sendMail.");
                }
                catch (Exception e)
                {
                    LOGGER.error(
                        "[MailSendTask finally] channel or connect catch Exception. {}",
                        e.getMessage());
                    LOGGER.error("", e);
                    return;
                }
            }
        }
    }

附上两个注解：

`@Primary` 优先考虑，优先考虑被注解的对象注入

`@Qualifier` 名字声明，声明后对名字进行使用

[SouthEast]: /images/20220130/4701189ec4364382bc7d58ef8aa14e87.png
[SouthEast 1]: /images/20220130/27a2154e30af45a28effc539e5694ed4.png
[SouthEast 2]: /images/20220130/4d86b9cff93347a5bc8039656e451f83.png
[SouthEast 3]: /images/20220130/6bc24c84e4af4434aeabec5c2e6b6218.png
[SouthEast 4]: /images/20220130/9cbb51a7ad9b40d3b77073567071cd7d.png
[SouthEast 5]: /images/20220130/a49b81b087ef4f5ca26f9f4a0c8c1067.png
[SouthEast 6]: /images/20220130/68d6742b368a4d07be02631a45e4c2aa.png
[SouthEast 7]: /images/20220130/8877452f2f32425fa5c284cf44d515be.png
[SouthEast 8]: /images/20220130/e1b6372d95d3460daef4032af32d80ca.png
[SouthEast 9]: /images/20220130/d82f35be038e419288cefcd577eb384c.png