This guide demonstrates how to integrate PHP applications with Apache Kafka using the rdkafka extension. The implementation covers both message producer and consumer patterns with practical examples.

Environment Setup

The following environment components are required for this integration:

• PHP 7.3 or higher
• Laravel 8 (or any PHP framework)
• Apache Kafka 2.6.0
• rdkafka PHP extension

Installing the rdkafka Extension

The rdkafka extension is the recommended PHP library for Kafka communication. Other PHP Kafka extensions suffer from outdated maintenance or complex dependency isues.

Precompiled Window DLLs are available through PECL. After downloading, perform the following steps:

1. Place php_rdkafka.dll in the PHP ext directory
2. Copy librdkafka.dll to the PHP root directory
3. Add the following line to php.ini:

   extension=php_rdkafka.dll
   

Verify the installation by running php -m and checking phpinfo() for the rdkafka module.

Producer Implementation

The producer service handles sending messages to Kafka topics. Below is a clean implementation:

<?php

namespace App\Services;

use RdKafka\Conf;
use RdKafka\Producer;

class MessageProducerService {

    public static function publish(string $message): void {
        $config = new Conf();
        $producer = new Producer($config);
        $producer->addBrokers('192.168.1.100');
        
        $topic = $producer->newTopic('user_events');
        $topic->produce(RD_KAFKA_PARTITION_UA, 0, $message);
        
        $producer->poll(1);
    }
}

Important: The poll() method must be called to trigger message delivery. Without it, messages will not reach the broker.

Consumer Implementation

Low-Level Consumer

The low-level consumer provides direct control over partition handling and offset management:

<?php

namespace App\Services;

use RdKafka\Conf;
use RdKafka\Consumer;

class MessageConsumerService {

    private const BROKER = '192.168.1.100';
    private const TOPIC = 'user_events';
    private const TIMEOUT_MS = 5000;

    public static function listen(): void {
        $config = new Conf();
        $consumer = new Consumer($config);
        $consumer->addBrokers(self::BROKER);

        $topic = $consumer->newTopic(self::TOPIC);
        $topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);

        while (true) {
            $message = $topic->consume(0, self::TIMEOUT_MS);
            
            if (null === $message || $message->err === RD_KAFKA_RESP_ERR__PARTITION_EOF) {
                continue;
            }
            
            if ($message->err) {
                error_log($message->errstr());
                break;
            }
            
            echo $message->payload . PHP_EOL;
        }
    }
}

High-Level Consumer

The high-level consumer (KafkaConsumer) provides automatic partition rebalancing and offset management:

<?php

namespace App\Services;

use RdKafka\Conf;
use RdKafka\TopicConf;
use RdKafka\KafkaConsumer;

class EventConsumerService {

    private const BROKER = '192.168.1.100';
    private const TOPIC = 'user_events';
    private const GROUP_ID = 'event_processors';

    public static function process(): void {
        $config = new Conf();
        
        $config->setRebalanceCb(function (KafkaConsumer $kafka, $err, array $partitions = null) {
            match ($err) {
                RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS => $kafka->assign($partitions),
                RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS => $kafka->assign(null),
                default => throw new \RuntimeException($err),
            };
        });

        $config->set('group.id', self::GROUP_ID);
        $config->set('metadata.broker.list', self::BROKER);
        $config->set('auto.offset.reset', 'earliest');

        $consumer = new KafkaConsumer($config);
        
        $topicConfig = new TopicConf();
        $topicConfig->set('enable.auto.commit', 'false');
        
        $topic = $consumer->newTopic(self::TOPIC, $topicConfig);
        $consumer->subscribe([self::TOPIC]);

        while (true) {
            $message = $consumer->consume(1000);
            
            match ($message->err) {
                RD_KAFKA_RESP_ERR_NO_ERROR => self::handleMessage($message),
                RD_KAFKA_RESP_ERR__PARTITION_EOF => usleep(100000),
                RD_KAFKA_RESP_ERR__TIMED_OUT => null,
                default => throw new \RuntimeException($message->errstr(), $message->err),
            };
        }
    }

    private static function handleMessage($message): void {
        echo $message->payload . PHP_EOL;
    }
}

Kafka Administration Commands

The following commands assist with Kafkka cluster management and testing:

# Start ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties

# Start Kafka Broker
bin/kafka-server-start.sh config/server.properties

# Create a new topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic user_events

# List all topics
bin/kafka-topics.sh --list --zookeeper localhost:2181

# Delete a topic
bin/kafka-topics.sh --delete --topic user_events --zookeeper localhost:2181

# Test with console producer
bin/kafka-console-producer.sh --broker-list 192.168.1.100:9092 --topic user_events

# Test with console consumer
bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.100:9092 --topic user_events --from-beginning

# Modify topic retention
bin/kafka-configs --zookeeper localhost:2181 --entity-type topics --entity-name user_events --alter --add-config retention.ms=3000

# Check running Java processes
jps

Key Considerations

• Ensure Kafka and ZooKeeper are running before testing PHP consumers or producers
• The poll() call in producers is essential for message delivery
• Choose between low-level and high-level consumers based on your partition management needs
• Verify broker connectivity before implementing application logic

Alternative Considerations

While Kafka excels in high-throughput scenarios and scales effectively with clustering, RabbitMQ offers simpler configuration and faster onboarding for teams requiring multi-language support. For Java-centric architectures, Kafka provides excellent integration capabilities. Teams using diverse technology stacks may find RabbitMQ more suitable due to its broader protocol support and simpler operational requirements.