ansible 一些基本使用方法的记录

使用 python 和 grpc库,开发rpc服务

grpc 简介

grpc 是一个采用Protobuf来做数据的序列化与反序列化,用 http2.0 作为通信协议的rpc框架

HTTP2.0

HTTP2.0 相比 HTTP1.1 有很大不同,HTTP1.1 还是基于文本协议的问答有序模式，但是 HTTP2.0 是基于二进制协议的乱序模式 (Duplexing).这意味同一个连接通道上多个请求并行时，服务器处理快的可以先返回而不用因为等待其它请求的响应而排队。HTTP2.0 对请求头的 key/value 做了字典处理，常用的 key/value 无需重复传送,而是通过引用内部字典的索引节省了请求头传输的流量.

Protobuf 通讯协议

Protobuf 协议是 Google 开源的二进制 RPC 通讯协议，它可能是互联网开源项目中使用最为广泛的 RPC 协议。

grpc 简单模式编程示例

编写协议文件 ping.proto

在目录下创建并编写文件 ping.proto 文件

syntax = "proto3";

package ping;

// ping service
service PingCalculator {
    rpc Calc(PingRequest) returns (PingResponse) {}
}

// ping input
message PingRequest {
   string n = 1;
}

// ping output
message PingResponse {
    string n = 1;
}

使用grpc_tools工具生成消息序列化类,服务端类和客户端类

• -I.: 指定协议文件的查找目录,这里为当前目录
• –python_out=. 指定消息序列化类文件的输出路径
• –grpc_python_out=. 指定服务端客户端类文件的输出路径

python3 -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ping.proto

根据服务器类，编写服务器具体逻辑实现

创建文件 server.py

#!/usr/bin/ python3
# -*- coding: utf-8 -*-
import grpc
import time
import ping_pb2
import ping_pb2_grpc
from concurrent import futures


class PingCalculatorServicer(ping_pb2_grpc.PingCalculatorServicer):
    def Calc(self, request, ctx):
        """在这里实现业务逻辑"""
        time.sleep(0.5)
        return ping_pb2.PingResponse(n=str(request.n))


def main():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))  # 多线程服务器
    servicer = PingCalculatorServicer()   # 实例化 ping 服务类
    ping_pb2_grpc.add_PingCalculatorServicer_to_server(servicer=servicer, server=server)  # 注册本地服务
    server.add_insecure_port('127.0.0.1:8080')  # 监听端口
    server.start()  # 开始接收请求
    try:
        time.sleep(1000)
    except KeyboardInterrupt:
        server.stop(0)  # 使用 ctrl+c 可以退出服务


if __name__ == '__main__':
    main()

使用客户端 Stub，编写客户端交互代码

单线程客户端

创建 client.py 文件

#!/usr/bin/ python3
# -*- coding: utf-8 -*-
"""单线程客户端"""
import grpc
import ping_pb2
import ping_pb2_grpc


def main():
    channel = grpc.insecure_channel('localhost:8080')
    # 使用 stub
    client = ping_pb2_grpc.PingCalculatorStub(channel)
    # 调用吧
    for i in range(10):
        print(i, client.Calc(ping_pb2.PingRequest(n=str(i))).n)


if __name__ == '__main__':
    main()

多线程客户端

创建 multithread_client.py 文件

#!/usr/bin/ python3
# -*- coding: utf-8 -*-
"""并行客户端"""
import grpc
import ping_pb2
import ping_pb2_grpc
from concurrent import futures


def ping(client, n):
    return client.Calc(ping_pb2.PingRequest(n=str(n))).n


def main():
    channel = grpc.insecure_channel("127.0.0.1:8080")
    client = ping_pb2_grpc.PingCalculatorStub(channel=channel)
    pool = futures.ThreadPoolExecutor(max_workers=4)  # 客户端使用线程池执行
    results = []
    for i in range(10):
        results.append((i, pool.submit(ping, client, str(i))))

    # 等待所有任务执行完毕
    pool.shutdown()

    for i, future in results:
        print(i, future.result())


if __name__ == '__main__':
    main()

分别运行服务器和客户端，观察输出结果

对比单线程和多线程客户端

从结果可以看出,多线程客户端可以很好的提升效率

grpc 流模式编程示例

相较于普通模式,流模式(Streaming) 可以理解为 gRPC 的异步调用

编写协议文件

创建 ping.proto 文件

syntax = "proto3";

package ping;

// ping 服务,注意在输入输出类型上增加了 stream 关键字
service PingCalculator {
    // ping method
    rpc Calc(stream PingRequest) returns (stream PingRequest) {}
}

// ping 请求
message PingRequest {
    int32 n = 1;
}

// ping 响应, 注意要与请求协议中的字段一一对应
message PingResponse {
    int32 n = 1;
}

使用grpc_tools工具生成消息序列化类,服务端类和客户端类

• -I.: 指定协议文件的查找目录,这里为当前目录
• –python_out=. 指定消息序列化类文件的输出路径
• –grpc_python_out=. 指定服务端客户端类文件的输出路径

python3 -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ping.proto

根据服务器类，编写服务器具体逻辑实现

创建 server.py 文件

#!/usr/bin/ python3
# -*- coding: utf-8 -*-
import grpc
import time
import random
from concurrent import futures
import ping_pb2
import ping_pb2_grpc


class PingCalculatorServer(ping_pb2_grpc.PingCalculatorServicer):
    def Calc(self, request_iterator, ctx):
        # 请求是一个迭代器,响应是一个生成器
        # request 是一个迭代器参数,对应的是一个 stream 请求
        for request in request_iterator:
            # 50% 的概率会有响应, 为了演示请求和响应不是一对一的效果
            if random.randint(0, 1) == 1:
                continue
            yield ping_pb2.PingResponse(n=request.n)   # 响应是一个生成器, 一个响应对应一个请求


def main():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    servicer = PingCalculatorServer()
    ping_pb2_grpc.add_PingCalculatorServicer_to_server(servicer, server)
    server.add_insecure_port("127.0.0.1:8083")
    server.start()

    try:
        time.sleep(1000)
    except KeyboardInterrupt:
        server.stop(0)


if __name__ == '__main__':
    main()

使用客户端 Stub，编写客户端交互代码

#!/usr/bin/ python3
# -*- coding: utf-8 -*-
import grpc
import ping_pb2
import ping_pb2_grpc


def generate_request():
    for i in range(1000):
        yield ping_pb2.PingRequest(n=i)


def main():
    channel = grpc.insecure_channel("127.0.0.1:8083")
    client = ping_pb2_grpc.PingCalculatorStub(channel)
    response_iterator = client.Calc(generate_request())
    # 请求是一个生成器,响应是一个迭代器
    for response in response_iterator:
        print("ping", response.n)


if __name__ == '__main__':
    main()

分别运行服务器和客户端，观察输出结果

通过实验可以看出, 有很多请求被跳过了.