| Name | gaterpc JSON |
| Version |
0.1.10
JSON |
| download |
| home_page | https://github.com/firejoke/gate-rpc |
| Summary | A RPC software based on ZeroMQ with built-in Majordomo. |
| upload_time | 2024-02-21 08:32:19 |
| maintainer | |
| docs_url | None |
| author | Shi Fan |
| requires_python | >=3.9 |
| license | BSD 3-Clause License Copyright (c) 2024, shifan All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| keywords |
zero
zeromq
rpc
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
msgpack
pyzmq
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
gate-rpc
############
使用 ZeroMQ 和 asyncio 开发的“gate-rpc”。
- 使用msgpack序列化消息
- 支持在线程池和进程池里运行普通函数
- 当函数返回的是 Generator 或 AsyncGenerator 时，会换成流式传输，接收端通过遍历 StreamReply 实例即可获得
- 当函数返回的值过大时，会转换成 HugeData 对象，压缩返回值并分块流式传输，默认块大小通过 Settings 的 HUGE_DATA_SIZEOF 来设置
- 使用队列传输和记录日志，避免因为日志导致事件循环被阻塞
- 使用可以异步设置键值和异步获取键值的 BoundedDict 来简化超时等待获取
- 在每次实例化 Worker、Server、AMajordomo、Client 各类之前，通过修改 Settings 的属性来修改运行配置
********
安装
********
可以直接使用pip安装，或直接下载源码随意放置。
::
pip install gaterpc
********
配置
********
在实例化 Worker、Service、AMajordomo、Client 各类之前，需要运行 Settings.setup 函数来配置全局配置 [#f1]_ ，
特殊返回值的序列化通过 MessagePack 的全局实例来定制 [#f2]_
::
# 可能会修改的几个主要配置
Settings.MESSAGE_MAX: int = Worker 和 Client 实例里等待处理的消息最大数量
Settings.HUGE_DATA_SIZEOF: int = 每次传输的结果值的最大大小，超过该值的将会被压缩并分片传输
Settings.HUGE_DATA_COMPRESS_MODULE: str = 使用的压缩模块的名称 [#f1]_
Settings.SERVICE_DEFAULT_NAME: str = 默认的服务名，当在实例化 Service 时如果不提供 name 参数则会以这个为服务名
Settings.MDP_INTERNAL_SERVICE_PREFIX: bytes = MDP 内部服务的前缀
Settings.MDP_HEARTBEAT_INTERVAL: int = 服务端和客户端相对于中间代理的心跳间隔时间，默认1500毫秒
Settings.MDP_HEARTBEAT_LIVENESS: int = 判定掉线的丢失心跳次数，即当超过该次数*心跳时间没有收到心跳则认为已经掉线，默认3次
Settings.REPLY_TIMEOUT: float = 客户端调用远程方法时，等待回复的超时时间，应设置的远远大于心跳时间，默认是一分钟
Settings.STREAM_REPLY_MAXSIZE: int = 流式数据使用的缓存队列的最大长度（使用的 asyncio.Queue）
Settings.REPLY_TIMEOUT: float = 获取回复的超时时间，也是流式传输的每一个子回复的超时时间
Settings.ZAP_PLAIN_DEFAULT_USER: str = ZAP 的 PLAIN 机制的默认用户名
Settings.ZAP_PLAIN_DEFAULT_PASSWORD: str = ZAP 的 PLAIN 机制的默认密码
Settings.ZAP_ADDR: str = ZAP 服务绑定的地址
Settings.ZAP_REPLY_TIMEOUT: float = 等待 ZAP 服务的回复的超时时间
Settings.setup()
# 特殊返回值的序列化配置 [#f2]_
from gaterpc.utils.message_pack
message_pack.prepare_pack = 在使用 msgpack.packb 时，传递给 default 参数的可执行对象
message_pack.unpack_object_hook = 在使用 msgpack.unpackb 时，传递给 object_hook 的可执行对象
message_pack.unpack_object_pairs_hook = 在使用 msgpack.unpackb 时，传递给 object_pairs_hook 的可执行对象
message_pack.unpack_object_list_hook = 在使用 msgpack.unpackb 时，传递给 list_hook 的可执行对象
.. rubric:: Footnotes
.. [#f1] Settings.HUGE_DATA_COMPRESS_MODULE 除了内置的 gzip，bz2，lzma，还可以使用外部模块，只要模块提供 compressor 和 decompressor 方法即可，
compressor 需要返回一个带有 compress 方法的增量压缩器对象，decompressor 需要返回一个带有 decompress 的增量解压缩器对象
.. [#f2] 单一返回值和生成器的元素返回值，以及巨型返回值都会使用 utils.msg_pack 和 utils.msg_unpack 来序列化和反序列化
这两个方法内部是使用的 utils.MessagePack 的全局实例，如果不能返回常规的“字符串”，“列表”，“字典”返回值，建议配置这几个配置
********
测试示范
********
实例化 ZAP 服务后，需要配置校验策略
::
zap = AsyncZAPService()
zap.configure_plain(
Settings.ZAP_DEFAULT_DOMAIN,
{
Settings.ZAP_PLAIN_DEFAULT_USER: Settings.ZAP_PLAIN_DEFAULT_PASSWORD
}
)
zap.start()
继承Worker类，用interface装饰希望被远程调用的方法，然后实例化一个Server来创建Worker的实例，这个worker实例的描述信息由server实例提供。
::
# Worker
class GRWorker(Worker):
@interface
async def atest(self, *args, **kwargs):
loop = self._get_loop()
return {
"name": "async atest",
"args": args,
"kwargs": kwargs,
"loop_time": loop.time()
}
@interface
def test(self, *args, **kwargs):
return {
"name": "test",
"args": args,
"kwargs": kwargs,
"loop_time": time()
}
@interface
def test_generator(self, maximum: int):
i = 0
while i < maximum:
yield i
i += 1
@interface
async def test_agenerator(self, maximum: int):
i = 0
while i < maximum:
await asyncio.sleep(0.1)
yield i
i += 1
Settings.setup()
gr = Service(name="SRkv")
gr_worker = gr.create_worker(
GRWorker, "tcp://127.0.0.1:5555",
zap_mechanism=Settings.ZAP_MECHANISM_PLAIN.decode("utf-8"),
zap_credentials=(
Settings.ZAP_PLAIN_DEFAULT_USER,
Settings.ZAP_PLAIN_DEFAULT_PASSWORD
)
)
gr_worker.run()
当要执行 IO 密集或 CPU 密集型操作时，可以在方法内使用执行器来执行，可以使用自带的两个执行器，也可以使用自定义的；
另，所有同步的函数都会使用默认执行器执行，默认执行器是 ThreadPoolExecutor 实例，可以修改。
::
@interface
async def test_io():
result = await self.run_in_executor(self.thread_executor, func, *args, **kwargs)
return result
@interface
async def test_cpu():
# 如果需要和 CPU 密集型执行器里的方法交换数据，可以使用 utils.SyncManager 来创建代理对象使用。
queue = SyncManager.Queue()
result = await self.run_in_executor(self.process_executor, func, queue, *args, **kwargs)
return result
实例化代理时会绑定两个地址，一个用于给后端服务连接上来，一个给前端客户端连接上来，bind 方法是绑定的给客户端访问的地址也就是前端地址。
::
# Majordomo
class GRMajordomo(AMajordomo):
# 可以新增内部处理程序，用于扩展分布式应用，所有内部处理程序只能接收kwargs关键词参数
# kwargs 的结构是固定的
# kwargs = {
# "client_id": client_id,
# "client_addr": client_addr,
# "request_id": request_id,
# "body": body
# }
@interface
def internal_x_process(**kwargs):
return stat_code
@interface
async def internal_y_process(**kwargs):
return stat_code
Settings.setup()
gr_majordomo = GRMajordomo(
backend_addr="tcp://127.0.0.1:5555",
zap_mechanism=Settings.ZAP_MECHANISM_PLAIN.decode("utf-8"),
zap_addr=Settings.ZAP_ADDR
)
gr_majordomo.bind("tcp://127.0.0.1:777")
gr_majordomo.run()
客户端直接连接代理地址，使用点语法调用远程方法，一般格式是 client.服务名.方法名，当直接使用 client.方法名时，会使用默认服务名调用。
::
# Client
Settings.setup()
gr_cli = Client(
broker_addr="tcp://127.0.0.1:777"
zap_mechanism=Settings.ZAP_MECHANISM_PLAIN.decode("utf-8"),
zap_credentials=(
Settings.ZAP_PLAIN_DEFAULT_USER,
Settings.ZAP_PLAIN_DEFAULT_PASSWORD
)
)
await gr_cli.SRkv.test("a", "b", "c", time=time())
await gr_cli.SRkv.atest("a", "b", "c", time=time())
async for i in await gr_cli.SRkv.test_agenerator(10):
print(i)
客户端调用的远程方法后，会创建一个延迟回调用来删掉缓存的已经执行完毕的请求，包括超时没拿到回复的请求，
而流式回复会每次回调时都检查一次该 StreamReply 实例是否已经结束，没结束就再创建一个延迟回调后续再检查
********
注意点
********
客户端和服务端对请求和回复的异步处理是使用的 utils.BoundedDict 异步字典来处理
::
# 请求远程方法
request_id = await request(service_name, body)
response = await requests.aget(request_id, timeout=reply_timeout)
# 接收回复
response = await socket.recv_multipart()
await requests.aset(request_id, response)
如果自定义方法的返回对象的大小无法使用 sys.getsizeof 准确获取，建议用 HugeData 包装后再返回
::
# data 必须要是 bytes 或 bytearray，简言之能用 memoryview 包装的
hd = HugeData(data=data, compress_module="gzip", compress_level=9)
# 或者不提供 data ，HugeData 初始化时会创建一个 Queue 的跨进程代理对象，往这个跨进程队列里传输数据即可
hd = HugeData(compress_module="gzip", compress_level=9)
d = process_data()
hd.data.put(d)
HugeData 的 compress 和 decompress 方法都会在进程池里执行增量压缩和增量解压缩，
返回的生成器每次获取的字节数大小不会超过 Settings.HUGE_DATA_SIZEOF ，
compress 方法对每一块返回的大小的限制是 HugeData 内部实现，
decompress 方法对每一块返回的大小限制则是由压缩模块来实现，会在调用解压缩器实例的 decompress 方法时传递一个 max_length 位置参数。
Raw data
{
"_id": null,
"home_page": "https://github.com/firejoke/gate-rpc",
"name": "gaterpc",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": "",
"keywords": "zero,zeroMQ,rpc",
"author": "Shi Fan",
"author_email": "Shi Fan <firejokeshi@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/9b/0e/e34288e1739bf1546490ab360f7cc59d1e9605291d15d253c0ad776f3259/gaterpc-0.1.10.tar.gz",
"platform": null,
"description": "gate-rpc\n############\n\n\u4f7f\u7528 ZeroMQ \u548c asyncio \u5f00\u53d1\u7684\u201cgate-rpc\u201d\u3002\n\n- \u4f7f\u7528msgpack\u5e8f\u5217\u5316\u6d88\u606f\n- \u652f\u6301\u5728\u7ebf\u7a0b\u6c60\u548c\u8fdb\u7a0b\u6c60\u91cc\u8fd0\u884c\u666e\u901a\u51fd\u6570\n- \u5f53\u51fd\u6570\u8fd4\u56de\u7684\u662f Generator \u6216 AsyncGenerator \u65f6\uff0c\u4f1a\u6362\u6210\u6d41\u5f0f\u4f20\u8f93\uff0c\u63a5\u6536\u7aef\u901a\u8fc7\u904d\u5386 StreamReply \u5b9e\u4f8b\u5373\u53ef\u83b7\u5f97\n- \u5f53\u51fd\u6570\u8fd4\u56de\u7684\u503c\u8fc7\u5927\u65f6\uff0c\u4f1a\u8f6c\u6362\u6210 HugeData \u5bf9\u8c61\uff0c\u538b\u7f29\u8fd4\u56de\u503c\u5e76\u5206\u5757\u6d41\u5f0f\u4f20\u8f93\uff0c\u9ed8\u8ba4\u5757\u5927\u5c0f\u901a\u8fc7 Settings \u7684 HUGE_DATA_SIZEOF \u6765\u8bbe\u7f6e\n- \u4f7f\u7528\u961f\u5217\u4f20\u8f93\u548c\u8bb0\u5f55\u65e5\u5fd7\uff0c\u907f\u514d\u56e0\u4e3a\u65e5\u5fd7\u5bfc\u81f4\u4e8b\u4ef6\u5faa\u73af\u88ab\u963b\u585e\n- \u4f7f\u7528\u53ef\u4ee5\u5f02\u6b65\u8bbe\u7f6e\u952e\u503c\u548c\u5f02\u6b65\u83b7\u53d6\u952e\u503c\u7684 BoundedDict \u6765\u7b80\u5316\u8d85\u65f6\u7b49\u5f85\u83b7\u53d6\n- \u5728\u6bcf\u6b21\u5b9e\u4f8b\u5316 Worker\u3001Server\u3001AMajordomo\u3001Client \u5404\u7c7b\u4e4b\u524d\uff0c\u901a\u8fc7\u4fee\u6539 Settings \u7684\u5c5e\u6027\u6765\u4fee\u6539\u8fd0\u884c\u914d\u7f6e\n\n********\n\u5b89\u88c5\n********\n\n\u53ef\u4ee5\u76f4\u63a5\u4f7f\u7528pip\u5b89\u88c5\uff0c\u6216\u76f4\u63a5\u4e0b\u8f7d\u6e90\u7801\u968f\u610f\u653e\u7f6e\u3002\n\n::\n\n pip install gaterpc\n\n\n********\n\u914d\u7f6e\n********\n\u5728\u5b9e\u4f8b\u5316 Worker\u3001Service\u3001AMajordomo\u3001Client \u5404\u7c7b\u4e4b\u524d\uff0c\u9700\u8981\u8fd0\u884c Settings.setup \u51fd\u6570\u6765\u914d\u7f6e\u5168\u5c40\u914d\u7f6e [#f1]_ \uff0c\n\u7279\u6b8a\u8fd4\u56de\u503c\u7684\u5e8f\u5217\u5316\u901a\u8fc7 MessagePack \u7684\u5168\u5c40\u5b9e\u4f8b\u6765\u5b9a\u5236 [#f2]_\n\n::\n\n # \u53ef\u80fd\u4f1a\u4fee\u6539\u7684\u51e0\u4e2a\u4e3b\u8981\u914d\u7f6e\n Settings.MESSAGE_MAX: int = Worker \u548c Client \u5b9e\u4f8b\u91cc\u7b49\u5f85\u5904\u7406\u7684\u6d88\u606f\u6700\u5927\u6570\u91cf\n Settings.HUGE_DATA_SIZEOF: int = \u6bcf\u6b21\u4f20\u8f93\u7684\u7ed3\u679c\u503c\u7684\u6700\u5927\u5927\u5c0f\uff0c\u8d85\u8fc7\u8be5\u503c\u7684\u5c06\u4f1a\u88ab\u538b\u7f29\u5e76\u5206\u7247\u4f20\u8f93\n Settings.HUGE_DATA_COMPRESS_MODULE: str = \u4f7f\u7528\u7684\u538b\u7f29\u6a21\u5757\u7684\u540d\u79f0 [#f1]_\n Settings.SERVICE_DEFAULT_NAME: str = \u9ed8\u8ba4\u7684\u670d\u52a1\u540d\uff0c\u5f53\u5728\u5b9e\u4f8b\u5316 Service \u65f6\u5982\u679c\u4e0d\u63d0\u4f9b name \u53c2\u6570\u5219\u4f1a\u4ee5\u8fd9\u4e2a\u4e3a\u670d\u52a1\u540d\n Settings.MDP_INTERNAL_SERVICE_PREFIX: bytes = MDP \u5185\u90e8\u670d\u52a1\u7684\u524d\u7f00\n Settings.MDP_HEARTBEAT_INTERVAL: int = \u670d\u52a1\u7aef\u548c\u5ba2\u6237\u7aef\u76f8\u5bf9\u4e8e\u4e2d\u95f4\u4ee3\u7406\u7684\u5fc3\u8df3\u95f4\u9694\u65f6\u95f4\uff0c\u9ed8\u8ba41500\u6beb\u79d2\n Settings.MDP_HEARTBEAT_LIVENESS: int = \u5224\u5b9a\u6389\u7ebf\u7684\u4e22\u5931\u5fc3\u8df3\u6b21\u6570\uff0c\u5373\u5f53\u8d85\u8fc7\u8be5\u6b21\u6570*\u5fc3\u8df3\u65f6\u95f4\u6ca1\u6709\u6536\u5230\u5fc3\u8df3\u5219\u8ba4\u4e3a\u5df2\u7ecf\u6389\u7ebf\uff0c\u9ed8\u8ba43\u6b21\n Settings.REPLY_TIMEOUT: float = \u5ba2\u6237\u7aef\u8c03\u7528\u8fdc\u7a0b\u65b9\u6cd5\u65f6\uff0c\u7b49\u5f85\u56de\u590d\u7684\u8d85\u65f6\u65f6\u95f4\uff0c\u5e94\u8bbe\u7f6e\u7684\u8fdc\u8fdc\u5927\u4e8e\u5fc3\u8df3\u65f6\u95f4\uff0c\u9ed8\u8ba4\u662f\u4e00\u5206\u949f\n Settings.STREAM_REPLY_MAXSIZE: int = \u6d41\u5f0f\u6570\u636e\u4f7f\u7528\u7684\u7f13\u5b58\u961f\u5217\u7684\u6700\u5927\u957f\u5ea6\uff08\u4f7f\u7528\u7684 asyncio.Queue\uff09\n Settings.REPLY_TIMEOUT: float = \u83b7\u53d6\u56de\u590d\u7684\u8d85\u65f6\u65f6\u95f4\uff0c\u4e5f\u662f\u6d41\u5f0f\u4f20\u8f93\u7684\u6bcf\u4e00\u4e2a\u5b50\u56de\u590d\u7684\u8d85\u65f6\u65f6\u95f4\n Settings.ZAP_PLAIN_DEFAULT_USER: str = ZAP \u7684 PLAIN \u673a\u5236\u7684\u9ed8\u8ba4\u7528\u6237\u540d\n Settings.ZAP_PLAIN_DEFAULT_PASSWORD: str = ZAP \u7684 PLAIN \u673a\u5236\u7684\u9ed8\u8ba4\u5bc6\u7801\n Settings.ZAP_ADDR: str = ZAP \u670d\u52a1\u7ed1\u5b9a\u7684\u5730\u5740\n Settings.ZAP_REPLY_TIMEOUT: float = \u7b49\u5f85 ZAP \u670d\u52a1\u7684\u56de\u590d\u7684\u8d85\u65f6\u65f6\u95f4\n Settings.setup()\n\n # \u7279\u6b8a\u8fd4\u56de\u503c\u7684\u5e8f\u5217\u5316\u914d\u7f6e [#f2]_\n from gaterpc.utils.message_pack\n message_pack.prepare_pack = \u5728\u4f7f\u7528 msgpack.packb \u65f6\uff0c\u4f20\u9012\u7ed9 default \u53c2\u6570\u7684\u53ef\u6267\u884c\u5bf9\u8c61\n message_pack.unpack_object_hook = \u5728\u4f7f\u7528 msgpack.unpackb \u65f6\uff0c\u4f20\u9012\u7ed9 object_hook \u7684\u53ef\u6267\u884c\u5bf9\u8c61\n message_pack.unpack_object_pairs_hook = \u5728\u4f7f\u7528 msgpack.unpackb \u65f6\uff0c\u4f20\u9012\u7ed9 object_pairs_hook \u7684\u53ef\u6267\u884c\u5bf9\u8c61\n message_pack.unpack_object_list_hook = \u5728\u4f7f\u7528 msgpack.unpackb \u65f6\uff0c\u4f20\u9012\u7ed9 list_hook \u7684\u53ef\u6267\u884c\u5bf9\u8c61\n\n\n.. rubric:: Footnotes\n.. [#f1] Settings.HUGE_DATA_COMPRESS_MODULE \u9664\u4e86\u5185\u7f6e\u7684 gzip\uff0cbz2\uff0clzma\uff0c\u8fd8\u53ef\u4ee5\u4f7f\u7528\u5916\u90e8\u6a21\u5757\uff0c\u53ea\u8981\u6a21\u5757\u63d0\u4f9b compressor \u548c decompressor \u65b9\u6cd5\u5373\u53ef\uff0c\n compressor \u9700\u8981\u8fd4\u56de\u4e00\u4e2a\u5e26\u6709 compress \u65b9\u6cd5\u7684\u589e\u91cf\u538b\u7f29\u5668\u5bf9\u8c61\uff0cdecompressor \u9700\u8981\u8fd4\u56de\u4e00\u4e2a\u5e26\u6709 decompress \u7684\u589e\u91cf\u89e3\u538b\u7f29\u5668\u5bf9\u8c61\n.. [#f2] \u5355\u4e00\u8fd4\u56de\u503c\u548c\u751f\u6210\u5668\u7684\u5143\u7d20\u8fd4\u56de\u503c\uff0c\u4ee5\u53ca\u5de8\u578b\u8fd4\u56de\u503c\u90fd\u4f1a\u4f7f\u7528 utils.msg_pack \u548c utils.msg_unpack \u6765\u5e8f\u5217\u5316\u548c\u53cd\u5e8f\u5217\u5316\n \u8fd9\u4e24\u4e2a\u65b9\u6cd5\u5185\u90e8\u662f\u4f7f\u7528\u7684 utils.MessagePack \u7684\u5168\u5c40\u5b9e\u4f8b\uff0c\u5982\u679c\u4e0d\u80fd\u8fd4\u56de\u5e38\u89c4\u7684\u201c\u5b57\u7b26\u4e32\u201d\uff0c\u201c\u5217\u8868\u201d\uff0c\u201c\u5b57\u5178\u201d\u8fd4\u56de\u503c\uff0c\u5efa\u8bae\u914d\u7f6e\u8fd9\u51e0\u4e2a\u914d\u7f6e\n\n********\n\u6d4b\u8bd5\u793a\u8303\n********\n\n\u5b9e\u4f8b\u5316 ZAP \u670d\u52a1\u540e\uff0c\u9700\u8981\u914d\u7f6e\u6821\u9a8c\u7b56\u7565\n\n::\n\n zap = AsyncZAPService()\n zap.configure_plain(\n Settings.ZAP_DEFAULT_DOMAIN,\n {\n Settings.ZAP_PLAIN_DEFAULT_USER: Settings.ZAP_PLAIN_DEFAULT_PASSWORD\n }\n )\n zap.start()\n\n\n\u7ee7\u627fWorker\u7c7b\uff0c\u7528interface\u88c5\u9970\u5e0c\u671b\u88ab\u8fdc\u7a0b\u8c03\u7528\u7684\u65b9\u6cd5\uff0c\u7136\u540e\u5b9e\u4f8b\u5316\u4e00\u4e2aServer\u6765\u521b\u5efaWorker\u7684\u5b9e\u4f8b\uff0c\u8fd9\u4e2aworker\u5b9e\u4f8b\u7684\u63cf\u8ff0\u4fe1\u606f\u7531server\u5b9e\u4f8b\u63d0\u4f9b\u3002\n\n::\n\n # Worker\n class GRWorker(Worker):\n @interface\n async def atest(self, *args, **kwargs):\n loop = self._get_loop()\n return {\n \"name\": \"async atest\",\n \"args\": args,\n \"kwargs\": kwargs,\n \"loop_time\": loop.time()\n }\n\n @interface\n def test(self, *args, **kwargs):\n return {\n \"name\": \"test\",\n \"args\": args,\n \"kwargs\": kwargs,\n \"loop_time\": time()\n }\n\n @interface\n def test_generator(self, maximum: int):\n i = 0\n while i < maximum:\n yield i\n i += 1\n\n @interface\n async def test_agenerator(self, maximum: int):\n i = 0\n while i < maximum:\n await asyncio.sleep(0.1)\n yield i\n i += 1\n\n Settings.setup()\n gr = Service(name=\"SRkv\")\n gr_worker = gr.create_worker(\n GRWorker, \"tcp://127.0.0.1:5555\",\n zap_mechanism=Settings.ZAP_MECHANISM_PLAIN.decode(\"utf-8\"),\n zap_credentials=(\n Settings.ZAP_PLAIN_DEFAULT_USER,\n Settings.ZAP_PLAIN_DEFAULT_PASSWORD\n )\n )\n gr_worker.run()\n\n\u5f53\u8981\u6267\u884c IO \u5bc6\u96c6\u6216 CPU \u5bc6\u96c6\u578b\u64cd\u4f5c\u65f6\uff0c\u53ef\u4ee5\u5728\u65b9\u6cd5\u5185\u4f7f\u7528\u6267\u884c\u5668\u6765\u6267\u884c\uff0c\u53ef\u4ee5\u4f7f\u7528\u81ea\u5e26\u7684\u4e24\u4e2a\u6267\u884c\u5668\uff0c\u4e5f\u53ef\u4ee5\u4f7f\u7528\u81ea\u5b9a\u4e49\u7684\uff1b\n\u53e6\uff0c\u6240\u6709\u540c\u6b65\u7684\u51fd\u6570\u90fd\u4f1a\u4f7f\u7528\u9ed8\u8ba4\u6267\u884c\u5668\u6267\u884c\uff0c\u9ed8\u8ba4\u6267\u884c\u5668\u662f ThreadPoolExecutor \u5b9e\u4f8b\uff0c\u53ef\u4ee5\u4fee\u6539\u3002\n\n::\n\n @interface\n async def test_io():\n result = await self.run_in_executor(self.thread_executor, func, *args, **kwargs)\n return result\n\n @interface\n async def test_cpu():\n # \u5982\u679c\u9700\u8981\u548c CPU \u5bc6\u96c6\u578b\u6267\u884c\u5668\u91cc\u7684\u65b9\u6cd5\u4ea4\u6362\u6570\u636e\uff0c\u53ef\u4ee5\u4f7f\u7528 utils.SyncManager \u6765\u521b\u5efa\u4ee3\u7406\u5bf9\u8c61\u4f7f\u7528\u3002\n queue = SyncManager.Queue()\n result = await self.run_in_executor(self.process_executor, func, queue, *args, **kwargs)\n return result\n\n\u5b9e\u4f8b\u5316\u4ee3\u7406\u65f6\u4f1a\u7ed1\u5b9a\u4e24\u4e2a\u5730\u5740\uff0c\u4e00\u4e2a\u7528\u4e8e\u7ed9\u540e\u7aef\u670d\u52a1\u8fde\u63a5\u4e0a\u6765\uff0c\u4e00\u4e2a\u7ed9\u524d\u7aef\u5ba2\u6237\u7aef\u8fde\u63a5\u4e0a\u6765\uff0cbind \u65b9\u6cd5\u662f\u7ed1\u5b9a\u7684\u7ed9\u5ba2\u6237\u7aef\u8bbf\u95ee\u7684\u5730\u5740\u4e5f\u5c31\u662f\u524d\u7aef\u5730\u5740\u3002\n\n::\n\n # Majordomo\n class GRMajordomo(AMajordomo):\n # \u53ef\u4ee5\u65b0\u589e\u5185\u90e8\u5904\u7406\u7a0b\u5e8f\uff0c\u7528\u4e8e\u6269\u5c55\u5206\u5e03\u5f0f\u5e94\u7528\uff0c\u6240\u6709\u5185\u90e8\u5904\u7406\u7a0b\u5e8f\u53ea\u80fd\u63a5\u6536kwargs\u5173\u952e\u8bcd\u53c2\u6570\n # kwargs \u7684\u7ed3\u6784\u662f\u56fa\u5b9a\u7684\n # kwargs = {\n # \"client_id\": client_id,\n # \"client_addr\": client_addr,\n # \"request_id\": request_id,\n # \"body\": body\n # }\n @interface\n def internal_x_process(**kwargs):\n return stat_code\n\n @interface\n async def internal_y_process(**kwargs):\n return stat_code\n\n Settings.setup()\n gr_majordomo = GRMajordomo(\n backend_addr=\"tcp://127.0.0.1:5555\",\n zap_mechanism=Settings.ZAP_MECHANISM_PLAIN.decode(\"utf-8\"),\n zap_addr=Settings.ZAP_ADDR\n )\n gr_majordomo.bind(\"tcp://127.0.0.1:777\")\n gr_majordomo.run()\n\n\u5ba2\u6237\u7aef\u76f4\u63a5\u8fde\u63a5\u4ee3\u7406\u5730\u5740\uff0c\u4f7f\u7528\u70b9\u8bed\u6cd5\u8c03\u7528\u8fdc\u7a0b\u65b9\u6cd5\uff0c\u4e00\u822c\u683c\u5f0f\u662f client.\u670d\u52a1\u540d.\u65b9\u6cd5\u540d\uff0c\u5f53\u76f4\u63a5\u4f7f\u7528 client.\u65b9\u6cd5\u540d\u65f6\uff0c\u4f1a\u4f7f\u7528\u9ed8\u8ba4\u670d\u52a1\u540d\u8c03\u7528\u3002\n\n::\n\n # Client\n Settings.setup()\n gr_cli = Client(\n broker_addr=\"tcp://127.0.0.1:777\"\n zap_mechanism=Settings.ZAP_MECHANISM_PLAIN.decode(\"utf-8\"),\n zap_credentials=(\n Settings.ZAP_PLAIN_DEFAULT_USER,\n Settings.ZAP_PLAIN_DEFAULT_PASSWORD\n )\n )\n await gr_cli.SRkv.test(\"a\", \"b\", \"c\", time=time())\n await gr_cli.SRkv.atest(\"a\", \"b\", \"c\", time=time())\n async for i in await gr_cli.SRkv.test_agenerator(10):\n print(i)\n\n\n\u5ba2\u6237\u7aef\u8c03\u7528\u7684\u8fdc\u7a0b\u65b9\u6cd5\u540e\uff0c\u4f1a\u521b\u5efa\u4e00\u4e2a\u5ef6\u8fdf\u56de\u8c03\u7528\u6765\u5220\u6389\u7f13\u5b58\u7684\u5df2\u7ecf\u6267\u884c\u5b8c\u6bd5\u7684\u8bf7\u6c42\uff0c\u5305\u62ec\u8d85\u65f6\u6ca1\u62ff\u5230\u56de\u590d\u7684\u8bf7\u6c42\uff0c\n\u800c\u6d41\u5f0f\u56de\u590d\u4f1a\u6bcf\u6b21\u56de\u8c03\u65f6\u90fd\u68c0\u67e5\u4e00\u6b21\u8be5 StreamReply \u5b9e\u4f8b\u662f\u5426\u5df2\u7ecf\u7ed3\u675f\uff0c\u6ca1\u7ed3\u675f\u5c31\u518d\u521b\u5efa\u4e00\u4e2a\u5ef6\u8fdf\u56de\u8c03\u540e\u7eed\u518d\u68c0\u67e5\n\n\n********\n\u6ce8\u610f\u70b9\n********\n\n\u5ba2\u6237\u7aef\u548c\u670d\u52a1\u7aef\u5bf9\u8bf7\u6c42\u548c\u56de\u590d\u7684\u5f02\u6b65\u5904\u7406\u662f\u4f7f\u7528\u7684 utils.BoundedDict \u5f02\u6b65\u5b57\u5178\u6765\u5904\u7406\n\n::\n\n # \u8bf7\u6c42\u8fdc\u7a0b\u65b9\u6cd5\n request_id = await request(service_name, body)\n response = await requests.aget(request_id, timeout=reply_timeout)\n # \u63a5\u6536\u56de\u590d\n response = await socket.recv_multipart()\n await requests.aset(request_id, response)\n\n\u5982\u679c\u81ea\u5b9a\u4e49\u65b9\u6cd5\u7684\u8fd4\u56de\u5bf9\u8c61\u7684\u5927\u5c0f\u65e0\u6cd5\u4f7f\u7528 sys.getsizeof \u51c6\u786e\u83b7\u53d6\uff0c\u5efa\u8bae\u7528 HugeData \u5305\u88c5\u540e\u518d\u8fd4\u56de\n\n::\n\n # data \u5fc5\u987b\u8981\u662f bytes \u6216 bytearray\uff0c\u7b80\u8a00\u4e4b\u80fd\u7528 memoryview \u5305\u88c5\u7684\n hd = HugeData(data=data, compress_module=\"gzip\", compress_level=9)\n # \u6216\u8005\u4e0d\u63d0\u4f9b data \uff0cHugeData \u521d\u59cb\u5316\u65f6\u4f1a\u521b\u5efa\u4e00\u4e2a Queue \u7684\u8de8\u8fdb\u7a0b\u4ee3\u7406\u5bf9\u8c61\uff0c\u5f80\u8fd9\u4e2a\u8de8\u8fdb\u7a0b\u961f\u5217\u91cc\u4f20\u8f93\u6570\u636e\u5373\u53ef\n hd = HugeData(compress_module=\"gzip\", compress_level=9)\n d = process_data()\n hd.data.put(d)\n\nHugeData \u7684 compress \u548c decompress \u65b9\u6cd5\u90fd\u4f1a\u5728\u8fdb\u7a0b\u6c60\u91cc\u6267\u884c\u589e\u91cf\u538b\u7f29\u548c\u589e\u91cf\u89e3\u538b\u7f29\uff0c\n\u8fd4\u56de\u7684\u751f\u6210\u5668\u6bcf\u6b21\u83b7\u53d6\u7684\u5b57\u8282\u6570\u5927\u5c0f\u4e0d\u4f1a\u8d85\u8fc7 Settings.HUGE_DATA_SIZEOF \uff0c\ncompress \u65b9\u6cd5\u5bf9\u6bcf\u4e00\u5757\u8fd4\u56de\u7684\u5927\u5c0f\u7684\u9650\u5236\u662f HugeData \u5185\u90e8\u5b9e\u73b0\uff0c\ndecompress \u65b9\u6cd5\u5bf9\u6bcf\u4e00\u5757\u8fd4\u56de\u7684\u5927\u5c0f\u9650\u5236\u5219\u662f\u7531\u538b\u7f29\u6a21\u5757\u6765\u5b9e\u73b0\uff0c\u4f1a\u5728\u8c03\u7528\u89e3\u538b\u7f29\u5668\u5b9e\u4f8b\u7684 decompress \u65b9\u6cd5\u65f6\u4f20\u9012\u4e00\u4e2a max_length \u4f4d\u7f6e\u53c2\u6570\u3002\n",
"bugtrack_url": null,
"license": "BSD 3-Clause License Copyright (c) 2024, shifan All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. ",
"summary": "A RPC software based on ZeroMQ with built-in Majordomo.",
"version": "0.1.10",
"project_urls": {
"Homepage": "https://github.com/firejoke/gate-rpc",
"Repository": "https://github.com/firejoke/gate-rpc"
},
"split_keywords": [
"zero",
"zeromq",
"rpc"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "5ba6bb5555961f683f3e9cfe0c8dc81a62bfbda4be1d42a58f7fd2e58a3f541b",
"md5": "dee5f582d8fbc7e21146da638638c1cc",
"sha256": "fb2ef8be1cf83d58f3e14bd146d9e9a6ae0c0ae4d6eff74753ed5c8f457874a4"
},
"downloads": -1,
"filename": "gaterpc-0.1.10-py3-none-any.whl",
"has_sig": false,
"md5_digest": "dee5f582d8fbc7e21146da638638c1cc",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 33475,
"upload_time": "2024-02-21T08:32:16",
"upload_time_iso_8601": "2024-02-21T08:32:16.550846Z",
"url": "https://files.pythonhosted.org/packages/5b/a6/bb5555961f683f3e9cfe0c8dc81a62bfbda4be1d42a58f7fd2e58a3f541b/gaterpc-0.1.10-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "9b0ee34288e1739bf1546490ab360f7cc59d1e9605291d15d253c0ad776f3259",
"md5": "a4697254d46328370a7655c814136ac2",
"sha256": "570cd7fa934c9aec70d958e0c1a319782d7c33aa96eae04583000a46d7d59f6a"
},
"downloads": -1,
"filename": "gaterpc-0.1.10.tar.gz",
"has_sig": false,
"md5_digest": "a4697254d46328370a7655c814136ac2",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 33912,
"upload_time": "2024-02-21T08:32:19",
"upload_time_iso_8601": "2024-02-21T08:32:19.044418Z",
"url": "https://files.pythonhosted.org/packages/9b/0e/e34288e1739bf1546490ab360f7cc59d1e9605291d15d253c0ad776f3259/gaterpc-0.1.10.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-02-21 08:32:19",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "firejoke",
"github_project": "gate-rpc",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [
{
"name": "msgpack",
"specs": [
[
">=",
"1.0.7"
]
]
},
{
"name": "pyzmq",
"specs": [
[
">=",
"25.1.2"
]
]
}
],
"lcname": "gaterpc"
}
