README

Contents

• Contents
  • About
  • Requirements
  • Installation
  • Producer
  • Consumer
  • TLS
  • Schema
  • Reader
  • Options
  • MCP Server for AI Agents
  • License

About

English | 中文

This is a Apache Pulsar client library implemented in php Reference PulsarApi.proto And support Swoole coroutine

Features

• Support URL (pulsar:// 、 pulsar+ssl:// 、 http:// 、 https://)
• Multi topic consumers
• TLS connection
• Automatic reconnection (Only Consumer)
• Message batching
• Message Properties
• Compression with zstd, zlib
• Authentication with jwt, basic

Requirements

• PHP >=7.1
• ZLib Extension（If you want to use zlib compression）
• Zstd Extension（If you want to use zstd compression）
• Swoole Extension(If you want to use in swoole)
  • Use in the swoole only requires that the SWOOLE_HOOK_SOCKETS、SWOOLE_HOOK_STREAM_FUNCTION or SWOOLE_HOOK_ALL

Tips

If the following error occurs, please install extension gmp or bcmath

Negative integers are only supported with GMP or BC (64bit) intextensions.

Installation

composer require ikilobyte/pulsar-client-php

Producer

<?php

use Pulsar\Authentication\Basic;
use Pulsar\Authentication\Jwt;
use Pulsar\Compression\Compression;
use Pulsar\Producer;
use Pulsar\ProducerOptions;
use Pulsar\MessageOptions;

require_once __DIR__ . '/vendor/autoload.php';

$options = new ProducerOptions();

// If permission authentication is available
// use JWT authentication
$options->setAuthentication(new Jwt('token')); 

// use Basic authentication
//$options->setAuthentication(new Basic('user','password'));

$options->setConnectTimeout(3);
$options->setTopic('persistent://public/default/demo');
$options->setCompression(Compression::ZLIB);
$producer = new Producer('pulsar://localhost:6650', $options);
// or use pulsar proxy address
//$producer = new Producer('http://localhost:8080', $options);

$producer->connect();

for ($i = 0; $i < 10; $i++) {
    $messageID = $producer->send(sprintf('hello %d',$i));
    
    $messageID = $producer->send(sprintf('hello properties %d',$i),[
        MessageOptions::PROPERTIES => [
           'key' => 'value',
           'ms'  => microtime(true),
        ],
    ]);
    echo 'messageID ' . $messageID . "\n";
}

// Sending delayed messages
for ($i = 0; $i < 10; $i++) {
    $producer->send(sprintf('hello-delay %d',$i),[
        MessageOptions::DELAY_SECONDS => $i * 5, // Seconds
    ]);
}

// Send Batch message 
// The underlying protocol will automatically package these messages into a message and send it to pulsar
$messages = [];
for ($i = 0;$i < 10;$i++) {
    $messages[] = json_encode([
          'id'    => $i,
          'now'   => date('Y-m-d H:i:s')
    ]);
}

$messageID = $producer->send($messages);
echo "batch message id ${messageID}\n";

// close
$producer->close();

Keepalive Connection (Recommended)

• require Swoole extension
• If it is a resident memory application, it is recommended to open it.
• Will keep connected, no need to repeatedly establish a connection
• Calling the close method closes the connection
• Please see example

$options->setKeepalive(true);

Message deduplication

• Message de-duplication is a feature provided by pulsar and is based on the producer name and sequence number ID
• The name of the same producer needs to be fixed and unique, generally distinguished by business latitude, and the sequence number ID of each message is unique and self-incrementing.
• Reference Pulsar Docs

$options = new ProducerOptions();
$options->setProducerName('name');

$producer = new Producer('pulsar://localhost:6650', $options);
$producer->send('body',[
    \Pulsar\MessageOptions::SEQUENCE_ID => 123456,
]);

Consumer

<?php

use Pulsar\Authentication\Jwt;
use Pulsar\Authentication\Basic;
use Pulsar\Consumer;
use Pulsar\ConsumerOptions;
use Pulsar\SubscriptionType;
use Pulsar\Proto\CommandSubscribe\InitialPosition;

require_once __DIR__ . '/vendor/autoload.php';

$options = new ConsumerOptions();

// If permission authentication is available
// use JWT authentication
$options->setAuthentication(new Jwt('token'));

// use Basic authentication
//$options->setAuthentication(new Basic('user','password'));

$options->setConnectTimeout(3);
$options->setTopic('persistent://public/default/demo');
$options->setSubscription('logic');
$options->setSubscriptionType(SubscriptionType::Shared);

// Initial position at which to set cursor when subscribing to a topic at first time.	
// default use InitialPosition::Latest()
// $options->setSubscriptionInitialPosition(InitialPosition::Earliest());


// Configure how many seconds Nack's messages are redelivered, the default is 1 minute
$options->setNackRedeliveryDelay(20);

$consumer = new Consumer('pulsar://localhost:6650', $options);
// or use pulsar proxy address
//$consumer = new Consumer('http://localhost:8080', $options);

$consumer->connect();

while (true) {
    $message = $consumer->receive();
    
    // get properties
    var_export($message->getProperties());
    
    echo sprintf('Got message 【%s】messageID[%s] topic[%s] publishTime[%s] redeliveryCount[%d]',
        $message->getPayload(),
        $message->getMessageId(),
        $message->getTopic(),
        $message->getPublishTime(),
        $message->getRedeliveryCount()
    ) . "\n";

    // ... 
    
    // Remember to confirm that the message is complete after processing
    $consumer->ack($message);
    
    // When processing fails, you can also execute the Nack
    // The message will be re-delivered after the specified time
    // $consumer->nack($message);
}

$consumer->close();

Receive Batch Message

• Only when the producer sends the message in bulk can the batch message be received.

$messages = $consumer->batchReceive();
foreach ($messages as $message) {
    // ...
    
    // Ack
    $consumer->ack($message);
}

Subscribe to multiple topics

$options->setTopics([
    'persistent://public/default/demo-1',
    'persistent://public/default/demo-2',
    'persistent://public/default/demo-3',
    //....
]);

Dead letter topic

// Assuming that the subject matter is： <topicname>-<subscriptionname>-DLQ
$options->setDeadLetterPolicy(6);

// Custom topic name
$options->setDeadLetterPolicy(6,'persistent://public/default/demo-dead');

// Custom subscription name
$options->setDeadLetterPolicy(6,'persistent://public/default/demo-dead','sub-name');

Reconnect（Only Support Consumer）

// start reconnect
$options->setReconnectPolicy(true);

// Reconnect interval(seconds)
$options->setReconnectPolicy(true,3);

// Maximum number of reconnections
$options->setReconnectPolicy(true,3,100);

Not loop Receive And Smooth exit

$running = true;

// kill -15 $PID  
pcntl_signal(SIGTERM,function() use (&$running){
    $running = false;
});

while ($running) {
    try {
        $message = $consumer->receive(false);
        
        // ...
    } catch (\Pulsar\Exception\MessageNotFound $e) {
        if ($e ->getCode() != \Pulsar\Exception\MessageNotFound::Ignore) {
            die($e->getMessage());
        }
        echo "Message Not Found\n";
        continue;
    } catch (Throwable $e) {
        echo $e->getMessage() . "\n";
        throw $e;
    } finally {
        pcntl_signal_dispatch();
    }
}

TLS

• Refer to the official documentation for certificate configuration

• Example

$tls = new \Pulsar\TLSOptions('./cert.pem','./cert.key.pem');

// Establishing a TLS connection without a certificate
//$tls = new \Pulsar\TLSOptions('','');

// CA Cert
$tls->setTrustCertsFilePath('./ca.cert.pem');

// optional
$tls->setAllowInsecureConnection(false);
$tls->setValidateHostname(true);
$options->setTLS($tls);

$consumer = new \Pulsar\Consumer('pulsar+ssl://localhost:6651',$options);
//$producer = new \Pulsar\Producer('pulsar+ssl://localhost:6651',$options);


// or https
$consumer = new \Pulsar\Consumer('https://localhost:8081',$options);
//$producer = new \Pulsar\Producer('https://localhost:8081',$options);

Schema

• Currently only supports INT8、INT16、INT32、INT64、DOUBLE、STRING、JSON，The following code uses JSON Schema as an example

• https://pulsar.apache.org/docs/2.11.x/schema-overview/

• https://avro.apache.org/docs/1.11.1/specification/

• model.php

<?php

class Person
{
    public $id;
    public $name;
    public $age;
    // ...
}

• Producer Statement Schema

<?php
$define = '{"type":"record","name":"Person","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"}]}';
$schema = new \Pulsar\Schema\SchemaJson($define, [
    'key' => 'value',
]);

// ... some code
$producerOptions->setSchema($schema);
$producer = new \Pulsar\Producer('xx',$options);
$producer->connect();

$person = new Person();
$person->id = 1;
$person->name = 'Tony';
$person->age = 18;

// directly send Person Object No need to manually convert to json string
$id = $producer->send($person);

• Consumer Statement Schema

<?php
$define = '{"type":"record","name":"Person","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"}]}';

$schema = new \Pulsar\Schema\SchemaJson($define, [
   'key' => 'value',
]);

// ... some code
$consumerOptions->setSchema($schema);
$consumer = new \Pulsar\Consumer('pulsar://xxx',$consumerOptions);
$consumer->connect();

while (true) {
   $message = $consumer->receive();
   $person = new Person();
   $message->getSchemaValue($person);
   echo sprintf(
       'payload %s id %d name %s age %d',
       $message->getPayload(),
       $person->id,
       $person->name,
       $person->age
   ) . "\n";
   
   // .. some code
}

Reader

<?php
use Pulsar\Message;
use Pulsar\Reader;
use Pulsar\ReaderOptions;

require_once __DIR__ . '/../vendor/autoload.php';


$options = new ReaderOptions();

// If permission authentication is available
// Only JWT authentication is currently supported
// $options->setAuthentication(new Jwt('token'));

$options->setConnectTimeout(3);
$options->setTopic('persistent://public/default/demo'); // support partition topic

// Read the latest message
$options->setStartMessageID(Message::latestMessageIdData());

// From the earliest message
// $options->setStartMessageID(Message::earliestMessageIdData());

// Start reading from a message
// $options->setStartMessageID(Message::deserialize('621:103:0'));

$reader = new Reader('pulsar://localhost:6650', $options);
$reader->connect();

while (true) {
    $message = $reader->next();
    echo sprintf('Got message 【%s】messageID[%s]  topic[%s] publishTime[%s]',
            $message->getPayload(),
            $message->getMessageId(),
            $message->getTopic(),
            $message->getPublishTime()
        ) . "\n";

}

$reader->close();

Options

• ProducerOptions
  • setTopic()
  • setAuthentication()
  • setConnectTimeout()
  • setProducerName()
  • setCompression()
  • setSchema()
  • setKeepalive()
• ConsumerOptions
  • setTopic()
  • setTopics()
  • setAuthentication()
  • setConnectTimeout()
  • setConsumerName()
  • setSubscription()
  • setSubscriptionType()
  • setNackRedeliveryDelay()
  • setReceiveQueueSize()
  • setDeadLetterPolicy()
  • setSubscriptionInitialPosition()
  • setReconnectPolicy()
  • setSchema()
• ReaderOptions
  • setTopic()
  • setAuthentication()
  • setConnectTimeout()
  • setReaderName()
  • setStartMessageID()
  • setReceiveQueueSize()
• MessageOptions
  • DELAY_SECONDS
  • SEQUENCE_ID
  • PROPERTIES
• TLSOption (v1.3.0)
  • __construct(string $certFilePath, string $keyFilePath)
  • setTrustCertsFilePath()
  • setValidateHostname()
  • setAllowInsecureConnection()

MessageNotFound ErrCode （v1.2.1）

• MessageNotFound::Ignore
• MessageNotFound::CommandParseFail

MCP Server for AI Agents

This library includes an MCP (Model Context Protocol) stdio server that lets AI agents (Cursor, Claude Desktop, etc.) produce, consume, and peek at Pulsar messages. No extra dependencies required.

Tools: pulsar_publish · pulsar_consume · pulsar_peek

Quick setup — add to ~/.cursor/mcp.json (Cursor) or claude_desktop_config.json (Claude Desktop):

{
  "mcpServers": {
    "pulsar": {
      "command": "php",
      "args": ["/path/to/pulsar-client-php/examples/mcp-server.php"],
      "env": {
        "PULSAR_BROKER_URL": "pulsar://localhost:6650",
        "PULSAR_TOKEN": ""
      }
    }
  }
}

See examples/mcp.md for full tool reference and testing instructions.

License

MIT LICENSE