a2s

a2s服务的python工具包，更容易大规模应用

启动服务

# 使用前请安装工具包
# pip install a2s
# 下载地址：http://192.168.3.10:8080/simple/a2s/

from a2s.a2s_server import watch, simple_params


# 方式一 使用简单方式启动   .py文件命令行参数   python server.py --h
@simple_params
@watch
def main(data: bytes, *args, **kwargs) -> bytes:
    '''

    :param data: 固定格式,调用端发送的数据data
    :param args: 固定格式
    :param kwargs: 固定格式
    :return: 返回调用端的 data
    '''
    # 内部处理程序
    data = data.decode("utf-8")
    data += "服务端处理过了数据"
    return data.encode("utf-8")


main()


# 方式二 自定义启动方式
@watch
def main(data: bytes, *args, **kwargs) -> bytes:
    '''

    :param data: 固定格式,调用端发送的数据data
    :param args: 固定格式
    :param kwargs: 固定格式
    :return:返回调用端的 data
    '''
    # 内部处理程序
    data = data.decode("utf-8")
    data += "服务端处理过了数据"
    return data.encode("utf-8")


main(nats_host="192.168.3.240", nats_port=19090, subject="test", queue="main")

方式一 更简单方式启动

方法二 灵活性更高

推荐使用 方法一

调用服务

a2s_client # 简化a2s客户端调用

from a2s.a2s_client import a2s_execute

if __name__ == '__main__':
    # bytes_data 调用端的 data 记得编码
    result = a2s_execute(a2s_ip="192.168.3.240:19094", topic="test", timeout=10, bytes_data=b"hello")
    # result 服务端返回的 data 记得解码
    print(result.decode("utf-8"))

监控

from a2s.a2s_monitor import watch_monitor

service_count = watch_monitor('192.168.3.240:9090', 'goods_service')

工具

dict(字典) 序列化反序列化

grpc 序列化反序列化

import json
import base64


def grpc_serialize(resp):
    """
    grpc对象序列化
    :param resp:resp = Text2VectorReq(text=data)
    :return:
    """
    resp.SerializeToString()
    return resp


def grpc_deserialize(req, data):
    """
    grpc对象反序列化
    :param req: req = Text2VectorResp()
    :param data: req的序列化对象
    :return:
    """
    req.ParseFromString(data)
    return req


def json_serialize(dict_data):
    """
    json对象序列化
    :param resp: 字典
    :return:
    """
    # 将字典转换为 JSON 格式的字符串
    json_data = json.dumps(dict_data)

    # 将 JSON 字符串转换为字节流
    bytes_data = json_data.encode('utf-8')
    return bytes_data


def json_deserialize(bytes_data):
    """
    grpc对象反序列化
    :param bytes_data:
    :return:
    """
    # 将字节流转换为 JSON 字符串
    json_data = bytes_data.decode('utf-8')

    # 将 JSON 字符串转换为字典对象
    dict_data = json.loads(json_data)
    return dict_data

# 二进制文件传输，json序列化解决方案：
# 1. 将二进制文件转换为base64编码的字符串
# 2. 将base64编码的字符串转换为二进制文件
def binary_to_base64(binary_data):
    """
    二进制转化为base64字符串
    :param binary_data:
    :return:
    """
    base64_str = base64.b64encode(binary_data).decode('utf-8')  # 将bytes转换为base64字符串
    return base64_str


def base64_to_binary(base64_str):
    """
    base64字符串转化为二进制
    :param base64_str:
    :return:
    """
    binary_data = base64.b64decode(base64_str)  # 将base64字符串转换为bytes
    return binary_data