PHP小码农

Appearance

Kafka PHP 客户端操作 ​

PHP 中操作 Kafka 主要通过 rdkafka 扩展（基于 librdkafka 的 C 扩展），提供高性能的生产和消费能力。本节将详细讲解 rdkafka 的安装配置、消息生产、消息消费、分区策略、Consumer Group Rebalance 以及生产级封装实践。

前置知识

阅读本节前，建议先了解：Kafka 基础 和 进程管理

安装与配置 ​

安装 rdkafka 扩展 ​

bash
# 安装 librdkafka（C 库）
# macOS
brew install librdkafka

# CentOS / RHEL
yum install -y librdkafka-devel

# Ubuntu / Debian
apt-get install -y librdkafka-dev

# 安装 PHP rdkafka 扩展
pecl install rdkafka

# 启用扩展
echo "extension=rdkafka.so" >> /path/to/php.ini

# 验证安装
php -m | grep rdkafka
php -r 'echo phpversion("rdkafka");'

连接配置 ​

php
<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;
use RdKafka\Producer;
use RdKafka\KafkaConsumer;

class KafkaConfigFactory
{
    /**
     * 创建 Producer 配置
     */
    public static function createProducerConfig(string $brokers = '127.0.0.1:9092'): Conf
    {
        $conf = new Conf();

        // Broker 地址列表
        $conf->set('bootstrap.servers', $brokers);

        // 消息发送超时
        $conf->set('message.timeout.ms', 5000);

        // 请求超时
        $conf->set('request.timeout.ms', 10000);

        // 重试配置
        $conf->set('retry.backoff.ms', 100);
        $conf->set('message.send.max.retries', 3);

        // ACK 确认级别
        // 0（none）：不等确认，最快但可能丢消息
        // 1（leader）：Leader 确认即可，折中方案
        // all（-1）：所有 ISR 副本确认，最可靠
        $conf->set('acks', 'all');

        // 启用幂等 Producer（防止重试导致重复）
        $conf->set('enable.idempotence', 'true');

        // 批量发送配置
        $conf->set('linger.ms', 5);         // 等待 5ms 累积批量消息
        $conf->set('batch.size', 16384);    // 批量大小 16KB

        // 压缩
        $conf->set('compression.type', 'snappy');

        // 日志级别
        $conf->set('log_level', 3); // 0=debug, 1=info, 2=notice, 3=warning, 4=error, 5=critical, 6=alert, 7=emerg

        return $conf;
    }

    /**
     * 创建 Consumer 配置
     */
    public static function createConsumerConfig(
        string $brokers = '127.0.0.1:9092',
        string $groupId = 'php-consumer-group'
    ): Conf {
        $conf = new Conf();

        $conf->set('bootstrap.servers', $brokers);
        $conf->set('group.id', $groupId);

        // Offset 提交策略
        $conf->set('enable.auto.commit', 'false');  // 关闭自动提交，手动控制
        $conf->set('auto.commit.interval.ms', '5000');

        // Offset 初始策略（当没有已提交的 Offset 时）
        // earliest: 从最早的消息开始消费
        // latest: 从最新的消息开始消费（默认）
        // none: 抛出异常（需要手动指定 Offset）
        $conf->set('auto.offset.reset', 'earliest');

        // 心跳和会话超时
        $conf->set('session.timeout.ms', 30000);       // 会话超时 30 秒
        $conf->set('heartbeat.interval.ms', 10000);     // 心跳间隔 10 秒
        $conf->set('max.poll.interval.ms', 300000);     // 最大处理时间 5 分钟

        // 每次拉取的最大消息数
        $conf->set('max.poll.records', 500);

        // 拉取超时
        $conf->set('fetch.max.wait.ms', 500);
        $conf->set('fetch.min.bytes', 1);

        // Rebalance 回调
        $conf->setRebalanceCb(function (RdKafka\KafkaConsumer $consumer, $err, array $partitions = null) {
            match ($err) {
                RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS => $consumer->assign($partitions),
                RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS => $consumer->assign(),
                default => null,
            };
        });

        return $conf;
    }
}

消息生产 ​

基础消息生产 ​

php
<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;
use RdKafka\Producer;
use RdKafka\Producer\Topic;
use RdKafka\Producer as RdKafkaProducer;

class KafkaProducerService
{
    private RdKafkaProducer $producer;
    private Topic $topic;

    public function __construct(string $brokers = '127.0.0.1:9092')
    {
        $conf = KafkaConfigFactory::createProducerConfig($brokers);
        $this->producer = new RdKafkaProducer($conf);
        $this->topic = $this->producer->newTopic('order-events');
    }

