Kafka 最佳实践

本节总结 Kafka 在 PHP 项目中的最佳实践，涵盖分区设计、消息可靠投递、Exactly-Once 语义、Consumer Group 调优、性能调优以及运维监控等内容，帮助团队构建高可靠、高性能的 Kafka 消息系统。

前置知识

阅读本节前，建议先了解：Kafka 基础 和 Kafka PHP 客户端操作

分区设计

分区数量规划

分区数量直接影响 Kafka 的并行度和吞吐量。

<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

class PartitionPlanner
{
    /**
     * 计算合理的分区数量
     *
     * 原则：
     * 1. 分区数 = 目标吞吐量 / 单分区吞吐量
     * 2. 分区数 >= Consumer Group 中最大 Consumer 数量
     * 3. 分区数不超过 Broker 数 * 1000（避免过多开销）
     * 4. 单个 Broker 上的分区数不超过 2000（默认限制）
     */
    public static function calculatePartitions(
        int $targetThroughputMBps,
        int $singlePartitionThroughputKBps = 10240, // 默认 10MB/s per partition
        int $maxConsumers = 1
    ): int {
        $throughputFactor = (int) ceil(
            ($targetThroughputMBps * 1024) / $singlePartitionThroughputKBps
        );
        $consumerFactor = $maxConsumers;

        return max($throughputFactor, $consumerFactor);
    }

    /**
     * 常见场景分区建议
     */
    public static function getRecommendation(string $scenario): array
    {
        return match ($scenario) {
            'low_traffic' => [
                'description'    => '低流量（< 100 msg/s）',
                'partitions'     => 3,
                'replication'    => 2,
                'retention_hours'=> 168, // 7 天
            ],
            'medium_traffic' => [
                'description'    => '中流量（100~10000 msg/s）',
                'partitions'     => 6,
                'replication'    => 3,
                'retention_hours'=> 168,
            ],
            'high_traffic' => [
                'description'    => '高流量（> 10000 msg/s）',
                'partitions'     => 12,
                'replication'    => 3,
                'retention_hours'=> 72,
            ],
            'event_sourcing' => [
                'description'    => '事件溯源（需要长期保留）',
                'partitions'     => 6,
                'replication'    => 3,
                'retention_hours'=> 8760, // 1 年
                'compaction'      => true,
            ],
            default => [
                'description' => '默认配置',
                'partitions'  => 3,
                'replication' => 2,
            ],
        };
    }
}

Topic 命名规范

<?php

namespace App\Constants;

class KafkaTopicConstants
{
    // 命名格式：{domain}.{entity}.{event-type}.{environment}
    // 示例：order.payment.created.prod

    // 业务 Topic
    public const ORDER_EVENTS         = 'order.events';
    public const USER_EVENTS          = 'user.events';
    public const PAYMENT_EVENTS       = 'payment.events';
    public const INVENTORY_EVENTS     = 'inventory.events';

    // 系统 Topic
    public const DEAD_LETTER          = 'dead.letter';
    public const RETRY_TOPIC          = 'retry.topic';

    // Consumer Group
    public const GROUP_ORDER_PROCESSOR = 'order-processor-group';
    public const GROUP_NOTIFICATION   = 'notification-group';
    public const GROUP_ANALYTICS      = 'analytics-group';

    /**
     * 根据环境获取 Topic 名称
     */
    public static function getTopic(string $baseTopic, string $env = 'prod'): string
    {
        if ($env === 'prod') {
            return $baseTopic;
        }
        return "{$baseTopic}.{$env}";
    }
}

消息可靠投递

Producer 端保障

<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;
use RdKafka\Producer;
use RdKafka\Producer\Topic;

class ReliableProducer
{
    private Producer $producer;
    private int $retryCount = 0;
    private int $maxRetries = 3;

    public function __construct(string $brokers = '127.0.0.1:9092')
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', $brokers);

        // 可靠性配置
        $conf->set('acks', 'all');                   // 所有 ISR 副本确认
        $conf->set('enable.idempotence', 'true');     // 幂等 Producer
        $conf->set('retries', '2147483647');          // 无限重试（由 delivery.timeout.ms 控制）
        $conf->set('delivery.timeout.ms', '120000');  // 交付超时 2 分钟
        $conf->set('max.in.flight.requests.per.connection', '5'); // 幂等场景下最大 5

