Laravel中使用RdKafka多消费者提升MongoDB插入性能的问题

核心前提：Kafka Topic分区数必须匹配消费者数量 #

你复制消费者实例没生效的核心原因大概率是Topic的分区数不足。Kafka的消费组机制会把Topic的分区均匀分配给同组内的消费者，若Topic只有1个分区，哪怕启动10个消费者，也只会有1个在处理消息，其余处于闲置状态。

操作步骤： #

1. 查看当前Topic的分区数：

   kafka-topics.sh --bootstrap-server broker1:9092 --describe --topic your_topic_name
   

2. 扩容分区数（假设要启动3个消费者，就设为3个分区，分区数只能增加不能减少）：

   kafka-topics.sh --bootstrap-server broker1:9092 --alter --topic your_topic_name --partitions 3
   

正确实现多消费者的两种方式 #

方式1：用Supervisor管理多个独立消费进程 #

Laravel单进程内复制消费者实例是串行执行，无法利用多核，必须启动多个独立进程。

1. 编写独立的消费命令

在Laravel中创建Artisan命令，比如kafka:consume：

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use RdKafka\Consumer;
use RdKafka\TopicConf;
use RdKafka\Conf;
use MongoDB\Client;

class KafkaConsume extends Command
{
    protected $signature = 'kafka:consume';
    protected $description = 'Consume Kafka messages and insert into MongoDB';

    public function handle()
    {
        // 配置Kafka消费者
        $conf = new Conf();
        $conf->set('group.id', 'mongodb_insert_group'); // 同一消费组
        $conf->set('metadata.broker.list', env('KAFKA_BROKERS'));
        $conf->set('auto.offset.reset', 'latest');

        $topicConf = new TopicConf();
        $topicConf->set('auto.commit.interval.ms', 1000);

        $consumer = new Consumer($conf);
        $topic = $consumer->newTopic(env('KAFKA_TOPIC'), $topicConf);
        $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

        // 初始化MongoDB客户端
        $mongoClient = new Client(env('MONGODB_URI'));
        $collection = $mongoClient->selectCollection(env('MONGODB_DB'), env('MONGODB_COLLECTION'));

        $batch = [];
        $batchSize = 1000; // 批量插入大小，根据实际调整

        while (true) {
            $message = $topic->consume(1000, RD_KAFKA_PARTITION_UA);
            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                    $data = json_decode($message->payload, true);
                    $batch[] = $data;

                    // 达到批量大小则插入
                    if (count($batch) >= $batchSize) {
                        $collection->insertMany($batch);
                        $batch = [];
                        $this->info('Inserted batch of ' . $batchSize . ' records');
                    }
                    break;
                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                    // 无新消息，处理剩余批量
                    if (!empty($batch)) {
                        $collection->insertMany($batch);
                        $batch = [];
                        $this->info('Inserted remaining ' . count($batch) . ' records');
                    }
                    usleep(100000); // 休眠100ms避免空轮询
                    break;
                default:
                    throw new \Exception($message->errstr(), $message->err);
            }
        }
    }
}

2. 配置Supervisor管理多进程

创建Supervisor配置文件（比如/etc/supervisor/conf.d/laravel-kafka.conf）：

[program:laravel-kafka-consumer]
process_name=%(program_name)s_%(process_num)02d
command=php /path/to/your/laravel/project/artisan kafka:consume
autostart=true
autorestart=true
user=www-data
numprocs=3 ; 启动3个消费者进程，对应Topic的3个分区
redirect_stderr=true
stdout_logfile=/path/to/your/laravel/project/storage/logs/kafka-consumer.log

然后重启Supervisor：

supervisorctl reread
supervisorctl update
supervisorctl start laravel-kafka-consumer:*

方式2：利用Laravel队列+Kafka驱动（如果适合你的场景） #

如果你的业务允许异步解耦，可以用Laravel队列，配置Kafka作为队列驱动，然后启动多个队列 worker：

1. 安装Laravel Kafka队列包（比如laravel-kafka/laravel-kafka）
2. 配置config/queue.php的Kafka驱动
3. 启动多个队列worker：

   php artisan queue:work kafka --queue=your_queue_name --daemon --timeout=60 --sleep=3 --tries=3 --processes=3
   

额外优化点提升吞吐量 #

• MongoDB批量插入：如上代码所示，用insertMany()替代单条insertOne()，能减少MongoDB的连接开销，大幅提升插入速度。
• 调整批量大小：根据服务器内存和MongoDB性能，调整batchSize（比如500-2000），找到最优值。
• 优化MongoDB连接：使用连接池，避免每次插入都新建连接；调整MongoDB的maxPoolSize配置。
• Kafka消费者配置优化：
  • 增大fetch.min.bytes和fetch.max.wait.ms，让消费者一次性拉取更多消息
  • 调整auto.commit.interval.ms，减少提交offset的频率
• 服务器资源优化：确保服务器CPU、内存、磁盘IO足够，Kafka Broker和MongoDB的资源也需匹配。

定时任务错误排查 #

定时任务报错常见原因及解决：

• 进程冲突：如果定时任务是启动消费者，不要重复执行，否则会出现多个同组消费者但分区分配异常。应该用Supervisor管理进程，定时任务只负责重启异常进程。
• 资源不足：检查MongoDB连接数是否耗尽（db.serverStatus().connections），Kafka连接数是否超限。
• 配置错误：确认Kafka Broker地址、消费组ID、Topic名称是否正确；MongoDB URI、数据库和集合名称是否正确。
• 日志排查：查看Laravel的storage/logs/laravel.log和Kafka消费者日志，定位具体错误信息。

内容的提问来源于stack exchange，提问作者Mộc Uyển Thanh