您可以创建架构，而无需任何外部库或添加任何额外的数据库复杂性等。
这将是我们在不同进程之间共享对象的服务器。
from multiprocessing.managers import SyncManager
class MyManager(SyncManager):
pass
syncdict = {}
def get_dict():
return syncdict
if __name__ == "__main__":
MyManager.register("syncdict", get_dict)
manager = MyManager(("127.0.0.1", 5000), authkey=b"password")
manager.start()
input()
manager.shutdown()
将此文件命名为server.py 并在不同的进程上运行它。只是python server.py 应该会很好。
让我们跳到我们的客户端实现。
这将是我们的客户端实现。
from multiprocessing.managers import SyncManager
from typing import Optional, Dict, Any, Union
class MyManager(SyncManager):
...
class Meta:
def __init__(self, *, port: int) -> None:
self.manager = MyManager(("127.0.0.1", port), authkey=b"password")
self.manager.connect()
MyManager.register("syncdict")
self.syndict = self.manager.syncdict()
def update(self, kwargs: Dict[Any, Any]) -> None:
self.syndict.update(kwargs)
def increase_one(self, key: str) -> None:
self.syndict.update([(key, self.syndict.get(key) + 1)])
def report(self, item: Union[str, int]) -> int:
return self.syndict.get(item)
meta = Meta(port=5000)
让我们将它与我们的 API 合并。
from fastapi import FastAPI, status
from multiprocessing.managers import SyncManager
from typing import Optional, Dict, Any, Union
class MyManager(SyncManager):
...
class Meta:
def __init__(self, *, port: int, **kwargs: Dict[Any, Any]):
self.manager = MyManager(("127.0.0.1", port), authkey=b"password")
self.manager.connect()
MyManager.register("syncdict")
self.syndict = self.manager.syncdict()
self.syndict.update(**kwargs)
def increase_one(self, key: str):
self.syndict.update([(key, self.syndict.get(key) + 1)])
def reset(self, key: str):
self.syndict.update([(key, 0)])
def report(self, item: Union[str, int]):
return self.syndict.get(item)
app = FastAPI()
meta = Meta(port=5000, cnt=0)
# increases the count variable in the meta object by 1
@app.get("/increment")
async def increment(key: str):
meta.increase_one(key)
return status.HTTP_200_OK
# returns a json containing the current count from the meta object
@app.get("/report")
async def report(key: str):
return {"count": meta.report(key)}
# resets the count in the meta object to 0
@app.get("/reset")
async def reset(key: str):
meta.reset(key)
return status.HTTP_200_OK
我将启动两个 API 实例，一个在 8000 上，另一个在 8001 上。
In: curl -X GET "http://127.0.0.1:8000/report?key=cnt"
Out: {"count": 0}
In: curl -X GET "http://127.0.0.1:8001/report?key=cnt"
Out: {"count": 0}
两者都以 0 值开始。现在让我们增加它
for _ in {1..10}; do curl -X GET "http://127.0.0.1:8000/increment?key=cnt" &; done
我在端口 8000 上运行 curl 10 次，这意味着 cnt 应该是 10。
让我们从端口8001检查一下：
In: curl -X GET "http://127.0.0.1:8001/report?key=cnt"
Out: {"cnt": 10}
像魅力一样工作。
需要考虑两件事。
- 您应该在不同的进程中启动您的应用程序。更具体地说，
uvicorn my_app:app 和您的服务器不应是父进程。
- 您可能想要添加类似正常关闭的内容。因为这是一个非常简单但高度可扩展的示例。