        // 发送失败回调
        $conf->setDrMsgCb(function (Producer $producer, \RdKafka\Message $message): void {
            error_log(sprintf(
                "Message delivery failed: %s (topic=%s, partition=%d)",
                $message->errstr(),
                $message->topic_name,
                $message->partition
            ));
        });

        // Brokers 断开回调
        $conf->setErrorCb(function (\RdKafka $kafka, int $err, string $reason): void {
            error_log("Kafka error: {$err} - {$reason}");
        });

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

    /**
     * 可靠发送（带重试）
     */
    public function sendReliable(
        string $topicName,
        string $payload,
        ?string $key = null
    ): bool {
        $topic = $this->producer->newTopic($topicName);

        for ($attempt = 1; $attempt <= $this->maxRetries; $attempt++) {
            try {
                $topic->produce(
                    partition: RD_KAFKA_PARTITION_UA,
                    key: $key,
                    payload: $payload,
                );

                // 等待消息发送确认
                $this->producer->flush(10000);

                return true;
            } catch (\Throwable $e) {
                error_log(sprintf(
                    "Send attempt %d/%d failed: %s",
                    $attempt,
                    $this->maxRetries,
                    $e->getMessage()
                ));

                if ($attempt === $this->maxRetries) {
                    // 所有重试失败，记录到死信队列或错误日志
                    error_log("Message permanently failed: {$payload}");
                    return false;
                }

                // 指数退避
                usleep((int) (100000 * pow(2, $attempt - 1)));
            }
        }

        return false;
    }

    /**
     * 发送并等待特定 Partition 的确认
     */
    public function sendWithAck(string $topicName, string $payload, ?string $key = null): array
    {
        $topic = $this->producer->newTopic($topicName);

        // 使用 producev 获取 Delivery Report
        $topic->producev(
            partition: RD_KAFKA_PARTITION_UA,
            key: $key,
            payload: $payload,
        );

        // poll 等待 Delivery Report
        while ($this->producer->getOutQLen() > 0) {
            $this->producer->poll(100);
        }

        return ['status' => 'sent'];
    }
}

Consumer 端保障

<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\KafkaConsumer;
use RdKafka\Message;

class ReliableConsumer
{
    private KafkaConsumer $consumer;
    private int $uncommittedCount = 0;
    private int $maxUncommitted = 500;

    /**
     * 带频率限制的 Offset 提交
     * 不是每条消息都提交 Offset，而是累积到一定数量后批量提交
     */
    public function consumeWithBatchCommit(callable $handler): void
    {
        while (true) {
            $message = $this->consumer->consume(1000);

            if ($message === null || $message->err !== RD_KAFKA_RESP_ERR_NO_ERROR) {
                continue;
            }

            try {
                $data = json_decode($message->payload, true) ?? $message->payload;
                $handler($data, $message);
                $this->uncommittedCount++;

                // 达到阈值后批量提交
                if ($this->uncommittedCount >= $this->maxUncommitted) {
                    $this->consumer->commitAsync();
                    $this->uncommittedCount = 0;
                }
            } catch (\Throwable $e) {
                // 处理失败，不提交 Offset
                error_log(sprintf(
                    "Consumer error: %s (offset=%d)",
                    $e->getMessage(),
                    $message->offset
                ));
            }
        }
    }

    /**
     * 优雅关闭消费者
     */
    public function gracefulShutdown(): void
    {
        // 捕获信号
        pcntl_signal(SIGTERM, function () {
            $this->consumer->commit();
            $this->consumer->close();
            exit(0);
        });
        pcntl_signal(SIGINT, function () {
            $this->consumer->commit();
            $this->consumer->close();
            exit(0);
        });
    }
}

Exactly-Once 语义

Exactly-Once 概念

语义	说明	实现复杂度
At Most Once	消息最多被处理一次（可能丢失）	最简单
At Least Once	消息至少被处理一次（可能重复）	中等
Exactly Once	消息恰好被处理一次（不丢不重）	最复杂

Kafka 事务

Kafka 事务（Kafka 2.5+）提供 Exactly-Once 语义的原子写入能力：

<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;
use RdKafka\Producer;
use RdKafka\Producer\Topic;

