Laravel中无法消费Kafka主题消息的问题排查求助

环境说明 #

• Laravel 10
• PHP 8.1
• mateusjunges/laravel-kafka 1.13
• Kafka Broker已确认接收消息，目标主题分区已创建

问题描述 #

生产者可成功发送消息至Kafka Broker，但通过HTTP路由触发消费者后，TestHandler中的日志未被记录，消息未被消费。

相关代码片段 #

生产者代码 #

<?php

namespace App\Http\Controllers;

use Illuminate\Http\Request;
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Message\Message;

class KafkaProducerController extends Controller
{
    public function produce(Request $request)
    {
        $new_message = new Message(
            headers: ['header-key' => 'header-value'],
            body: ['key' => 'value'],
            key: 'kafka key here'  
        );
        
        $producer = Kafka::publishOn('ramses')->withMessage($new_message);
        $producer->send();
    }
}

消费者代码 #

<?php

namespace App\Http\Controllers;

use Illuminate\Http\Request;
use Illuminate\Http\Response;
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Contracts\KafkaConsumerMessage;
use Illuminate\Support\Facades\Log;
use App\Handlers\TestHandler;

class KafkaConsumerController extends Controller
{
    public function consume(Request $request)
    {
        $messageReceived = '';
        
        $consumer = Kafka::createConsumer()
            ->subscribe('ramses')
            ->withBrokers('localhost:9092')
            ->withConsumerGroupId('group')
            ->withHandler(new TestHandler)
            ->build();
            
        $consumer->consume();

        return response()->make($messageReceived, 200);
    }
}

路由配置 #

<?php

use Illuminate\Support\Facades\Route;
use App\Http\Controllers\KafkaProducerController;
use App\Http\Controllers\KafkaConsumerController;

Route::get('/produce', [KafkaProducerController::class, 'produce']);
Route::get('/consume', [KafkaConsumerController::class, 'consume']);

消息处理器 #

<?php

namespace App\Handlers;

use Illuminate\Support\Facades\Log;
use Junges\Kafka\Contracts\KafkaConsumerMessage;

class TestHandler
{
    public function __invoke(KafkaConsumerMessage $message)
    {
        Log::debug('Message received!', [
            'body' => $message->getBody(),
            'headers' => $message->getHeaders(),
            'partition' => $message->getPartition(),
            'key' => $message->getKey(),
            'topic' => $message->getTopicName()
        ]);
    }
}

Kafka配置文件 #

<?php

return [
    'brokers' => env('KAFKA_BROKERS', 'localhost:9092'),

    'consumer_group_id' => env('KAFKA_CONSUMER_GROUP_ID', 'group'),

    'offset_reset' => env('KAFKA_OFFSET_RESET', 'latest'),

    'auto_commit' => env('KAFKA_AUTO_COMMIT', true),

    'sleep_on_error' => env('KAFKA_ERROR_SLEEP', 5),

    'partition' => env('KAFKA_PARTITION', 0),

    'compression' => env('KAFKA_COMPRESSION_TYPE', 'snappy'),

    'debug' => env('KAFKA_DEBUG', true),
];

排查方向 #

1. 消费者运行模式错误 #

Kafka消费者默认是长轮询阻塞模式，通过HTTP路由触发时，HTTP请求会因超时提前终止消费者进程，导致无法等待和处理消息。

解决方法：改用Artisan命令启动消费者：

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Junges\Kafka\Facades\Kafka;
use App\Handlers\TestHandler;

class KafkaConsumer extends Command
{
    protected $signature = 'kafka:consume';
    protected $description = 'Start Kafka consumer for ramses topic';

    public function handle()
    {
        $consumer = Kafka::createConsumer()
            ->subscribe('ramses')
            ->withConsumerGroupId('group')
            ->withHandler(new TestHandler)
            ->build();

        $consumer->consume();
    }
}

执行命令启动消费者：

php artisan kafka:consume

2. 消费者组偏移量问题 #

若该消费者组曾提交过偏移量，且offset_reset设置为latest，消费者只会消费启动后新发送的消息，不会处理历史消息。

解决方法：

• 临时将offset_reset改为earliest
• 手动重置消费者组偏移量（Kafka命令行工具）：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group group --reset-offsets --to-earliest --topic ramses --execute

3. 日志级别限制 #

TestHandler使用Log::debug()，若Laravel日志级别设置高于debug（如.env中LOG_LEVEL=info），则debug日志不会被记录。

解决方法：

• 将.env中LOG_LEVEL改为debug
• 替换日志方法为Log::info()

4. 依赖扩展问题 #

确保rdkafka扩展已正确安装并启用，执行以下命令检查：

php -m | grep rdkafka

5. 配置冲突 #

消费者代码中手动指定了withBrokers和withConsumerGroupId，需确保与.env中的配置一致，避免冲突。

替代方案 #

若mateusjunges/laravel-kafka使用存在问题，可考虑：

• 使用laravel-kafka/laravel-kafka（原项目的后续维护版本，适配新版Laravel）
• 基于rdkafka扩展封装原生生产者/消费者逻辑
• 使用Laravel队列的Kafka驱动（如laravel-kafka/queue），将Kafka作为队列后端使用

内容的提问来源于stack exchange，提问作者Ramses Kouam