Trabajo con Kafka en PHP: bibliotecas, conexión y ejemplos de código

Trabajo con Kafka en PHP: bibliotecas, conexión y ejemplos de código

\n\n

Apache Kafka es una plataforma distribuida para transmisión de datos en tiempo real que permite publicar, almacenar y procesar flujos de mensajes. En el ecosistema PHP, para trabajar con Kafka se utiliza la extensión php-rdkafka, un envoltorio sobre la biblioteca librdkafka escrita en C. En este artículo analizaremos cómo instalar la herramienta, escribir un productor y un consumidor, y veremos casos de uso típicos.

\n\n

¿Qué es php-rdkafka?

\n\n

php-rdkafka es la extensión oficial de PHP para trabajar con Apache Kafka. Proporciona una interfaz de alto rendimiento para enviar y recibir mensajes. La extensión soporta todas las funcionalidades clave de Kafka: tópicos, particiones, grupos de consumidores, commits manuales, compresión y mucho más.

\n\n

Características principales:

\n

\n
• Creación de productores para enviar mensajes a tópicos.
\n
• Creación de consumidores para leer mensajes de tópicos.
\n
• Soporte para grupos de consumidores y rebalanceo automático.
\n
• Trabajo con particiones y gestión manual de offsets.
\n
• Compresión de mensajes (gzip, snappy, lz4, zstd).
\n
• Envío asíncrono y procesamiento por lotes.
\n

\n\n

Instalación de php-rdkafka

\n\n

La instalación de la extensión es posible a través de PECL o compilando desde el código fuente. El método recomendado es mediante PECL:

\n\n

# Instalación mediante PECL\npecl install rdkafka\n\n# Para Ubuntu/Debian, instale primero librdkafka-dev\nsudo apt-get install librdkafka-dev\npecl install rdkafka\n\n# Agregue la extensión en php.ini\nextension=rdkafka.so

\n\n

Después de la instalación, verifique que la extensión esté cargada:

\n\n

php -m | grep rdkafka

\n\n

Si usa Composer, adicionalmente puede instalar el paquete arnaud-lb/php-rdkafka para la autocarga:

\n\n

composer require arnaud-lb/php-rdkafka

\n\n

Ejemplo de productor

\n\n

El productor envía mensajes a un tópico de Kafka. Veamos un ejemplo básico con envío 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// Envío de mensaje\n$topic->produce(RD_KAFKA_PARTITION_UA, 0, '¡Hola, Kafka desde PHP!');\n\n// Esperar el envío de todos los mensajes\n$producer->flush(1000);\necho "Mensaje enviado\\n";

\n\n

Explicación del código:

\n

\n
• metadata.broker.list — dirección del broker de Kafka (por defecto localhost:9092).
\n
• RD_KAFKA_PARTITION_UA — selección automática de partición.
\n
• flush(1000) — bloquea la ejecución hasta enviar todos los mensajes (timeout de 1 segundo).
\n

\n\n

Ejemplo de consumidor

\n\n

El consumidor lee mensajes de un tópico. Es importante especificar el grupo de consumidores y la estrategia 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 "Esperando mensajes...\\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 "Mensaje recibido: " . $message->payload . "\\n";\n            break;\n        case RD_KAFKA_RESP_ERR__PARTITION_EOF:\n            echo "Fin de partición alcanzado\\n";\n            break;\n        case RD_KAFKA_RESP_ERR__TIMED_OUT:\n            echo "Timeout\\n";\n            break;\n        defau