/**
 * Kafka 事务 Producer（需要 librdkafka 1.5+）
 * 确保多个消息的原子写入
 *
 * 使用场景：
 * - 消费 A Topic 的消息，处理后写入 B Topic 和 C Topic
 * - 需要保证 A 的 Offset 提交和 B、C 的写入同时成功或同时失败
 */
class TransactionalProducer
{
    private Producer $producer;
    private int $transactionalId;

    /**
     * 初始化事务 Producer
     * 每个事务 Producer 实例必须有唯一的 transactional.id
     */
    public function __construct(string $brokers, string $transactionalId)
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', $brokers);
        $conf->set('transactional.id', $transactionalId);
        $conf->set('transaction.timeout.ms', '60000');
        $conf->set('acks', 'all');

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

    /**
     * 在事务中发送多条消息
     * 所有消息要么全部成功，要么全部失败
     */
    public function sendInTransaction(array $messages): bool
    {
        try {
            // 初始化事务
            $this->producer->initTransactions(10000);

            // 开始事务
            $this->producer->beginTransaction();

            foreach ($messages as $msg) {
                $topic = $this->producer->newTopic($msg['topic']);
                $topic->produce(
                    partition: RD_KAFKA_PARTITION_UA,
                    key: $msg['key'] ?? null,
                    payload: $msg['payload'],
                );
            }

            $this->producer->flush(10000);

            // 提交事务
            $this->producer->commitTransaction(10000);

            return true;
        } catch (\Throwable $e) {
            // 回滚事务
            try {
                $this->producer->abortTransaction(10000);
            } catch (\Throwable $abortError) {
                error_log("Transaction abort failed: {$abortError->getMessage()}");
            }

            error_log("Transaction failed: {$e->getMessage()}");
            return false;
        }
    }
}

Exactly-Once 与幂等性

Kafka 原生只能保证消息投递的 Exactly-Once（消息不重复不丢失），但消费端的业务处理仍需实现幂等性：

• enable.idempotence=true：保证 Producer 端不产生重复消息
• Kafka 事务：保证跨多个 Topic 的原子写入
• Consumer 端：仍然需要结合业务 ID 做幂等检查

性能调优

Producer 调优

<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

use RdKafka\Conf;

class ProducerTuning
{
    /**
     * 高吞吐量 Producer 配置
     */
    public static function highThroughputConfig(string $brokers): Conf
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', $brokers);

        // 批量发送
        $conf->set('linger.ms', '10');          // 等待 10ms 积累批量
        $conf->set('batch.size', '65536');       // 批量大小 64KB
        $conf->set('buffer.memory', '67108864'); // 缓冲区 64MB

        // 压缩（减少网络传输量）
        $conf->set('compression.type', 'lz4');  // lz4 压缩（速度最快）

        // ACK 级别（权衡可靠性和性能）
        $conf->set('acks', '1');                 // Leader 确认即可

        // 异步发送（不等待确认）
        // 注意：需要配合回调处理发送失败的消息

        return $conf;
    }

    /**
     * 低延迟 Producer 配置
     */
    public static function lowLatencyConfig(string $brokers): Conf
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', $brokers);

        // 禁用批量发送
        $conf->set('linger.ms', '0');           // 立即发送
        $conf->set('batch.size', '0');           // 禁用批量

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

        // ACK 级别
        $conf->set('acks', '1');

        return $conf;
    }

    /**
     * 高可靠性 Producer 配置
     */
    public static function highReliabilityConfig(string $brokers): Conf
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', $brokers);

        // 所有副本确认
        $conf->set('acks', 'all');

        // 幂等性
        $conf->set('enable.idempotence', 'true');

        // 重试
        $conf->set('retries', '2147483647');
        $conf->set('delivery.timeout.ms', '120000');

        // 限制在途请求数
        $conf->set('max.in.flight.requests.per.connection', '5');

        return $conf;
    }
}

Consumer 调优

<?php

declare(strict_types=1);

namespace App\MessageQueue\Kafka;

