[实例篇] ThinkPHP5+RabbitMQ 环境下 RPC 运行实例

看完这篇文章，你将会知道：
a. RabbitMQ 环境下 RPC 实例执行流程；
b. ThinkPHP 框架下自定义 cli 命令

运行环境：Docker + ThinkPHP5 + RabbitMQ，环境搭建完毕之后 composer 安装 RabbitMQ 的 PHP 版本。

1.启动 docker, 启动 rabbitmq

2.rpc 客户端代码：

<?php

namespace app\index\controller;
use think\Controller;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use app\command\FibonacciRpcClient;

class Symbol extends Controller{
    public function rpc(){
        $n = input('n');
        $fibonacci_rpc = new FibonacciRpcClient();
        $response = $fibonacci_rpc->call($n);
        echo ' [.] Got ', $response, "\n";
    }
}

3.rpc 服务端代码：

<?php
namespace app\command;

use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;
use think\Request;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class rpcserver extends Command
{
    /**
     * 重写configure
     * {@inheritdoc}
     */
    protected function configure()
    {
        $this
            // 命令的名字（"think" 后面的部分）
            ->setName('rpcserver')
            // 配置一个参数 使用$input->getArgument('username')获取
            // ->addArgument('username')
            // 运行 "php think list" 时的简短描述
            ->setDescription('定时任务微服务.')
            // 运行命令时使用 "--help" 选项时的完整命令描述
            ->setHelp("定时任务微服务 无参数");
    }
    /**
     *  * 重写execute
     *  * {@inheritdoc}
     *  
     * @param Input $input
     * @param Output $output
     */
    protected function execute(Input $input, Output $output)
    {
        // 注意 new 后面的 '\' 一定要加！
        $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
        $channel = $connection->channel();

        $channel->queue_declare('rpc_queue', false, false, false, false);
        echo " [x] Awaiting RPC requests\n";

        $callback = function($req){
            $n = intval($req->body);
            echo ' [.] fib(', $n, ")\n";
            $msg = new \PhpAmqpLib\Message\AMQPMessage($this->fib($n), ['correlation_id' => $req->get('correlation_id')]);
            $req->delivery_info['channel']->basic_publish($msg, '', $req->get('reply_to'));
            $req->delivery_info['channel']->basic_ack($req->delivery_info['delivery_tag']);
        };

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

        while ($channel->is_consuming()) {
            $channel->wait();
        };

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

    public function fib($n)
    {
        sleep(10);
        if ($n == 0) {
            return 0;
        }
        if ($n == 1) {
            return 1;
        }

        return $this->fib($n - 1) + $this->fib($n - 2);
    }
}

4.启动服务端：注意命令调用时的当前路径

cli 模式下执行：

D:\wamp64\www\tp51>php think rpcserver

5.启动客户端：

浏览器中访问

http://localhost/tp51/public/index.php/index/symbol/rpc/n/2

6.结果描述：

服务端处于监听状态，等待新任务投递过来。当客户端访问后，即创建了新的任务。服务端接收到新的任务后开始执行，执行完毕将结果返回给客户端。

7.应用场景：

RPC 主要的应用场景是将耗时、费力的复杂性计算从业务程序中解耦，客户端通过队列通信将任务下发，服务端负责执行任务，完成回调。

8.后续：

[原理篇] ThinkPHP5+RabbitMQ 环境下 RPC 运行原理

9.说明：

a. 文中实例代码来自官网，了解更多信息可自行访问 RabbitMQ 官方网站。
b. 有关ThinkPHP command 代码来源于网络，侵删。