Hyperf/kafka 默认单协程单分区消费，因ConsumerProcess仅启一个进程，未触发Kafka分区并行能力；真正并发需多ConsumerProcess实例参与Rebalance，自动分摊分区，配合手动commit与合理session.timeout.ms配置。
为什么 hyperf/kafka 默认消费者不是高并发的
默认配置下，Hyperf\Kafka\Consumer 启动的是单协程单分区消费模式，即使你设置了多个 max_poll_records，也仅在一个协程里顺序拉取、顺序处理。真正卡点在于：每个 consumer group 的每个分区只能被一个消费者实例独占，而 hyperf/kafka 的 ConsumerProcess 默认只起一个进程 + 一个协程，根本没触发 Kafka 的并行分片能力。
关键不是“能不能并发”，而是“是否让多个协程/进程各自绑定不同分区”——这需要显式控制分区分配策略和消费者实例拓扑。
- 不要依赖
auto_offset_reset=latest就以为能“自动扩容”；Kafka 不会主动把分区甩给新协程 -
num_consumers配置项在hyperf/kafkav2.x+ 中已被移除，它不控制并发数，只影响连接池初始化 - 真正起作用的是
partition_assignment_strategy和进程/协程粒度的 consumer 实例数量
如何用多 ConsumerProcess 实现分区级并发
最直接有效的方式是启动多个独立的 ConsumerProcess，每个绑定固定分区（或让 Kafka 自动均衡），而非试图在一个进程内“协程化拉取消息”。Hyperf 的进程模型天然适合这种拆分。
操作步骤：
- 在
config/autoload/processes.php中注册多个 consumer 进程，例如：'kafka_consumer_user' => [ 'handler' => \App\Process\KafkaUserConsumerProcess::class, 'redirect_stdin_stdout' => false, 'pipe' => 0, ], 'kafka_consumer_order' => [ 'handler' => \App\Process\KafkaOrderConsumerProcess::class, 'redirect_stdin_stdout' => false, 'pipe' => 0, ], - 每个
Process类继承Hyperf\Kafka\ConsumerProcess，但重写getConsumerConfig()，确保group_id相同（属于同一 consumer group），且不手动指定topics分区（交由 Kafka Rebalance） - 通过
php bin/hyperf.php start启动后，每个进程会参与 group rebalance，Kafka 自动将 topic 分区分配给不同进程 —— 这才是真正的横向并发
注意：enable_auto_commit 必须设为 false，否则多个进程可能重复提交 offset；手动 commit 要放在业务处理成功后，且建议用 $consumer->commitBatch() 批量提交。
poll() 调用时机与内存泄漏风险
很多人把 $consumer->poll() 放在 while(true) 循环里，却不控制频率或消息积压，导致协程内存持续增长、GC 压力大，最终 OOM。
- 不要无休止
poll()：每次 poll 后必须检查$message === null，遇到空响应应co::sleep(0.01)让出协程，避免 busy-loop -
max_poll_records不宜设过大（如 >500），否则单次 fetch 数据量大，反向拖慢处理节奏；推荐 100–200，配合较快的业务处理速度更稳 - 若业务耗时波动大，建议用
Hyperf\AsyncQueue\Driver\RedisDriver中转消息，把 Kafka 消费和业务逻辑解耦 —— 毕竟 Kafka 不是队列中间件，它的语义是流式分发，不是任务分发
示例节选（在 handle() 内）：
while (true) {
$message = $this->consumer->poll(1000);
if ($message === null) {
co::sleep(0.01);
continue;
}
// 处理 $message->value
$this->consumer->commit();
}
如何验证是否真的并发消费了
光看日志时间戳不准，因为协程调度快、打印有竞争。真实验证要落到 Kafka 元数据和客户端行为上。
- 用
kafka-consumer-groups.sh --bootstrap-server xxx --group your_group --describe查看当前分配：输出中CONSUMER列应显示多个 client.id（对应多个进程），PARTITION应分散在不同行 - 在每个
ConsumerProcess的handle()开头加var_dump(getmypid(), \Co::tid());，确认是不同 OS 进程、不同协程 ID - 故意让某条消息处理失败（如抛异常），观察其他分区是否继续消费 —— 如果整个 consumer group 卡住，说明 offset 提交或 rebalance 逻辑有误
最容易被忽略的一点：Kafka 的 session.timeout.ms（默认 45s）必须大于 consumer 单次完整处理的最大耗时，否则会被踢出 group，触发不必要的 rebalance。这个值要在 getConsumerConfig() 里显式调大，比如设为 60000（60秒）。
