PHP数据管道与ApacheKafka集成

PHP数据管道与Apache Kafka集成Apache Kafka是分布式流处理平台。PHP可以通过Kafka扩展或HTTP API与Kafka集成。今天说说PHP中Kafka的集成和数据管道实现。Kafka的核心概念包括Topic、Producer和Consumer。Producer发送消息到TopicConsumer从Topic读取消息。php// Kafka生产者class KafkaProducer{private string $broker;private string $topic;public function __construct(string $broker, string $topic){$this-broker $broker;$this-topic $topic;}public function send(string $key, mixed $value): bool{$payload json_encode([key $key,value $value,timestamp time(),]);$ch curl_init({$this-broker}/topics/{$this-topic});curl_setopt_array($ch, [CURLOPT_RETURNTRANSFER true,CURLOPT_POST true,CURLOPT_POSTFIELDS $payload,CURLOPT_HTTPHEADER [Content-Type: application/vnd.kafka.json.v2json],CURLOPT_TIMEOUT 5,]);$response curl_exec($ch);$httpCode curl_getinfo($ch, CURLINFO_HTTP_CODE);curl_close($ch);return $httpCode 200;}public function sendBatch(array $messages): int{$records [];foreach ($messages as $key $value) {$records[] [key (string)$key, value $value];}$payload json_encode([records $records]);$ch curl_init({$this-broker}/topics/{$this-topic});curl_setopt_array($ch, [CURLOPT_RETURNTRANSFER true,CURLOPT_POST true,CURLOPT_POSTFIELDS $payload,CURLOPT_HTTPHEADER [Content-Type: application/vnd.kafka.json.v2json],CURLOPT_TIMEOUT 10,]);$response curl_exec($ch);$httpCode curl_getinfo($ch, CURLINFO_HTTP_CODE);curl_close($ch);return $httpCode 200 ? count($messages) : 0;}}// Kafka消费者class KafkaConsumer{private string $broker;private string $topic;private string $groupId;private int $offset 0;public function __construct(string $broker, string $topic, string $groupId php-consumer){$this-broker $broker;$this-topic $topic;$this-groupId $groupId;}public function consume(int $timeout 5000): array{$url {$this-broker}/consumers/{$this-groupId};// 创建消费者$ch curl_init($url);curl_setopt_array($ch, [CURLOPT_RETURNTRANSFER true,CURLOPT_POST true,CURLOPT_POSTFIELDS json_encode([name $this-groupId,format json,auto.offset.reset earliest,]),CURLOPT_HTTPHEADER [Content-Type: application/vnd.kafka.v2json],]);curl_exec($ch);curl_close($ch);// 订阅主题$ch curl_init({$url}/subscription);curl_setopt_array($ch, [CURLOPT_RETURNTRANSFER true,CURLOPT_POST true,CURLOPT_POSTFIELDS json_encode([topics [$this-topic]]),CURLOPT_HTTPHEADER [Content-Type: application/vnd.kafka.v2json],]);curl_exec($ch);curl_close($ch);return [];}public function subscribe(callable $handler): void{echo 消费者启动: {$this-topic}\n;while (true) {$url {$this-broker}/consumers/{$this-groupId}/records;$ch curl_init($url);curl_setopt_array($ch, [CURLOPT_RETURNTRANSFER true,CURLOPT_TIMEOUT 30,CURLOPT_HTTPHEADER [Accept: application/vnd.kafka.json.v2json],]);$response curl_exec($ch);$records json_decode($response, true) ?? [];foreach ($records as $record) {$handler($record[key], $record[value]);}curl_close($ch);usleep(100000);}}}?基于Kafka的数据管道实现phpclass KafkaDataPipeline{private KafkaProducer $producer;private array $processors [];public function __construct(KafkaProducer $producer){$this-producer $producer;}public function addProcessor(callable $processor): void{$this-processors[] $processor;}public function process(array $rawData): bool{$data $rawData;foreach ($this-processors as $processor) {$data $processor($data);if ($data null) return false;}$key $data[id] ?? uniqid();return $this-producer-send((string)$key, $data);}public function processBatch(array $items): int{$success 0;foreach ($items as $item) {if ($this-process($item)) {$success;}}return $success;}}?Kafka与PHP结合可以构建强大的数据管道。Kafka处理高吞吐量的数据流PHP负责业务逻辑处理。事件驱动架构中Kafka作为事件总线PHP服务作为事件的产生者和消费者。对于大规模的数据管道建议使用专业的Kafka客户端库如php-rdkafka扩展。