PHP code example of easyswoole / nsq

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

easyswoole / nsq example snippets

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Nsq;

= new Nsq(new Config(host: '127.0.0.1', port: 4150));
    try {
        $nsq->publish('topic.tests', 'hello');
        $nsq->multiPublish('topic.tests', ['first', 'second']);
        $nsq->deferredPublish('topic.tests', 'later', 1000); // 毫秒
    } finally {
        $nsq->close();
    }
});

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Message\Message;
use EasySwoole\Nsq\Nsq;

));
    try {
        $nsq->subscribe('topic.tests', 'tests.consuming');
        $nsq->consume(function (Message $message, Nsq $client): bool {
            echo $message->body;
            // $message->id、attempts、timestamp（Unix 纳秒）均为只读字段。
            // $client->touch($message);  // 延长消息处理时限
            // $client->stop();           // 当前回调完成并确认后退出
            return true;                 // true/null => FIN；false => REQ
        }, requeueDelayMilliseconds: 1000);
    } finally {
        $nsq->close();
    }
});

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Nsq;

umer = new Nsq(new Config(readTimeout: -1));
    try {
        $consumer->subscribe('topic.tests', 'manual.consuming');
        $consumer->ready(1);
        $message = $consumer->receive();

        echo $message->body;
        $consumer->finish($message);
        // 如果业务处理失败，改为 $consumer->requeue($message, 1000)。
        // 处理较长任务时可调用 touch($message) 延长服务端消息超时。
    } finally {
        $consumer->close();
    }
});

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Message\Message;
use EasySwoole\Nsq\Nsq;

 {
        Swoole\Coroutine::create(function () use ($worker): void {
            $consumer = new Nsq(new Config(readTimeout: -1));
            try {
                $consumer->subscribe('topic.tests', 'tests.consuming');
                $consumer->consume(function (Message $message) use ($worker): void {
                    echo "worker={$worker}: {$message->body}\n";
                });
            } finally {
                $consumer->close();
            }
        });
    }
});

$config = new EasySwoole\Nsq\Config(
    host: '127.0.0.1',
    port: 4150,
    connectTimeout: 3.0,
    readTimeout: 60.0,
    writeTimeout: 3.0,
    maxFrameSize: 16_777_216,
    identify: ['client_id' => 'worker-1', 'heartbeat_interval' => 30000],
    authSecret: 'secret',
    tls: new EasySwoole\Nsq\TlsConfig(
        caFile: '/etc/nsq/tls/ca.pem',
        serverName: 'nsqd.example.com',
    ),
);

$tls = new EasySwoole\Nsq\TlsConfig(
    caFile: '/etc/nsq/tls/ca.pem',
    clientCertificate: '/etc/nsq/tls/client.pem',
    clientPrivateKey: '/etc/nsq/tls/client-key.pem',
    passphrase: 'private-key-password', // 无口令私钥可省略
    serverName: 'nsqd.example.com',
);
$config = new EasySwoole\Nsq\Config(tls: $tls);

Swoole\Coroutine\run(function (): void {
    $lookup = new EasySwoole\Nsq\Lookup\Nsqlookupd([
        'http://127.0.0.1:4161',
        'http://127.0.0.2:4161',
    ]);
    $endpoints = $lookup->lookupHosts('topic.tests');
    // ['127.0.0.1:4150', '[::1]:4150', ...]
});

$lookup = new EasySwoole\Nsq\Lookup\Nsqlookupd(
    ['https://nsqlookupd.example.com:4162'],
    tls: new EasySwoole\Nsq\TlsConfig(caFile: '/etc/nsq/tls/ca.pem'),
);

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Lookup\Nsqlookupd;
use EasySwoole\Nsq\Nsq;

:4161']);
    foreach ($lookup->lookupHosts('topic.tests') as $endpoint) {
        Swoole\Coroutine::create(function () use ($endpoint): void {
            $address = parse_url('tcp://' . $endpoint);
            $consumer = new Nsq(new Config(
                host: trim($address['host'], '[]'),
                port: $address['port'],
                readTimeout: -1,
            ));
            try {
                $consumer->subscribe('topic.tests', 'tests.consuming');
                $consumer->consume(function ($message): void { echo $message->body; });
            } finally {
                $consumer->close();
            }
        });
    }
});

sh
php tests/mock-nsqd.php 4150
# 可选认证：MOCK_NSQ_AUTH_SECRET=secret php tests/mock-nsqd.php 4150