    /**
     * 发送消息（同步等待确认）
     */
    public function send(string $message, ?string $key = null): bool
    {
        try {
            // produce 方法
            // RD_KAFKA_PARTITION_UA: 自动分配分区
            // $key: 用于分区路由的 Key（相同 Key 到同一 Partition）
            // $message: 消息内容
            $this->topic->produce(
                partition: RD_KAFKA_PARTITION_UA,
                msgflags: 0,
                key: $key,
                payload: $message,
            );

            // 必须调用 poll 让底层处理事件（发送消息、回调等）
            // 在实际应用中，应该在循环中调用
            $this->producer->poll(0);

            return true;
        } catch (\Throwable $e) {
            return false;
        }
    }

    /**
     * 发送 JSON 消息
     */
    public function sendJson(string $topicName, array $data, ?string $key = null): bool
    {
        $topic = $this->producer->newTopic($topicName);

        $payload = json_encode(array_merge([
            'event_id'   => uniqid('evt_', true),
            'event_time' => date('Y-m-d\TH:i:s\Z'),
            'version'    => '1.0',
        ], $data));

        $topic->produce(
            partition: RD_KAFKA_PARTITION_UA,
            key: $key,
            payload: $payload,
        );

        $this->producer->poll(0);

        return true;
    }

    /**
     * 刷新消息队列（确保所有消息发送完成）
     */
    public function flush(int $timeoutMs = 10000): void
    {
        $this->producer->flush($timeoutMs);
    }

    public function __destruct()
    {
        $this->producer->flush(10000);
    }
}

异步生产与回调 ​

php
<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;
use RdKafka\Message;
use RdKafka\Producer;

class AsyncKafkaProducer
{
    private Producer $producer;
    private int $pendingCallbacks = 0;

    public function __construct(string $brokers = '127.0.0.1:9092')
    {
        $conf = KafkaConfigFactory::createProducerConfig($brokers);

        // 设置消息发送回调
        $conf->setDrMsgCb(function (Producer $producer, Message $message): void {
            if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
                // 发送成功
                // log: "Message delivered to partition {$message->partition} at offset {$message->offset}"
            } else {
                // 发送失败
                // log: "Delivery failed: {$message->errstr()}"
            }
            $this->pendingCallbacks--;
        });

        $this->producer = new Producer($conf);
    }

    /**
     * 异步发送消息（不等待确认）
     */
    public function sendAsync(string $topicName, string $payload, ?string $key = null): void
    {
        $topic = $this->producer->newTopic($topicName);

        $topic->produce(
            partition: RD_KAFKA_PARTITION_UA,
            key: $key,
            payload: $payload,
        );

        $this->pendingCallbacks++;
    }

    /**
     * 事件循环（处理回调）
     */
    public function poll(int $timeoutMs = 100): void
    {
        while ($this->pendingCallbacks > 0) {
            $this->producer->poll($timeoutMs);
        }
    }
}

消息消费 ​

基础消息消费 ​

php
<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;
use RdKafka\KafkaConsumer;
use RdKafka\Message;
use RdKafka\TopicPartition;

class KafkaConsumerService
{
    private KafkaConsumer $consumer;

    public function __construct(
        string $brokers,
        string $groupId,
        array $topics
    ) {
        $conf = KafkaConfigFactory::createConsumerConfig($brokers, $groupId);
        $this->consumer = new KafkaConsumer($conf);
        $this->consumer->subscribe($topics);
    }

    /**
     * 消费消息循环
     */
    public function consume(callable $handler, int $timeoutMs = 1000): void
    {
        while (true) {
            $message = $this->consumer->consume($timeoutMs);

            if ($message === null) {
                continue;
            }

            match ($message->err) {
                RD_KAFKA_RESP_ERR_NO_ERROR => $this->handleMessage($message, $handler),
                RD_KAFKA_RESP_ERR__PARTITION_EOF => null, // 到达分区末尾
                RD_KAFKA_RESP_ERR__TIMED_OUT => null,    // 消费超时
                default => throw new \RuntimeException(
                    'Kafka consume error: ' . $message->errstr()
                ),
            };
        }
    }

