gRPC for python 到底如何处理 unix 域套接字

发布于 2025-01-16 04:40:01 字数 1479 浏览 4 评论 0原文

经过尝试一些事情后,我可以运行基于同一主机的 UDS 客户端-服务器 gRPC 连接。由于我没有找到此设置的任何详细文档,因此我主要担心的是我不知道我是否正确使用了 gRPC API。
任何参考/资源/解释都会受到

我查看的来源的热烈欢迎:
gRPC python API
gRPC python 文档
Python 中带有 Unix 域套接字的 gRPC 服务器 (SO)

gRPC连接:

服务器
import grpc
import my_pb2_grpc
from concurrent import futures

server = None
#...
class Listener(my_pb2_grpc.MyServiceServicer):
  #...

def create_connection():
    if server is None:
        server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
        my_pb2_grpc.add_MyServiceServicer_to_server(Listener(), server)
        server.add_insecure_port("unix:///path/to/uds.sock")
        server.start()

def stop_connection():
    if server:
        server.stop(0)
        server = None
客户端
import grpc
import my_pb2_grpc

channel = None
stub = None
#...

def create_connection():
    if channel is None:
        channel = grpc.insecure_channel("unix:///path/to/uds.sock")
        stub = my_pb2_grpc.MyServiceStub(channel)

def stop_connection():
    if channel:
        channel.close()
        channel = None
        stub = None

After trying a few things, I got to a point where I can run client-server gRPC connection that is UDS based on the same host. Since I didn't find any detailed documentation for this setup, my main concern is that I don't know if I use gRPC API properly.
Any reference/resource/explanation would be warmly welcomed

sources I looked at:
gRPC python API
gRPC python docs
gRPC server in Python with Unix domain socket (SO)

gRPC connection:

server
import grpc
import my_pb2_grpc
from concurrent import futures

server = None
#...
class Listener(my_pb2_grpc.MyServiceServicer):
  #...

def create_connection():
    if server is None:
        server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
        my_pb2_grpc.add_MyServiceServicer_to_server(Listener(), server)
        server.add_insecure_port("unix:///path/to/uds.sock")
        server.start()

def stop_connection():
    if server:
        server.stop(0)
        server = None
client
import grpc
import my_pb2_grpc

channel = None
stub = None
#...

def create_connection():
    if channel is None:
        channel = grpc.insecure_channel("unix:///path/to/uds.sock")
        stub = my_pb2_grpc.MyServiceStub(channel)

def stop_connection():
    if channel:
        channel.close()
        channel = None
        stub = None

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。
列表为空,暂无数据
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文