引言

之前已经离职写 C# 的同事今天突然来找我，问我python的问题：

聊天记录截图

一开始以为他改行写python了，原来是他们不再用 matlab ，想通过python的 scipy 库来实现一些算法。

其实都知道python的胶水特性，但真正调用的时候确实会有这样那样的问题。

那怎么办呢？

其实除了直接使用代码调用，更合适的方式其实就是把 python 程序变成一个 中间件，将任务丢给 python 程序，等 结果 计算完成后，再扔回去。

1、RPC是什么

我们将使用 RPC 来实现 中间件，它的全称是 “Remote Procedure Call Protocol”，意思是 远程过程调用。

具体是什么呢？它就像是 云函数，它能够让一个程序像调用本地函数一般，通过网络去调用另一个程序的函数。

虽说是通过网络的，那如果部署在同一台电脑上，那就相当于 一个进程调用另一个进程 般丝滑。

2、RPC使用

我们将使用谷歌的 grpc，它支持主流编程语言，且是跨平台的。

2.1 安装库

pip install grpcio grpcio-tools

2.2 定义服务

创建一个 Test.proto 文件，需要在这个文件中定义我们的公开服务。

Test.proto

syntax = "proto3";

message AddRequest {
    repeated float x = 1;
    int32 y = 2;
}

message AddResponse {
    int32 result = 1;
}

service Calculator {
    rpc Add (AddRequest) returns (AddResponse) {}
}

我们设定了一个 Add 的服务，里面参数是 x和y，其中 x对应的是一个浮点数的列表，y是一个整型，返回值也是一个 整型。

这里为了测试多个数据类型，使用了两种参数，只是为了演示设置的。

2.3 生成服务代码

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. Test.proto       

运行后，将生成两个文件：Test_pb2.py和Test_pb2_grpc.py。

.py文件

2.4 创建服务代码

现在我们创建一个 TestService.py 来实现服务。

TestService.py

# -*- coding: utf-8 -*-
from concurrent import futures
import grpc
import Test_pb2_grpc
import Test_pb2
import time


class TestService(Test_pb2_grpc.CalculatorServicer):
    def Add(self, x, y):
        print(x)
        print(y)

        return Test_pb2.AddResponse(result=100)


def run():
    # 设置接收最大为1g
    server_option = [('grpc.max_receive_message_length', 1 * 1024 * 1024 * 1024)]
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), options=server_option)
    Test_pb2_grpc.add_CalculatorServicer_to_server(TestService(), server)
    port = 8245
    server.add_insecure_port(f'[::]:{port}')
    server.start()

    print('服务启动')
    try:
        while True:
            print('进入循环')
            time.sleep(60)
    except KeyboardInterrupt:
        server.stop(0)


if __name__ == '__main__':
    run()

我们启动了一个端口为 8245 的服务，只打印了 x和y的值，返回了 100。

2.5 测试服务

我们编写一个 TestClient.py 来测试服务。

TestClient.py

import grpc
import Test_pb2_grpc
import Test_pb2

def test_run():
    conn = grpc.insecure_channel('localhost:8245')
    client = Test_pb2_grpc.CalculatorStub(channel=conn)

    request = Test_pb2.AddRequest(
        x=[1.1, 1.2, 1.3],
        y=5,
    )
    response = client.Add(request)
    print(response)


if __name__ == '__main__':
    test_run()

我们使用 python 来测试服务，当然如果你有其他语言的程序，可以使用相应语言程序的 grpc 测试服务。

2.6 查看输出

TestService.py 的输出

x: 1.1
x: 1.2
x: 1.3
y: 5

TestService.py 的输出

x: 1.1
x: 1.2
x: 1.3
y: 5

TestClient.pyd 的输出

result: 100

可以看到，服务获取到了客户端的参数，而客户端得到了服务端的结果。

结尾

现在你已经知道了如何通过 RPC 来实现不同语言与 python 的交互，它能够像使用本地函数一样，调用 python 的实现，可以说非常适合跨语言开发。