    /**
     * 处理单条消息
     */
    private function handleMessage(Message $message, callable $handler): void
    {
        try {
            $data = json_decode($message->payload, true) ?? $message->payload;

            $result = $handler([
                'topic'     => $message->topic_name,
                'partition' => $message->partition,
                'offset'    => $message->offset,
                'key'       => $message->key,
                'headers'   => $message->headers ?? [],
                'timestamp' => $message->timestamp,
                'payload'   => $data,
            ]);

            if ($result === true) {
                // 同步提交当前 Offset
                $this->consumer->commit($message);
            }
            // result === false: 不提交 Offset（消息将被重新消费）
        } catch (\Throwable $e) {
            // 异常处理：记录错误但不提交 Offset
            error_log(sprintf(
                "Kafka consume error: %s (topic=%s, partition=%d, offset=%d)",
                $e->getMessage(),
                $message->topic_name,
                $message->partition,
                $message->offset
            ));
        }
    }

    /**
     * 手动提交 Offset
     */
    public function commit(): void
    {
        $this->consumer->commit();
    }

    /**
     * 异步提交 Offset（性能更好，但不保证立即完成）
     */
    public function commitAsync(): void
    {
        $this->consumer->commitAsync();
    }

    /**
     * 关闭消费者
     */
    public function close(): void
    {
        $this->consumer->close();
    }
}

指定 Offset 消费 ​

php
<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\KafkaConsumer;
use RdKafka\TopicPartition;

class OffsetSeeker
{
    /**
     * 从指定 Offset 开始消费
     */
    public function seekToOffset(
        KafkaConsumer $consumer,
        string $topic,
        int $partition,
        int $offset
    ): void {
        $topicPartition = new TopicPartition($topic, $partition, $offset);
        $consumer->assign([$topicPartition]);
    }

    /**
     * 从最早的消息开始消费
     */
    public function seekToBeginning(KafkaConsumer $consumer, string $topic, int $partition): void
    {
        $topicPartition = new TopicPartition($topic, $partition);
        $consumer->seekToBeginning([$topicPartition]);
    }

    /**
     * 从最新的消息开始消费
     */
    public function seekToEnd(KafkaConsumer $consumer, string $topic, int $partition): void
    {
        $topicPartition = new TopicPartition($topic, $partition);
        $consumer->seekToEnd([$topicPartition]);
    }

    /**
     * 获取指定 Partition 的水位（Watermark）
     */
    public function getWatermarks(
        KafkaConsumer $consumer,
        string $topic,
        int $partition
    ): array {
        $consumer->queryWatermarkOffsets(
            $topic,
            $partition,
            $low,
            $high,
            1000 // timeout ms
        );

        return [
            'low'  => $low,   // 最早可用 Offset
            'high' => $high,  // 最新 Offset（下一条待写入的）
        ];
    }

    /**
     * 获取消费者当前的 Offset（已提交的）
     */
    public function getCommittedOffsets(
        KafkaConsumer $consumer,
        string $topic,
        int $partition
    ): ?int {
        $topicPartition = new TopicPartition($topic, $partition);
        $committed = $consumer->getCommittedOffsets([$topicPartition], 1000);

        if (isset($committed[$topic->partition])) {
            return $committed[$topic->partition]->getOffset();
        }

        return null;
    }
}

分区策略 ​

自定义分区器 ​

php
<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;
use RdKafka\Producer;

class CustomPartitioner
{
    /**
     * 创建带自定义分区器的 Producer
     * rdkafka 默认使用 consistent_random 分区器（hash(key) + 随机）
     */
    public static function createWithPartitioner(string $brokers, string $partitionerName): Producer
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', $brokers);

        // 内置分区器：
        // - consistent_random（默认）：Key 存在则 hash，否则随机
        // - consistent：纯 hash
        // - random：纯随机
        // - murmur2：MurmurHash2（Java Kafka 默认）
        // - murmur2_random：MurmurHash2 + 随机
        $conf->set('partitioner', $partitionerName);

        // 自定义分区回调（仅 consistent/consistent_random 支持）
        $conf->setPartitionerCb(function (TopicPartition $topicPartition, array $key, array $partitions) {
            // $key: [0 => key_bytes, 1 => key_length]
            // $partitions: 可用分区 ID 列表

            if ($key[1] > 0) {
                // 有 Key，按 hash 分区
                $hashValue = crc32($key[0]);
                $partitionIndex = abs($hashValue) % count($partitions);
                $topicPartition->setPartition($partitions[$partitionIndex]);
            } else {
                // 无 Key，轮询分区
                static $counter = 0;
                $partitionIndex = $counter % count($partitions);
                $counter++;
                $topicPartition->setPartition($partitions[$partitionIndex]);
            }
        });

