为什么要在5.x弃用consumeThreadMax

[Liyuhao02]

由于consumeThreadMin和consumeThreadMax在5.x版本当中被弃用，我有几个问题

1. 由于线程数量无法再设置核心线程数以及最大线程数，线程数量只能设置为固定线程，假设我设置为固定20，如果我在多Topic的情况下，一旦项目启动并创建了消费者，那么创建的消费者线程数量就将会是 n(topic数量) * 20, 这样就会有大量的线程进入Waiting状态并浪费资源
2. 消费者线程数调低会导致业务高峰期消费不及时
3. 如果按照上述逻辑，并且合并Topic也会有问题，如果同一Topic内，多个业务逻辑比如A、B、C，其中A会产生大量消息，B、C很少，如果A产生了大量消息并导致堆积会造成B、C也造成堆积

所以结合上述的问题，为什么不设计成允许动态线程池，低峰期自动降低线程数，高峰期自动开辟线程到最大线程数，队列满之后的线程认为来不及处理不处理呢，虚心请教一下

[freedom0123]

下面是我个人看到的一些，不对的话，请指正：

@Component
@RocketMQMessageListener(topic = "coding-filter-message", consumerGroup = "coding")
@Slf4j
public class CodingFilterMessageRocketMqListener implements RocketMQListener, RocketMQPushConsumerLifecycleListener {

    @Override
    public ConsumeResult consume(MessageView messageView) {
        ByteBuffer body = messageView.getBody();
        String message = StandardCharsets.UTF_8.decode(body).toString();
        log.info("consume message: {}", message);
        return ConsumeResult.SUCCESS;
    }
    @Override
    public void prepareStart(PushConsumerBuilder consumerBuilder) {
        consumerBuilder.setConsumptionThreadCount(16);
    }
}

对于每个 Topic 消费者，可以自行设置线程数，这也是扩展点吧，当然这个线程数是核心线程数，也是最大线程数.。下面是构建的线程池，构建的阻塞队列是无界的，也就是你这里配置的最大线程数实际上也不会生效

this.consumptionExecutor = new ThreadPoolExecutor(
            consumptionThreadCount,
            consumptionThreadCount,
            60,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(),
            new ThreadFactoryImpl("MessageConsumption", this.getClientId().getIndex()));

[freedom0123]

我用的版本是：

 <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-v5-client-spring-boot-starter</artifactId>
            <version>2.3.4</version>
        </dependency>

[Liyuhao02]

这个我上次也看到了，但是有个问题在于如果我们业务Topic设计不合理或者确实需要很多Topic的情况下，可能会在启动服务时自动创建大量线程，举例我们有40个Topic，每个Topic对应的Consume配置线程组线程数为20，在服务启动时就会创建 40 * 20 =800 个消费者线程在等待消息，产生大量资源浪费，在业务高峰期也无法进行扩容，所以我想的是如果支持配置最大、最小线程数，就可以在业务低峰期保留较低的消费者线程数量，业务高峰期自动创建线程数量，例如consumeThreadMin = 5, consumeThreadMax = 64，这样既不会浪费资源，也不会造成高峰期消费者线程数量不足

[chathoward712-cpu]

RocketMQ 5.x 的设计理念是 简单、可预测、可靠，而不是依赖复杂的动态线程池。
更倾向水平扩展 Broker + ConsumerGroup 水平扩展 → 高峰处理能力。