README

Laravel的Pub-Sub抽象层。 由Superbalist/laravel-pubsub派生

此包是一个包装器，将php-pubsub引入Laravel。

为了支持Laravel 4，请使用https://github.com/Superbalist/laravel4-pubsub

请注意，Laravel 5.3仅支持到版本2.0.2。

2.0.3+将支持后续的Laravel 5.4及以上版本。

以下适配器受支持

• 本地
• /dev/null
• Redis
• Kafka（以下将提供单独的安装说明）
• Google Cloud
• HTTP

安装

composer require entanet/laravel-pubsub

在app.php中注册服务提供者

'providers' => [
    // ...
    Superbalist\LaravelPubSub\PubSubServiceProvider::class,
]

在app.php中注册外观

'aliases' => [
    // ...
    'PubSub' => Superbalist\LaravelPubSub\PubSubFacade::class,
]

该包有一个默认配置，它使用以下环境变量。

PUBSUB_CONNECTION=redis

REDIS_HOST=localhost
REDIS_PASSWORD=null
REDIS_PORT=6379

KAFKA_BROKERS=localhost
KAFKA_SECURITY_PROTOCOL=kafka-security-protocol
KAFKA_SASL_USERNAME=kafka-sasl-username
KAFKA_SASL_PASSWORD=kafks-sasl-password

GOOGLE_CLOUD_PROJECT_ID=your-project-id-here
GOOGLE_CLOUD_KEY_FILE=path/to/your/gcloud-key.json

HTTP_PUBSUB_URI=null
HTTP_PUBSUB_SUBSCRIBE_CONNECTION=redis

如果KAFKA_SECURITY_PROTOCOL设置为"SASL_SSL"或"SASL_PLAINTEXT"，则将使用存储在KAFKA_SASL_USERNAME & KAFKA_SASL_PASSWORD中的凭据与kafka服务器/集群进行身份验证。

要自定义配置文件，请使用Artisan发布包配置。

php artisan vendor:publish --provider="Superbalist\LaravelPubSub\PubSubServiceProvider"

然后您可以在app/config/pubsub.php中编辑生成的配置。

Kafka适配器安装

请注意，尽管该包捆绑了对php-pubsub-kafka适配器的支持，但适配器默认不包含在内。

这是因为KafkaPubSubAdapter依赖于外部依赖librdkafka c库和php-rdkafka PECL扩展。

如果您计划使用此适配器，您需要按照以下安装说明安装这些依赖项。

然后您可以使用以下方式包含适配器

composer require superbalist/php-pubsub-kafka

用法

// get the pub-sub manager
$pubsub = app('pubsub');

// note: function calls on the manager are proxied through to the default connection
// eg: you can do this on the manager OR a connection
$pubsub->publish('channel_name', 'message');

// get the default connection
$pubsub = app('pubsub.connection');
// or
$pubsub = app(\Superbalist\PubSub\PubSubAdapterInterface::class);

// get a specific connection
$pubsub = app('pubsub')->connection('redis');

// publish a message
// the message can be a string, array, bool, object - anything which can be json encoded
$pubsub->publish('channel_name', 'this is where your message goes');
$pubsub->publish('channel_name', ['key' => 'value']);
$pubsub->publish('channel_name', true);

// publish multiple messages
$messages = [
    'message 1',
    'message 2',
];
$pubsub->publishBatch('channel_name', $messages);

// subscribe to a channel
$pubsub->subscribe('channel_name', function ($message) {
    var_dump($message);
});

// all the above commands can also be done using the facade
PubSub::connection('kafka')->publish('channel_name', 'Hello World!');

PubSub::connection('kafka')->subscribe('channel_name', function ($message) {
    var_dump($message);
});

创建一个订阅者

该包包括一个辅助命令php artisan make:subscriber MyExampleSubscriber来生成新的订阅者命令类。

许多pub-sub适配器将包含阻塞的subscribe()调用，因此这些命令最好作为以supervisor进程运行的守护进程来运行。

此生成命令将创建文件app/Console/Commands/MyExampleSubscriber.php，其中将包含

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Superbalist\PubSub\PubSubAdapterInterface;

class MyExampleSubscriber extends Command
{
    /**
     * The name and signature of the subscriber command.
     *
     * @var string
     */
    protected $signature = 'subscriber:name';

    /**
     * The subscriber description.
     *
     * @var string
     */
    protected $description = 'PubSub subscriber for ________';

    /**
     * @var PubSubAdapterInterface
     */
    protected $pubsub;

    /**
     * Create a new command instance.
     *
     * @param PubSubAdapterInterface $pubsub
     */
    public function __construct(PubSubAdapterInterface $pubsub)
    {
        parent::__construct();

        $this->pubsub = $pubsub;
    }

    /**
     * Execute the console command.
     */
    public function handle()
    {
        $this->pubsub->subscribe('channel_name', function ($message) {

        });
    }
}

Kafka订阅者

对于使用php-pubsub-kafka适配器的订阅者，您可能希望为每个订阅者更改consumer_group_id。

为此，您需要使用PubSubConnectionFactory为每个订阅者创建一个新的连接。这是因为一旦创建连接，就无法更改consumer_group_id。

以下是如何做到这一点的示例

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Superbalist\LaravelPubSub\PubSubConnectionFactory;
use Superbalist\PubSub\PubSubAdapterInterface;

class MyExampleKafkaSubscriber extends Command
{
    /**
     * The name and signature of the subscriber command.
     *
     * @var string
     */
    protected $signature = 'subscriber:name';

    /**
     * The subscriber description.
     *
     * @var string
     */
    protected $description = 'PubSub subscriber for ________';

    /**
     * @var PubSubAdapterInterface
     */
    protected $pubsub;

    /**
     * Create a new command instance.
     *
     * @param PubSubConnectionFactory $factory
     */
    public function __construct(PubSubConnectionFactory $factory)
    {
        parent::__construct();

        $config = config('pubsub.connections.kafka');
        $config['consumer_group_id'] = self::class;
        $this->pubsub = $factory->make('kafka', $config);
    }

    /**
     * Execute the console command.
     */
    public function handle()
    {
        $this->pubsub->subscribe('channel_name', function ($message) {

        });
    }
}

添加自定义驱动程序

请参阅php-pubsub文档中的编写适配器部分。

要包含您的自定义驱动程序，您可以调用extend()函数。

$manager = app('pubsub');
$manager->extend('custom_connection_name', function ($config) {
    // your callable must return an instance of the PubSubAdapterInterface
    return new MyCustomPubSubDriver($config);
});

// get an instance of your custom connection
$pubsub = $manager->connection('custom_connection_name');