Ipc Broadcaster

在 Hyperf Server Worker 和用户進程之間廣播可序列化消息。

安裝

shell

composer require friendsofhyperf/ipc-broadcaster

該包會自動註冊 ConfigProvider，將 BroadcasterInterface 綁定到 AllProcessesBroadcaster，並註冊 Server Worker 接收消息所需的監聽器。該組件沒有需要發佈的配置文件。

該包的 composer.json 僅聲明瞭 hyperf/event ~3.2.0，且沒有聲明可選依賴。請在提供廣播器所需 Server、Process、容器和 DI 運行時類的 Hyperf Server 應用中使用。

廣播消息

broadcast() 函數接受 IpcMessageInterface 實例或閉包。默認通過 AllProcessesBroadcaster 將消息發送到其他所有 Server Worker 和所有已註冊的協程用户進程。

類消息

繼承 IpcMessage 並實現 handle()。IpcMessage 還通過 InteractsWithFromWorkerId 提供 getFromWorkerId() 和 setFromWorkerId()。Server Worker 收到消息時，getFromWorkerId() 包含發送方的 Worker ID。

php

<?php

namespace App\Broadcasting;

use FriendsOfHyperf\IpcBroadcaster\IpcMessage;

use function FriendsOfHyperf\IpcBroadcaster\broadcast;

class FooMessage extends IpcMessage
{
    public function __construct(private string $foo)
    {
    }

    public function handle(): void
    {
        echo $this->foo;
    }
}

broadcast(new FooMessage('bar'));

消息對象及其屬性必須可序列化，才能跨進程傳輸。

閉包消息

閉包在廣播前會包裝為 ClosureIpcMessage 並進行序列化，因此閉包捕獲的值也必須可序列化。在常規 Hyperf DI 環境中，有類型聲明的閉包參數可從容器解析。

php

use function FriendsOfHyperf\IpcBroadcaster\broadcast;

broadcast(function () {
    echo 'Hello world';
});

選擇目標

注入 BroadcasterInterface 可使用默認的全進程廣播行為，也可以構造特定廣播器來選擇目標：

php

use FriendsOfHyperf\IpcBroadcaster\ServerBroadcaster;
use FriendsOfHyperf\IpcBroadcaster\UserProcessesBroadcaster;

$serverBroadcaster = new ServerBroadcaster($container, id: 1);
$serverBroadcaster->broadcast($message);

$userProcessBroadcaster = new UserProcessesBroadcaster(name: 'reporting', id: 0);
$userProcessBroadcaster->broadcast($message);

ServerBroadcaster 接受容器和可選的 Worker ID；未傳 ID 時，會發送到當前 Worker 之外的所有 Server Worker。UserProcessesBroadcaster 接受可選的進程名稱和進程 ID；兩者均未傳時，會發送到 Hyperf ProcessCollector 收集的所有已註冊進程。

在用户進程中處理消息

組件會為 Server Worker 收到的消息自動調用 handle()。在用户進程中，Hyperf 會派發 Hyperf\Process\Event\PipeMessage，組件不會自動調用其中消息的 handle() 方法。需要讓用户進程執行消息時，請註冊監聽器：

php

use FriendsOfHyperf\IpcBroadcaster\Contract\IpcMessageInterface;
use Hyperf\Event\Contract\ListenerInterface;
use Hyperf\Process\Event\PipeMessage;

class UserProcessPipeMessageListener implements ListenerInterface
{
    public function listen(): array
    {
        return [PipeMessage::class];
    }

    public function process(object $event): void
    {
        if ($event instanceof PipeMessage && $event->data instanceof IpcMessageInterface) {
            $event->data->handle();
        }
    }
}

在當前 Worker 中運行

廣播默認不會在當前 Server Worker 中執行。給消息添加 RunsInCurrentWorker 後，會先在當前 Worker 中調用一次 handle()，再發送到所選目標：

php

use FriendsOfHyperf\IpcBroadcaster\IpcMessage;
use FriendsOfHyperf\IpcBroadcaster\Traits\RunsInCurrentWorker;

class RefreshMessage extends IpcMessage
{
    use RunsInCurrentWorker;

    public function handle(): void
    {
        // Refresh local state.
    }
}

協程服務器限制

觸發 MainCoroutineServerStart 後，跨進程廣播會被禁用，廣播器調用不會發送消息便直接返回。使用 RunsInCurrentWorker 的消息仍會在廣播器返回前在本地執行一次。

Ipc Broadcaster ​

安裝 ​

廣播消息 ​

類消息 ​

閉包消息 ​

選擇目標 ​

在用户進程中處理消息 ​

在當前 Worker 中運行 ​

協程服務器限制 ​