最近我不得不在具有跨平台兼容性的 python 中对 stdin、stdout 进行 I/O 读取。
对于 linux：
对于 linux，我们可以使用 select 模块。它是 posix select 函数的包装器实现。它允许您传递多个文件描述符，等待它们准备好。一旦他们准备好，您将收到通知并可以执行read/write 操作。这里有一些小代码可以让你有个想法
这里nodejs是一个带有nodejs镜像的docker环境
stdin_buf = BytesIO(json.dumps(fn) + "\n")
stdout_buf = BytesIO()
stderr_buf = BytesIO()
rselect = [nodejs.stdout, nodejs.stderr] # type: List[BytesIO]
wselect = [nodejs.stdin] # type: List[BytesIO]
while (len(wselect) + len(rselect)) > 0:
rready, wready, _ = select.select(rselect, wselect, [])
try:
if nodejs.stdin in wready:
b = stdin_buf.read(select.PIPE_BUF)
if b:
os.write(nodejs.stdin.fileno(), b)
else:
wselect = []
for pipes in ((nodejs.stdout, stdout_buf), (nodejs.stderr, stderr_buf)):
if pipes[0] in rready:
b = os.read(pipes[0].fileno(), select.PIPE_BUF)
if b:
pipes[1].write(b)
else:
rselect.remove(pipes[0])
if stdout_buf.getvalue().endswith("\n"):
rselect = []
except OSError as e:
break
适用于窗户
此代码示例具有涉及 stdin、stdout 的读取和写入操作。
现在此代码不适用于 Windows 操作系统，因为在 Windows 上选择实现不允许标准输入、标准输出作为参数传递。
文档说：
Windows 上的文件对象是不可接受的，但套接字是可接受的。在 Windows 上，底层的 select() 函数由 WinSock 库提供，并且不处理并非源自 WinSock 的文件描述符。
首先我必须提到，有很多用于在 Windows 上读取非阻塞 I/O 的库，例如 asyncio(python 3)、gevent(for python 2.7)、msvcrt，然后是 @987654329 @'s win32event 如果您的套接字已准备好接收 read/write 数据，则会提醒您。但是他们都不允许我在stdin/stdout上读写，给出错误，例如
An operation is performend on something that is not a socket
Handles only expect integer values等。
我还没有尝试过其他一些库，例如 twister。
现在，为了在 Windows 平台上实现上述代码中的功能，我使用了threads。这是我的代码：
stdin_buf = BytesIO(json.dumps(fn) + "\n")
stdout_buf = BytesIO()
stderr_buf = BytesIO()
rselect = [nodejs.stdout, nodejs.stderr] # type: List[BytesIO]
wselect = [nodejs.stdin] # type: List[BytesIO]
READ_BYTES_SIZE = 512
# creating queue for reading from a thread to queue
input_queue = Queue.Queue()
output_queue = Queue.Queue()
error_queue = Queue.Queue()
# To tell threads that output has ended and threads can safely exit
no_more_output = threading.Lock()
no_more_output.acquire()
no_more_error = threading.Lock()
no_more_error.acquire()
# put constructed command to input queue which then will be passed to nodejs's stdin
def put_input(input_queue):
while True:
sys.stdout.flush()
b = stdin_buf.read(READ_BYTES_SIZE)
if b:
input_queue.put(b)
else:
break
# get the output from nodejs's stdout and continue till otuput ends
def get_output(output_queue):
while not no_more_output.acquire(False):
b=os.read(nodejs.stdout.fileno(), READ_BYTES_SIZE)
if b:
output_queue.put(b)
# get the output from nodejs's stderr and continue till error output ends
def get_error(error_queue):
while not no_more_error.acquire(False):
b = os.read(nodejs.stderr.fileno(), READ_BYTES_SIZE)
if b:
error_queue.put(b)
# Threads managing nodejs.stdin, nodejs.stdout and nodejs.stderr respectively
input_thread = threading.Thread(target=put_input, args=(input_queue,))
input_thread.start()
output_thread = threading.Thread(target=get_output, args=(output_queue,))
output_thread.start()
error_thread = threading.Thread(target=get_error, args=(error_queue,))
error_thread.start()
# mark if output/error is ready
output_ready=False
error_ready=False
while (len(wselect) + len(rselect)) > 0:
try:
if nodejs.stdin in wselect:
if not input_queue.empty():
os.write(nodejs.stdin.fileno(), input_queue.get())
elif not input_thread.is_alive():
wselect = []
if nodejs.stdout in rselect:
if not output_queue.empty():
output_ready = True
stdout_buf.write(output_queue.get())
elif output_ready:
rselect = []
no_more_output.release()
no_more_error.release()
output_thread.join()
if nodejs.stderr in rselect:
if not error_queue.empty():
error_ready = True
stderr_buf.write(error_queue.get())
elif error_ready:
rselect = []
no_more_output.release()
no_more_error.release()
output_thread.join()
error_thread.join()
if stdout_buf.getvalue().endswith("\n"):
rselect = []
no_more_output.release()
no_more_error.release()
output_thread.join()
except OSError as e:
break
所以对我来说最好的选择是线程。如果您想了解更多，这篇文章将是nice read。