客户端:

       $fibonacci_rpc = new FibonacciRpcClient();
        $response = $fibonacci_rpc->call(json_encode([
            'type' => 'test',
        ]));
        echo ' [.] Got ', $response, "\n";

class FibonacciRpcClient
{
    private $connection;
    private $channel;
    private $callback_queue;
    private $response;
    private $corr_id;

    public function __construct() {
        $this->connection = ...getConnection();
        $this->channel = $this->connection->channel();
        [$this->callback_queue, ,] = $this->channel->queue_declare();
        $this->channel->basic_consume(
            $this->callback_queue,
            '',
            false,
            true,
            false,
            false,
            [
                $this,
                'onResponse',
            ]
        );
    }

    public function onResponse($rep) {
        if ($rep->get('correlation_id') == $this->corr_id) {
            $this->response = $rep->body;
        }
    }

    public function call($n) {
        $this->response = null;
        $this->corr_id = uniqid();

        $msg = new \PhpAmqpLib\Message\AMQPMessage(
            (string)$n,
            [
                'correlation_id' => $this->corr_id,
                'reply_to' => $this->callback_queue,
            ]
        );
        $this->channel->basic_publish($msg, '', 'rpc_queue');
        while ($this->response === null) {
            $this->channel->wait();
        }
        return $this->response;
    }
}

服务端:

      $connection = ...getConnection();
        $channel = $connection->channel();

        $channel->queue_declare('rpc_queue', false, false, false, false);

        echo " [x] Awaiting RPC requests\n";
        $callback = function (\PhpAmqpLib\Message\AMQPMessage $req) {
            $body = $req->body;
            // echo ' [.] ', $body, "\n";
            $result = $this->process($body);
            $msg = new AMQPMessage(
                (string)($result),
                ['correlation_id' => $req->get('correlation_id')]
            );

            $req->delivery_info['channel']->basic_publish(
                $msg,
                '',
                $req->get('reply_to')
            );
            $req->ack();
        };

        $channel->basic_qos(null, 1, null);
        $channel->basic_consume('rpc_queue', '', false, false, false, false, $callback);

        while ($channel->is_open()) {
            try {
                $channel->wait(null, false, rand(33, 44));
            } catch (\PhpAmqpLib\Exception\AMQPTimeoutException $e) {
            } catch (\Throwable $e) {
                \nomodels\DingtalkRobot\DingtalkRobot::getInstance()->sendMessage($e->getMessage());
            }
            if ($this->getMtime() != $this->mtime) {
                break;
            }
        }

        $channel->close();
        $connection->close();