問題描述得很詳細(xì)，非常贊，一看就有回答的欲望。
workerman做這樣的業(yè)務(wù)非常擅長(zhǎng)，以下是示例代碼

安裝 workerman/http-client

為了異步執(zhí)行http回調(diào)，要裝下 workerman/http-client

composer require workerman/http-client

服務(wù)端代碼

以下是服務(wù)端代碼示例 start.php

<?php

use Workerman\Connection\TcpConnection;
use Workerman\Http\Client;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
use Workerman\Connection\AsyncTcpConnection;
require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker('http://0.0.0.0:2345');

// 保存訂閱主題->回調(diào)url的數(shù)組
global $subjects;
$subjects = [];

// 進(jìn)程啟動(dòng)時(shí)建立一個(gè)到A服務(wù)的連接
$worker->onWorkerStart = function() {
    $con = new AsyncTcpConnection('ws://echo.websocket.org:443');
    $con->transport = 'ssl'; // tcp協(xié)議不需要設(shè)置
    $con->onWebSocketConnect = function(AsyncTcpConnection $con, ) {
        $con->send('hello');
    };
    // 收到A服務(wù)的訂閱數(shù)據(jù)后轉(zhuǎn)發(fā)給訂閱者
    $con->onMessage = function(AsyncTcpConnection $con, $subject) {
        global $subjects;
        foreach ($subjects[$subject] ?? [] as $callbackUrl) {
            $httpClient = new Client();
            $httpClient->post($callbackUrl, ['subject' => $subject]);
            echo "callback $callbackUrl\n";
        }
    };
    // 為了模擬A主動(dòng)推送訂閱消息，這里定時(shí)向A服務(wù)發(fā)送數(shù)據(jù)，A會(huì)返回hello
    \Workerman\Timer::add(10, function () use ($con) {
        $con->send('hello');
    });
    $con->connect();
};

$worker->onMessage = function (TcpConnection $httpConnection, Request $request) {
    $callbackUrl = $request->get('callback');
    // 訂閱主題
    $subject = $request->get('subject');
    // 沒有回調(diào)url或者訂閱主題認(rèn)為是同步指令
    if (!$callbackUrl || !$subject) {
        $con = new AsyncTcpConnection('ws://echo.websocket.org:443');
        $con->transport = 'ssl'; // tcp協(xié)議不需要設(shè)置
        $con->onWebSocketConnect = function (AsyncTcpConnection $con) {
            $con->send('hello');
        };
        $con->onMessage = function (AsyncTcpConnection $con, $data) use ($httpConnection) {
            $httpConnection->send($data);
            $con->close();
        };
        $con->connect();
        return;
    }
    // 記錄主題和回調(diào)之間的關(guān)系
    global $subjects;
    if (!isset($subjects[$subject])) {
        $subjects[$subject] = [];
    }
    if (in_array($callbackUrl, $subjects[$subject])) {
        $httpConnection->send('already subscribed');
        return;
    }
    $subjects[$subject][] = $callbackUrl;
    $httpConnection->send('subscribed');
};

Worker::runAll();

啟動(dòng)

php start.php start

測(cè)試

同步請(qǐng)求url類似 http://127.0.0.1:2345
訂閱測(cè)試url類似 http://127.0.0.1:2345/?subject=hello&callback=<url地址>

說明

為了方便測(cè)試，上面代碼用的wss協(xié)議測(cè)試，你需要改成自己的協(xié)議

多進(jìn)程時(shí)要加一句 $worker->reusePort = true;

$worker = new Worker('http://0.0.0.0:2345');
$worker->reusePort = true;

學(xué)習(xí)了。