基于Spring Boot+Java实现Kafka Topic高优先级监听的方案咨询

方案一：通过线程优先级实现相对优先级 #

这种方式利用操作系统的线程调度机制，给高优先级Topic的消费线程设置更高优先级，让其优先获得CPU资源处理消息，适合不需要严格绝对优先级的场景。

实现步骤： #

1. 定义两个不同的容器工厂，分别配置线程优先级：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, MyRequest> highPriorityContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, MyRequest> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(1);
    // 高优先级线程设为系统最高优先级
    factory.getContainerProperties().setConsumerThreadPriority(Thread.MAX_PRIORITY);
    return factory;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, MyRequest> lowPriorityContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, MyRequest> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(1);
    // 低优先级线程设为系统最低优先级
    factory.getContainerProperties().setConsumerThreadPriority(Thread.MIN_PRIORITY);
    return factory;
}

// 共用的消费者基础配置
@Bean
public ConsumerFactory<String, MyRequest> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka集群地址");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "你的消费组ID");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    // 补充其他必要配置（如自动提交策略、重置偏移量规则等）
    return new DefaultKafkaConsumerFactory<>(props);
}

2. 拆分监听方法，分别绑定对应的容器工厂：

@KafkaListener(topics = "topic-p", containerFactory = "highPriorityContainerFactory")
public void handleHighPriorityMsg(@Payload MyRequest myRequest, @Headers HeadersAccessor headers) {
    processMsg(myRequest);
}

@KafkaListener(topics = "topic-np", containerFactory = "lowPriorityContainerFactory")
public void handleLowPriorityMsg(@Payload MyRequest myRequest, @Headers HeadersAccessor headers) {
    processMsg(myRequest);
}

// 统一的消息处理逻辑
private void processMsg(MyRequest request) {
    // 业务代码实现
}

注意：线程优先级依赖操作系统调度逻辑，是相对优先级，无法做到绝对的「处理完所有高优先级消息再处理低优先级」，但多数业务场景下足以满足需求。

方案二：通过优先级队列实现严格优先级 #

如果需要严格保证高优先级消息全部处理完成后再处理低优先级消息，可以采用「消费消息入优先级队列+工作线程按优先级取数」的方式。

实现步骤： #

1. 定义带优先级的消息包装类：

public class PrioritizedMsg implements Comparable<PrioritizedMsg> {
    // 数值越小，优先级越高
    private static final int HIGH_PRIORITY = 1;
    private static final int LOW_PRIORITY = 2;

    private MyRequest payload;
    private int priority;

    public PrioritizedMsg(MyRequest payload, String topic) {
        this.payload = payload;
        this.priority = "topic-p".equals(topic) ? HIGH_PRIORITY : LOW_PRIORITY;
    }

    @Override
    public int compareTo(PrioritizedMsg other) {
        return Integer.compare(this.priority, other.priority);
    }

    // getter方法
    public MyRequest getPayload() {
        return payload;
    }
}

2. 配置优先级队列和工作线程：

@Configuration
public class PriorityQueueConfig {

    @Bean
    public PriorityBlockingQueue<PrioritizedMsg> priorityMsgQueue() {
        return new PriorityBlockingQueue<>();
    }

    @Bean
    public MsgProcessor msgProcessor(PriorityBlockingQueue<PrioritizedMsg> queue) {
        MsgProcessor processor = new MsgProcessor(queue);
        // 启动工作线程，可根据业务需求调整线程数量
        new Thread(processor).start();
        return processor;
    }

    public static class MsgProcessor implements Runnable {
        private final PriorityBlockingQueue<PrioritizedMsg> queue;

        public MsgProcessor(PriorityBlockingQueue<PrioritizedMsg> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    PrioritizedMsg msg = queue.take();
                    // 处理消息
                    processMsg(msg.getPayload());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }

        private void processMsg(MyRequest payload) {
            // 业务代码实现
        }
    }
}

3. 修改监听方法，将消息放入优先级队列：

@Autowired
private PriorityBlockingQueue<PrioritizedMsg> priorityMsgQueue;

@KafkaListener(topics = {"topic-p", "topic-np"})
public void listenMsg(@Payload MyRequest myRequest, @Headers HeadersAccessor headers) {
    String topic = headers.getHeader(KafkaHeaders.RECEIVED_TOPIC).toString();
    priorityMsgQueue.put(new PrioritizedMsg(myRequest, topic));
}

注意：如果高优先级消息持续大量涌入，低优先级消息可能会被长期阻塞，可根据业务需求添加超时降级或流量控制逻辑。

原代码问题说明 #

你之前在同一个方法上添加两个@KafkaListener注解，会创建两个独立的消费者容器并行消费各自的Topic，因此无法实现优先级控制，需要按上述方案调整。

内容的提问来源于stack exchange，提问作者gayathri s