为什么`stdin.read()`不读取整个缓冲区？

Question

我有以下这段代码：

def get_input(self):
    """
    Reads command from stdin, returns its JSON form
    """
    json_string = sys.stdin.read()
    print("json string is: "+json_string)
    json_data =json.loads(json_string)
    return json_data

def accept_commands(self):
    while True:
        json_data = self.get_input()
        command = self.command_analyzer.is_command(json_data) # Check wether the command exists. Return it if it does
        #The command exists
        if command is not None:
            #The addon is not currently active
            if analyzer.intent(json_data) not in self.active_addons:
                self.activate_addon(command,json_data)

            #The addon is active and so we need to send the data to the subprocess
            else:
                self.communicate_with_addon(command,json_data,json_string)

这段代码读取了一个从其他程序发送过来的json字符串。这个json是从标准输入（stdin）读取的。 但是我不知道为什么，输出结果是这样的：

json string is: <Some json here>
json string is: 
Traceback (most recent call last):
  File "/Users/Matan/Documents/workspace/ProjectSH/addonmanager/addon_manager.py", line 63, in <module>
    manager.accept_commands()
  File "/Users/Matan/Documents/workspace/ProjectSH/addonmanager/addon_manager.py", line 49, in accept_commands
    json_data = self.get_input()
  File "/Users/Matan/Documents/workspace/ProjectSH/addonmanager/addon_manager.py", line 42, in get_input
    json_data =json.loads(json_string)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/json/__init__.py", line 338, in loads
    return _default_decoder.decode(s)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/json/decoder.py", line 365, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/json/decoder.py", line 383, in raw_decode
    raise ValueError("No JSON object could be decoded")

这个json是从以下地方发送过来的：

class MessageReceiver:

    def __init__(self):
        '''
        Connect to the AMQP broker and starts listening for messages.
        Creates the a Popen object to pass command info to the addon_manager script (which
        is in charge of managing scripts)
        '''
        addon_manager_path = configuration.addon_manager_path()
        addon_manager_path = os.path.join(addon_manager_path,'addon_manager.py')
        execute = "python " + addon_manager_path
        self.addon_manager = subprocess.Popen(execute, stdin=subprocess.PIPE, shell=True)


        self.component_name= configuration.get_attribute("name")

        if len(sys.argv)>1:
            host_ip = sys.argv[1]
        else:
            host_ip = 'localhost'

        #Start a connection to the AMQP server
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=host_ip))

        #Create a channel to the server
        self.channel = self.connection.channel()

        self.channel.queue_declare(queue="kitchen")

        #callback method to be called when data is received
        #It sends the data that is received by the client to the addon_manager
        def data_received(ch, method, properties, body):
            ##TODO: Might want to add some checks. Is body a JSON? etc.
            print("writing data to addon_manager")
            self.addon_manager.communicate(body)


        self.channel.basic_consume(data_received,queue='kitchen',no_ack=True)


        self.channel.start_consuming()

这里面有什么问题吗？

Answer 1

看起来你在调用get_input这个函数的时候，调用了两次，而第二次的时候缓冲区里没有任何内容。你应该在尝试解析JSON之前，先检查一下缓冲区里是否有东西：

def get_input(self):
    """
    Reads command from stdin, returns its JSON form
    """
    json_string = sys.stdin.read()
    if json_string:
        print("json string is: "+json_string)
        json_data =json.loads(json_string)
        return json_data
    else:
        return None

Answer 2

默认情况下，stdin.read() 会一直等待，直到把整个缓冲区都读完。如果你只解码到了一个 JSON 对象，那说明在 stdin 被另一个进程关闭之前，只发送了这个对象。

如果你需要处理多个 JSON 数据块，你应该：

a) 不要在写入的进程中关闭流，

b) 在 Python 中不要进行阻塞读取。

可以尝试分块或逐行读取；具体方法可以参考 如何在 Python 中加载和解析包含多个 JSON 对象的文件 和 如何使用 'json' 模块一次读取一个 JSON 对象？，这些链接提供了逐行读取或使用多行的 JSON 对象的技巧。

你可以根据这些方法来创建一个生成器函数；你可以循环调用这个函数，每次返回一个 JSON 对象，中间会阻塞等待下一个 JSON 对象：

def get_input(self):
    for line in sys.stdin:
        yield json.loads(line)

还有

def accept_commands(self):
    for json_data in self.get_input():
        # do something with `json_data`

你正在使用 Popen.communicate() 来写入管道。这个方法会在写入后关闭管道，然后等待进程结束。

如果你希望管道保持打开状态，就不要使用 Popen.communicate，而是直接写入 Popen.stdin 管道。