Trabalhando com Kafka em PHP: bibliotecas, conexão e exemplos de código

Trabalhando com Kafka em PHP: bibliotecas, conexão e exemplos de código

\n\n

Apache Kafka é uma plataforma distribuída para streaming de dados que permite publicar, armazenar e processar fluxos de mensagens em tempo real. No ecossistema PHP, a extensão php-rdkafka é usada para trabalhar com Kafka — um wrapper sobre a biblioteca librdkafka, escrita em C. Neste artigo, vamos abordar como instalar a ferramenta, escrever um produtor e um consumidor, além de analisar cenários típicos de uso.

\n\n

O que é php-rdkafka?

\n\n

php-rdkafka é a extensão oficial do PHP para trabalhar com Apache Kafka. Ela fornece uma interface de alto desempenho para enviar e receber mensagens. A extensão suporta todos os recursos principais do Kafka: tópicos, partições, grupos de consumidores, commits manuais, compactação e muito mais.

\n\n

Principais recursos:

\n

\n
• Criação de produtores (producers) para enviar mensagens para tópicos.
\n
• Criação de consumidores (consumers) para ler mensagens de tópicos.
\n
• Suporte a grupos de consumidores e rebalanceamento automático.
\n
• Trabalho com partições e gerenciamento manual de offsets.
\n
• Compactação de mensagens (gzip, snappy, lz4, zstd).
\n
• Envio assíncrono e processamento em lote.
\n

\n\n

Instalação do php-rdkafka

\n\n

A instalação da extensão pode ser feita via PECL ou compilação a partir do código-fonte. A maneira recomendada é via PECL:

\n\n

# Instalação via PECL\npecl install rdkafka\n\n# Para Ubuntu/Debian, instale previamente o librdkafka-dev\nsudo apt-get install librdkafka-dev\npecl install rdkafka\n\n# Adicione a extensão no php.ini\nextension=rdkafka.so

\n\n

Após a instalação, verifique se a extensão foi carregada:

\n\n

php -m | grep rdkafka

\n\n

Se você usa Composer, também pode instalar o pacote arnaud-lb/php-rdkafka para carregamento automático:

\n\n

composer require arnaud-lb/php-rdkafka

\n\n

Exemplo de produtor (Producer)

\n\n

O produtor envia mensagens para um tópico Kafka específico. Veja um exemplo básico com envio síncrono:

\n\n

<?php\nrequire_once 'vendor/autoload.php';\n\n$conf = new RdKafka\\Conf();\n$conf->set('metadata.broker.list', 'localhost:9092');\n\n$producer = new RdKafka\\Producer($conf);\n$topic = $producer->newTopic('test_topic');\n\n// Envio de mensagem\n$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Olá, Kafka do PHP!');\n\n// Aguarda o envio de todas as mensagens\n$producer->flush(1000);\necho "Mensagem enviada\\n";

\n\n

Explicações do código:

\n

\n
• metadata.broker.list — endereço do broker Kafka (padrão localhost:9092).
\n
• RD_KAFKA_PARTITION_UA — seleção automática de partição.
\n
• flush(1000) — bloqueia a execução até que todas as mensagens sejam enviadas (timeout de 1 segundo).
\n

\n\n

Exemplo de consumidor (Consumer)

\n\n

O consumidor lê mensagens de um tópico. É importante especificar o grupo do consumidor e a estratégia de commit de offsets:

\n\n

<?php\nrequire_once 'vendor/autoload.php';\n\n$conf = new RdKafka\\Conf();\n$conf->set('metadata.broker.list', 'localhost:9092');\n$conf->set('group.id', 'my_consumer_group');\n$conf->set('auto.offset.reset', 'earliest');\n\n$consumer = new RdKafka\\KafkaConsumer($conf);\n$consumer->subscribe(['test_topic']);\n\necho "Aguardando mensagens...\\n";\n\nwhile (true) {\n    $message = $consumer->consume(120 * 1000); // timeout de 2 minutos\n    switch ($message->err) {\n        case RD_KAFKA_RESP_ERR_NO_ERROR:\n            echo "Mensagem recebida: " . $message->payload . "\\n";\n            break;\n        case RD_KAFKA_RESP_ERR__PARTITION_EOF:\n            echo "Fim da partição alcançado\\n";\n            break;\n        case RD_KAFKA_RESP_ERR__TIMED_OUT:\n            echo "Timeout\\n";\n            break;\n        defau