class ConsumerTuning
{
    /**
     * Consumer 配置参数说明
     *
     * 会话超时相关：
     * - session.timeout.ms: Consumer 被认为"死亡"的超时时间（默认 45s）
     * - heartbeat.interval.ms: 心跳间隔（默认 3s，应 < session.timeout.ms / 3）
     * - max.poll.interval.ms: 两次 poll 调用的最大间隔（默认 5 分钟）
     *
     * 拉取相关：
     * - max.poll.records: 每次 poll 返回的最大消息数（默认 500）
     * - fetch.min.bytes: 最小拉取字节数
     * - fetch.max.wait.ms: 等待足够数据的最大时间
     * - fetch.max.bytes: 每次拉取的最大字节数
     *
     * Offset 提交相关：
     * - enable.auto.commit: 是否自动提交（推荐 false）
     * - auto.commit.interval.ms: 自动提交间隔
     */
    public static function getOptimalConfig(int $processingTimeMs): array
    {
        $maxPollInterval = max(300000, $processingTimeMs * 10); // 10 倍处理时间

        return [
            'session.timeout.ms' => '30000',
            'heartbeat.interval.ms' => '10000',
            'max.poll.interval.ms' => (string) $maxPollInterval,
            'max.poll.records' => (string) min(500, (int) (300000 / $processingTimeMs)),
        ];
    }
}

运维监控

关键监控指标

<?php

declare(strict_types=1);

namespace App\Monitor;

class KafkaMonitor
{
    /**
     * 通过 JMX 或 Kafka API 获取指标
     * 关键指标清单：
     *
     * Broker 级别：
     * - BytesIn/BytesOut: 网络吞吐量
     * - MessagesInPerSec: 消息生产速率
     * - UnderReplicatedPartitions: 副本不足的分区数（> 0 为异常）
     * - OfflinePartitions: 离线分区数（> 0 为异常）
     *
     * Topic 级别：
     * - Topic 的消息大小
     * - 各 Partition 的 Leader 所在 Broker
     *
     * Consumer Group 级别：
     * - Consumer Lag: 积压消息数（越低越好）
     * - Members: 消费者数量
     * - Assigned Partitions: 分配的分区数
     */
    public function getConsumerLag(string $brokers, string $groupId): array
    {
        $apiUrl = "http://localhost:8080/consumers/{$groupId}/lags";

        // 使用 JMX Exporter 或 Kafka REST Proxy 获取
        // 此处为伪代码，实际需要对接监控服务
        return [];
    }

    /**
     * Consumer Lag 告警
     */
    public function checkLag(array $lagData): array
    {
        $alerts = [];

        foreach ($lagData as $topic => $partitions) {
            foreach ($partitions as $partition => $lag) {
                if ($lag > 10000) {
                    $alerts[] = [
                        'level'    => 'critical',
                        'topic'     => $topic,
                        'partition' => $partition,
                        'lag'       => $lag,
                        'message'   => "Consumer lag too high: {$lag} messages",
                    ];
                } elseif ($lag > 1000) {
                    $alerts[] = [
                        'level'    => 'warning',
                        'topic'     => $topic,
                        'partition' => $partition,
                        'lag'       => $lag,
                        'message'   => "Consumer lag elevated: {$lag} messages",
                    ];
                }
            }
        }

        return $alerts;
    }
}

注意事项

1. 日志保留：根据业务需求配置 retention.ms 或 retention.bytes，避免磁盘空间不足
2. 消息大小：控制单条消息大小，超大消息会显著降低吞吐量
3. Consumer Lag：持续监控，Lag 持续增长说明消费者处理能力不足
4. Rebalance 频率：频繁 Rebalance 影响消费性能，排查原因并优化
5. 监控告警：监控 UnderReplicatedPartitions、OfflinePartitions、Consumer Lag

最佳实践总结

1. 分区设计：根据吞吐量需求和消费者数量合理规划分区数
2. 可靠投递：Producer 使用 acks=all + 幂等性 + 重试机制
3. Offset 管理：手动提交 Offset，配合业务幂等性保证 Exactly-Once
4. 性能调优：生产环境使用批量和压缩提升吞吐量
5. 消费策略：批量消费 + 异步提交 Offset，减少提交开销
6. 监控告警：全面监控 Broker、Topic、Consumer Group 的核心指标

下一节

继续学习：PHP 教程首页

参考链接

• Kafka 生产者优化
• Kafka 消费者优化
• Kafka Exactly-Once 语义
• Kafka 监控最佳实践