MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

消息队列的消息优先级处理

2022-11-132.4k 阅读

消息队列与消息优先级概述

在后端开发的消息队列场景中，消息队列是一种用于在不同组件或服务之间异步传递消息的机制。它解耦了消息的发送者和接收者，使得系统可以更灵活、可靠地运行。常见的消息队列系统有 RabbitMQ、Kafka、RocketMQ 等。

消息优先级则是在消息队列的基础上，为不同的消息赋予不同的优先级别。高优先级的消息应该比低优先级的消息更早地被处理。这种机制在很多场景下都非常有用。例如，在电商系统中，订单支付消息可能比商品评论消息具有更高的优先级，因为支付消息的及时处理对于保证交易的顺利进行至关重要。

消息队列实现消息优先级的方式

不同的消息队列系统实现消息优先级的方式有所不同。

RabbitMQ 的消息优先级实现

RabbitMQ 从 3.5.0 版本开始支持消息优先级。要使用消息优先级，首先需要在声明队列时设置x-max-priority参数，该参数指定了队列支持的最大优先级数。例如：

import pika

# 连接 RabbitMQ 服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明队列并设置最大优先级为 10
channel.queue_declare(queue='priority_queue', arguments={'x-max-priority': 10})

在发送消息时，可以通过设置properties中的priority字段来指定消息的优先级：

# 发送高优先级消息
channel.basic_publish(exchange='',
                      routing_key='priority_queue',
                      body='High priority message',
                      properties=pika.BasicProperties(priority=10))

# 发送低优先级消息
channel.basic_publish(exchange='',
                      routing_key='priority_queue',
                      body='Low priority message',
                      properties=pika.BasicProperties(priority=1))

消费者在接收消息时，RabbitMQ 会按照优先级顺序将消息发送给消费者。

Kafka 的消息优先级实现

Kafka 本身并没有直接支持消息优先级的功能。但是，可以通过一些间接的方式来模拟消息优先级。一种常见的方法是使用多个主题（Topic），每个主题对应一个优先级。例如，创建三个主题：high_priority_topic、medium_priority_topic、low_priority_topic。 生产者根据消息的优先级将消息发送到不同的主题：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaPriorityProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 发送高优先级消息
        ProducerRecord<String, String> highPriorityRecord = new ProducerRecord<>("high_priority_topic", "High priority message");
        producer.send(highPriorityRecord);

        // 发送中优先级消息
        ProducerRecord<String, String> mediumPriorityRecord = new ProducerRecord<>("medium_priority_topic", "Medium priority message");
        producer.send(mediumPriorityRecord);

        // 发送低优先级消息
        ProducerRecord<String, String> lowPriorityRecord = new ProducerRecord<>("low_priority_topic", "Low priority message");
        producer.send(lowPriorityRecord);

        producer.close();
    }
}

消费者则可以根据需求，先消费高优先级主题的消息，再消费中优先级主题的消息，最后消费低优先级主题的消息：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaPriorityConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "priority_group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 先消费高优先级主题
        consumer.subscribe(Arrays.asList("high_priority_topic"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("High Priority: " + record.value());
            }
        }

        // 再消费中优先级主题
        consumer.subscribe(Arrays.asList("medium_priority_topic"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Medium Priority: " + record.value());
            }
        }

        // 最后消费低优先级主题
        consumer.subscribe(Arrays.asList("low_priority_topic"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Low Priority: " + record.value());
            }
        }
    }
}

RocketMQ 的消息优先级实现

RocketMQ 本身也不直接支持消息优先级。不过，可以通过在消息体中添加优先级字段，然后在消费者端进行排序来实现类似的效果。 生产者在发送消息时，在消息体中添加优先级字段：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class RocketMQPriorityProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("PriorityProducerGroup");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 高优先级消息
        Message highPriorityMessage = new Message("PriorityTopic", "High".getBytes());
        highPriorityMessage.putUserProperty("priority", "10");
        SendResult highSendResult = producer.send(highPriorityMessage);
        System.out.println("High Priority Send Result: " + highSendResult);

        // 低优先级消息
        Message lowPriorityMessage = new Message("PriorityTopic", "Low".getBytes());
        lowPriorityMessage.putUserProperty("priority", "1");
        SendResult lowSendResult = producer.send(lowPriorityMessage);
        System.out.println("Low Priority Send Result: " + lowSendResult);

        producer.shutdown();
    }
}