        return new Producer($conf);
    }
}

Consumer Group Rebalance ​

Rebalance 监听 ​

php
<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;
use RdKafka\KafkaConsumer;
use RdKafka\TopicPartition;

class RebalanceAwareConsumer
{
    private KafkaConsumer $consumer;

    public function __construct(string $brokers, string $groupId, array $topics)
    {
        $conf = KafkaConfigFactory::createConsumerConfig($brokers, $groupId);

        // 注册 Rebalance 回调
        $conf->setRebalanceCb(function (KafkaConsumer $consumer, int $err, ?array $partitions = null): void {
            switch ($err) {
                case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
                    // 被分配了新的 Partition
                    $consumer->assign($partitions);
                    $this->onPartitionsAssigned($partitions);
                    break;

                case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
                    // Partition 被回收（Rebalance 前）
                    $this->onPartitionsRevoked($partitions);
                    $consumer->assign();
                    break;

                default:
                    // 异常
                    error_log("Rebalance error: {$err}");
                    break;
            }
        });

        $this->consumer = new KafkaConsumer($conf);
        $this->consumer->subscribe($topics);
    }

    /**
     * Partition 分配回调
     * 可在此处初始化状态、恢复 Offset
     */
    private function onPartitionsAssigned(array $partitions): void
    {
        foreach ($partitions as $tp) {
            error_log(sprintf(
                "Assigned partition: topic=%s, partition=%d, offset=%d",
                $tp->getTopic(),
                $tp->getPartition(),
                $tp->getOffset()
            ));
        }
    }

    /**
     * Partition 回收回调
     * 可在此处提交 Offset、清理资源
     */
    private function onPartitionsRevoked(array $partitions): void
    {
        // 提交当前处理的 Offset
        $this->consumer->commit();

        foreach ($partitions as $tp) {
            error_log(sprintf(
                "Revoked partition: topic=%s, partition=%d",
                $tp->getTopic(),
                $tp->getPartition()
            ));
        }
    }
}

实战示例 ​

php
<?php

declare(strict_types=1);

// producer_example.php — 生产消息

require __DIR__ . '/vendor/autoload.php';

use App\MessageQueue\Kafka\KafkaProducerService;

$producer = new KafkaProducerService('127.0.0.1:9092');

// 发送订单创建事件
$producer->sendJson('order-events', [
    'event_type' => 'order.created',
    'order_id'   => 1001,
    'user_id'    => 2001,
    'amount'     => 99.90,
    'items'      => [
        ['product_id' => 'SKU-001', 'quantity' => 2],
        ['product_id' => 'SKU-002', 'quantity' => 1],
    ],
], 'order_id_1001');

$producer->flush();

// consumer_example.php — 消费消息

require __DIR__ . '/vendor/autoload.php';

use App\MessageQueue\Kafka\KafkaConsumerService;

$consumer = new KafkaConsumerService(
    brokers: '127.0.0.1:9092',
    groupId: 'order-processor',
    topics: ['order-events'],
);

$consumer->consume(function (array $message): bool {
    $payload = $message['payload'];
    $eventType = $payload['event_type'] ?? 'unknown';

    error_log(sprintf(
        "Processing event: %s (topic=%s, partition=%d, offset=%d)",
        $eventType,
        $message['topic'],
        $message['partition'],
        $message['offset']
    ));

    // 处理业务逻辑...
    // OrderService::handleEvent($payload);

    return true; // ACK
});

注意事项 ​

  1. poll 调用：Producer 必须定期调用 poll() 来处理底层事件，否则可能内存泄漏
  2. 手动提交：生产环境使用手动提交 Offset（enable.auto.commit=false），确保消息处理成功后再提交
  3. 错误处理：消费异常时不要提交 Offset，让消息被重新消费
  4. Consumer Lag：监控消费者 Lag（消息积压量），及时扩容
  5. 资源清理：Consumer 进程退出前必须调用 close()，否则会触发 Rebalance

最佳实践 ​

  1. 配置调优：根据业务调整 linger.ms、batch.size、compression.type 等参数
  2. 幂等生产：开启 enable.idempotence=true 防止网络重试导致消息重复
  3. 批量消费：在 Consumer 中批量处理消息后统一提交 Offset，减少提交次数
  4. 优雅退出：捕获 SIGTERM/SIGINT 信号，先提交 Offset 再关闭 Consumer
  5. 连接复用：Producer 和 Consumer 应长期运行，避免频繁创建销毁

下一节

继续学习：Kafka 最佳实践

参考链接 ​