原生PHP不能操作kafka？

原生 PHP 完全可以操作 Kafka，无需 Swoole、Laravel 或其他框架。
核心依赖是 rdkafka 扩展（基于 librdkafka C 库），这是 PHP 官方支持的 Kafka 客户端，适用于 CLI 脚本、FPM、Worker 等所有原生 PHP 环境。

一、扩展安装：rdkafka 是唯一选择

📦 1. 依赖安装

# 1. 安装 librdkafka（C 库）
## Ubuntu/Debian
sudo apt-get install librdkafka-dev

## CentOS/RHEL
sudo yum install librdkafka-devel

## macOS
brew install librdkafka

# 2. 安装 PHP 扩展
pecl install rdkafka

# 3. 启用扩展
echo "extension=rdkafka.so" >> /etc/php/8.1/cli/php.ini

🔑 真相：rdkafka 是 Confluent 官方推荐的 PHP 客户端，非第三方玩具。

✅ 2. 验证安装

<?php
// test_kafka.php
if (extension_loaded('rdkafka')) {
echo "rdkafka 扩展已安装\\n";
echo "版本: " . RdKafka\\RD_KAFKA_VERSION . "\\n";
} else {
die("请安装 rdkafka 扩展");
}

✅ 这是 100% 原生 PHP 代码（无框架、无 Swoole）。

二、核心操作：生产者与消费者

📤 1. 生产者（Producer）— FPM/CLI 兼容

<?php
// producer.php
$conf = new RdKafka\\Conf();
$conf->set('metadata.broker.list', 'kafka:9092');
$conf->set('queue.buffering.max.ms', '100'); // 批量发送

$producer = new RdKafka\\Producer($conf);
$topic = $producer->newTopic("user_events");

// 发送消息
$topic->produce(
RD_KAFKA_PARTITION_UA, // 自动分配分区
0, // 消息标志
json_encode(['user_id' => 123, 'action' => 'login']),
'user_123' // Key（用于分区路由）
);

// 等待发送完成（可选）
$producer->poll(0);
echo "消息已发送\\n";

📥 2. 消费者（Consumer）— 仅 CLI

<?php
// consumer.php
$conf = new RdKafka\\Conf();
$conf->set('metadata.broker.list', 'kafka:9092');
$conf->set('group.id', 'user_events_processor'); // 固定组名！

// 错误回调
$conf->setErrorCb(function ($kafka, $err, $reason) {
error_log("Kafka error [$err]: $reason");
});

// Rebalance 回调（关键！）
$conf->setRebalanceCb(function ($kafka, $err, $partitions) {
switch ($err) {
case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
$kafka->assign($partitions);
break;
case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
$kafka->assign([]);
break;
}
});

$consumer = new RdKafka\\KafkaConsumer($conf);
$consumer->subscribe(['user_events']);

while (true) {
$message = $consumer->consume(1000); // 1秒超时
if ($message->err) {
if ($message->err === RD_KAFKA_RESP_ERR__PARTITION_EOF) {
continue; // 无新消息
}
throw new Exception($message->errstr());
}

$event = json_decode($message->payload, true);
echo "处理事件: " . $event['action'] . "\\n";

// 手动提交 Offset（Exactly-Once 关键）
$consumer->commit();
}

⚠️ 重要：Kafka 消费者必须运行在 CLI 环境（FPM 会阻塞进程池）。

3. 避坑指南：五大高危误区

🚫 误区 1：“原生 PHP 不能用 rdkafka”

• 真相：
  • rdkafka 是 PECL 官方扩展；
  • 支持所有 PHP SAPI（CLI/FPM）；
• 解法：通过 pecl install rdkafka 安装；

🚫 误区 2：“必须用 Swoole 协程”

• 真相：
  • rdkafka 是同步阻塞模型；
  • Swoole 协程需特殊适配（swoole-kafka）；
• 解法：原生 PHP 用同步模型，CLI 消费者独立进程运行；

🚫 误区 3：“动态 group.id”

• 真相：
  • group.id = uniqid() → 每个实例独立消费 → 重复处理；
• 解法：消费者组名必须固定；

🚫 误区 4：“忽略 Rebalance 回调”

• 真相：
  • 未处理 REVOKE_PARTITIONS → Offset 未提交 → 重复消费；
• 解法：必须实现 setRebalanceCb；

🚫 误区 5：“Web 请求中直接消费”

• 真相：
  • while (true) { consume() } 阻塞 FPM 进程 → 502；
• 解法：PHP 仅作为生产者，消费者用独立 CLI 脚本；

四、架构实践：原生 PHP + Kafka 的正确姿势

🌐 1. 生产者模式（FPM 安全）

• 场景：用户注册、支付事件；
• 流程：// user_register.php (FPM)
  session_start();
  // 1. 业务逻辑
  insertUser($data);

  // 2. 异步发送事件（非阻塞）
  if (function_exists('fastcgi_finish_request')) {
  fastcgi_finish_request(); // 提前返回 HTTP 响应
  }

  // 3. 发送 Kafka
  $producer = new RdKafka\\Producer($conf);
  $topic = $producer->newTopic("user_registered");
  $topic->produce(RD_KAFKA_PARTITION_UA, 0, json_encode($data));

🖥️ 2. 消费者模式（CLI 独立）

• 场景：异步发邮件、更新搜索索引；
• 部署：# Supervisor 管理消费者进程
  [program:kafka-consumer]
  command=php /app/consumer.php
  autostart=true
  autorestart=true
  numprocs=3 # 3 个消费者实例

🔁 3. 消息可靠性配置

// 生产者高可靠配置
$conf->set('acks', 'all'); // 所有 ISR 确认
$conf->set('enable.idempotence', 'true'); // 幂等
$conf->set('retries', '2147483647'); // 无限重试
$conf->set('compression.codec', 'snappy'); // 压缩

// 消费者高可靠配置
$conf->set('enable.auto.commit', 'false'); // 手动提交
$conf->set('auto.offset.reset', 'earliest'); // 从头消费

五、终极心法：Kafka 是管道，PHP 是端点

不要让 PHP 处理流，
而让 PHP 触发流。

• 错误架构：
  • FPM 中消费 Kafka → 进程耗尽；
• 正确架构：
  • FPM 生产事件 → CLI 消费处理；
• 结果：
  • 前者是系统瓶颈，后者是可靠管道。

真正的 Kafka 集成，
不在“客户端多强”，
而在“架构多清”。

六、行动建议：今日 Kafka 集成

## 2025-10-05 Kafka 集成

### 1. 安装 rdkafka
– [ ] pecl install rdkafka
– [ ] 验证 php -m | grep rdkafka

### 2. 实现生产者
– [ ] user_register.php → 发送 user_registered 事件

### 3. 实现消费者
– [ ] consumer.php → 处理事件 + 手动提交 Offset

### 4. 部署验证
– [ ] Docker 启动 Kafka
– [ ] Supervisor 运行消费者
– [ ] curl 触发生产者 → 验证消费

✅ 完成即构建原生 PHP + Kafka 系统。

当你停止用“FPM 消费”冒险，
开始用“CLI 生产者/消费者”分离，
Kafka 就从工具，
变为可靠数据管道。

这，才是专业 PHP 工程师的流处理观。