消费者在接收消息后，根据优先级字段进行排序处理：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;

public class RocketMQPriorityConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("PriorityConsumerGroup");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("PriorityTopic", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                Collections.sort(msgs, new Comparator<MessageExt>() {
                    @Override
                    public int compare(MessageExt o1, MessageExt o2) {
                        int priority1 = Integer.parseInt(o1.getUserProperty("priority"));
                        int priority2 = Integer.parseInt(o2.getUserProperty("priority"));
                        return priority2 - priority1;
                    }
                });

                for (MessageExt msg : msgs) {
                    System.out.println("Priority: " + msg.getUserProperty("priority") + ", Message: " + new String(msg.getBody()));
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Consumer Started.");
    }
}

消息优先级处理的应用场景

电商系统中的应用

在电商系统中，订单相关的消息优先级处理非常关键。比如，订单支付成功的消息需要尽快处理，以便更新库存、安排发货等后续操作。而用户评价消息的处理可以相对滞后。通过设置消息优先级，支付成功消息可以被赋予较高的优先级，优先进入处理流程，确保交易的及时完成。

金融系统中的应用

在金融系统中，交易相关的消息具有不同的优先级。例如，实时交易确认消息应该比账户余额查询消息具有更高的优先级。高优先级的交易确认消息可以及时通知用户交易结果，保证金融交易的准确性和及时性。

物流系统中的应用

在物流系统中，货物到达通知消息可能比物流轨迹更新消息具有更高的优先级。当货物到达目的地时，需要尽快通知相关人员进行处理，以提高物流效率。通过消息优先级机制，可以确保重要的货物到达消息优先被处理。

消息优先级处理的挑战与解决方案

性能问题

在消息队列中处理消息优先级可能会带来一定的性能开销。例如，RabbitMQ 在处理优先级队列时，需要额外的计算资源来维护优先级顺序。为了应对这个问题，可以在设计队列时合理设置最大优先级数，避免设置过高导致性能下降。同时，可以对队列进行适当的分区，将高优先级消息和低优先级消息分别存储在不同的分区，以提高处理效率。

消息堆积问题

如果高优先级消息持续不断地进入队列，可能会导致低优先级消息长时间堆积。为了解决这个问题，可以设置一定的策略，例如在高优先级消息处理一段时间后，暂停处理高优先级消息，优先处理一定数量的低优先级消息，以保证低优先级消息也能得到及时处理。

优先级判断的准确性

在实际应用中，准确判断消息的优先级是很重要的。如果优先级判断失误，可能会导致重要消息得不到及时处理。为了提高优先级判断的准确性，需要深入了解业务需求，建立合理的优先级判断规则。同时，可以通过人工审核或者机器学习等方式来优化优先级判断机制。

消息优先级处理的最佳实践

合理设置优先级

根据业务需求，合理设置消息的优先级。不要过度设置高优先级消息，避免低优先级消息被长时间忽略。可以通过对业务流程的分析，确定不同类型消息的优先级范围。

定期清理优先级队列

定期清理优先级队列中的过期消息或者处理失败的消息，避免这些消息占用过多的资源，影响正常消息的处理。

监控与优化

建立监控机制，实时监控消息队列的优先级处理情况。通过监控数据，可以及时发现性能问题、消息堆积等异常情况，并进行针对性的优化。例如，根据监控数据调整优先级队列的参数，优化消息处理策略等。

通过以上对消息队列消息优先级处理的深入分析和实践指导，希望能帮助后端开发人员更好地利用消息队列的优先级机制，提升系统的性能和可靠性。在实际应用中，需要根据具体的业务场景和消息队列系统的特点，灵活选择合适的实现方式和优化策略。