pynsq太烂了
起初是因为需要在 Web API 里消费一个消息队列(nsq),给浏览器返回 EventSource 做实时输出。但是没想到官方库pynsq居然写得这么渣:
def _handle_term_signal(sig_num, frame):
logging.getLogger(__name__).info(
'TERM Signal handler called with signal %r', sig_num)
tornado.ioloop.IOLoop.current().stop()
def run():
"""
Starts any instantiated :class:`nsq.Reader` or :class:`nsq.Writer`
"""
signal.signal(signal.SIGTERM, _handle_term_signal)
signal.signal(signal.SIGINT, _handle_term_signal)
tornado.ioloop.IOLoop.current().start()
这都什么玩意儿?一个库何德何能也敢去乱劫持 signal?最他妈逗的是这个库的推荐写法:
import nsq
r = nsq.Reader(message_handler=handler,
lookupd_http_addresses=['http://127.0.0.1:4161'],
topic='nsq_reader', channel='asdf', lookupd_poll_interval=15)
nsq.run()
这个 r
莫名其妙的就能跑起来了。真是魔法。anyway,在Web API兼容这一坨是失败了,爆了一堆错,对付魔法的办法就是隔离,所以打算糊个脚本,通过多进程+PIPE来得到数据。
给 subprocess 加一个异步补丁
糊的时候发现,subprocess
模块的 .communicate()
方法是 blocking的, 居然不支持 async/await,于是只能继续糊:
import subprocess, fcntl, asyncio
p = subprocess.Popen(
cmd, bufsize=0, text=True, stdin=subprocess.PIPE,
stderr=subprocess.PIPE, stdout=subprocess.PIPE, close_fds=True)
fcntl.fcntl(p.stdout.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
buf = ''
while p.poll() is None:
try:
# read() with O_NONBLOCK might gets None thus TypeError
buf += (p.stdout.read(1024) or '')
contents = buf.split('\n')
if len(contents) < 2:
raise TypeError('Not enough')
except TypeError:
await asyncio.sleep(1)
yield None
continue
buf = contents.pop(-1)
for x in contents:
yield x
p.terminate()
这里用 buf
、contents
是因为直接调用消息队列,每个消息都是分割好的;但是通过管道来读。得自己去处理字节流如何分割的问题,也就是传说简中特供的「粘包」问题。
另外子进程可能因为Web框架的种种原因,不自觉退出或者变僵尸进程,需要向 stdout/stderr 做输出的时候检查一下 BrokenPipeError 处理后事主动退出即可。
还有子进程如果输出完毕,需要代码里调用一下 p.terminate()
,才能回收,避免变zombie
试了一下,也不是不能用,成功!正觉得自己又行了的时候,发现官方:
https://docs.python.org/3.8/library/asyncio-subprocess.html
啊这。白忙活。
超长待机 subprocess
想起来之前还有个问题,一个API调用一个subprocess,然后等它执行,不想管。这个时候可以加一个参数 detach 模式:
subprocess.Popen(..., start_new_session=True)
然后在同进程用里跑一个后台检查循环去检查子进程的 WNOHANG
:
while 1:
try:
chpid, retcode, res = os.wait3(os.WNOHANG)
except ChildProcessError:
break
sleep(5)
if chpid == os.getpid():
do_sth()
break
想了下好像也搞复杂了。要不直接跑一个 sh -c "blah.sh &"
算了。