使用 PyZMQ 流式传输标准输出

Question

我有一个客户端应用程序，它向远程服务器发送消息，消息从服务器得到处理，结果被发送回客户端。

一个简化的例子如下：

client.py

import zmq
def get_message(msg, ip, port, socket, send_socket_type='pyobj'):
    socket.connect("tcp://%s:%s" % (ip, port))
    socket.send_string(msg)
    if send_socket_type=='pyobj':
        msg = socket.recv_pyobj()
    if send_socket_type=='string':
        msg = socket.recv_string()
    if send_socket_type=='json':
        msg = socket.recv_json()
    return msg
    socket.close()

server.py

import zmq
import time

port=9999
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:%s" % port)


def processmessage(msg):
    for i in range(10):
        print(i)
        time.sleep(1)
    return msg

while True:
    msg = socket.recv()
    print('message received from client:', msg)
    cmd = processmessage(msg)
    socket.send_pyobj(cmd)
    print('sending to client: %s ' % msg)
    time.sleep(0.2)

测试：

server_env={'IP': ip, 'PORT': port}

context = zmq.Context()
socket = context.socket(zmq.REQ)

messages = ['first message', 'second message', 'third message']

for i in messages:
  msg = get_message('message', ip=server_env['IP'], port=server_env['PORT'], socket)
  print(msg)

我的问题：

• 如何扩展这个简单的客户端-服务器示例以允许客户端接收由运行服务器端的进程processmessage() 生成的标准输出？

Answer 1

您可以在处理消息时使用上下文管理器将 sys.stdout 重定向到 StringIO 对象：

from contextlib import contextmanager
import io
import sys

@contextmanager
def capture():
    save_stdout = sys.stdout
    sys.stdout = io.StringIO()
    try:
        yield sys.stdout
    finally:
        sys.stdout = save_stdout


def processmessage(msg):
    with capture() as stdout:
        for i in range(10):
            print(i)
            time.sleep(1)
    return {
        'reply': msg,
        'stdout': stdout.getvalue(),
    }

使用此上下文管理器可以在任何代码块期间捕获标准输出，并在上下文结束后通过stdout.getvalue() 检索标准输出上发生的任何输出。然后，您可以发送包含两个字段的回复：结果本身和处理期间捕获的标准输出。

如果您有兴趣，也可以通过以下方式对 stderr 执行相同操作：

1. 将 stderr 重定向到同一个 StringIO 对象，因此 stdout/err 合并在一起，或者
2. 如果您想将它们分开，则将 stderr 发送到它自己的 StringIO 对象。