Python多线程问题

Question

代码：

#!/usr/bin/env python

import os
import sys
import threading

class fifo_buffer(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.dict_buffer = {}

    def run(self):
        path =  "/media/ramdisk/sample.fifo"
        fifo = open(path, "r")
        count = 0
        for line in fifo:
            count = count + 1
            self.dict_buffer[count] = line
            #print str(count) + " -> " + self.dict_buffer[count]
        fifo.close()

    def get_packet(self, index):
        return self.dict_buffer[index]

    def len_dict(self):
        print "Length dict: " + str(len(self.dict_buffer))

def main():
   fb = fifo_buffer()
   fb.start()
   print "get_packet(2): " + fb.get_packet(2)   # Input Error
   fb.len_dict()                                # Length showing zero

if __name__ == "__main__":
   main()

1. 运行函数：

   我正在尝试从 fifo 文件中读取数据并将数据存储到字典中。 此运行函数将几乎无限执行。

2. get_packet 或 len_dict 函数：

   我正在尝试获取字典的详细信息，但无法这样做。

我的疑问是 run 函数何时将数据从 fifo 文件存储到字典，而假设将无限进行，使用 get_packet 或 len_dict 我想访问此字典数据结构，但无法访问它

我是线程新手，因此也欢迎任何其他方法。

Answer 1

我认为问题可能在于您没有弄清楚多线程是如何工作的。 调用fb.start()后，两个线程（主线程和run()线程从fifo文件中读取数据并存入字典）同时工作。

调用print "get_packet(2): " + fb.get_packet(2)时遇到的KeyError是因为当主线程调用get_packet(2)时，run()线程还没有将数据追加到字典中。

所以只有在调用 join() 之后，才能访问字典数据？？

不完全是。请继续阅读。

当 run() 线程无限执行时，他们有什么方法可以访问字典吗？

您可以在 run() thread 执行时处理/访问 run() thread 中的字典，但显然该字典会随着时间而变化，您只能访问 run() thread 已经处理过的行.您还可以在执行run() thread 时在主线程 中处理/访问字典，但正如我之前所说，run() thread 可能没有将数据附加到字典中（您只能访问这些行run() thread 也已处理）。

我已经测试了代码，添加了一个新方法并进行了一些修改。它运作良好。修改后的代码如下。

#!/usr/bin/env python

import os
import sys
import threading

class fifo_buffer(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.dict_buffer = {}

    def run(self):
        print("run() starts.")
        path = "./sample.fifo"
        fifo = open(path, "r")
        count = 0
        for line in fifo:
            count = count + 1
            self.dict_buffer[count] = line
            print("in for expression: before show_dict()")
            self.show_dict()    # you could process dict_buffer here, but obviously this dict changes over time, and you could only access the lines that run() thread has already processed.
            #print str(count) + " -> " + self.dict_buffer[count]
        fifo.close()
        print("run() completes.")

    def get_packet(self, index):
        return self.dict_buffer[index]

    def len_dict(self):
        print("Length dict: " + str(len(self.dict_buffer)))

    def show_dict(self):
        print(self.dict_buffer)

def main():
   fb = fifo_buffer()
   fb.start()
   # fb.join()
   # print("get_packet(2): " + fb.get_packet(2))
   # you could process dict_buffer here, but you could only process/access the lines which run() thread has already processed.
   fb.len_dict()
   fb.show_dict()

if __name__ == "__main__":
   main()

Answer 2

正如我在我的一个 cmets 中所说，这个示例（尤其是 main 函数）确实没有从多线程的 PoV 中理解。这是代码“稍微”修改：

#!/usr/bin/env python

import os
import sys
import threading


class FifoBufferThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.dict_buffer = {}

    def run(self):
        print("Begin inner thread run...")
        path = "/media/ramdisk/sample.fifo"
        count = 0
        with open(path, "r") as fifo:
            for line in fifo:
                count += 1
                self.dict_buffer[count] = line
                #print str(count) + " -> " + self.dict_buffer[count]
        print("End inner thread run")

    def get_packet(self, index):
        return self.dict_buffer.get(index, None)

    def get_dict_len(self):
        return len(self.dict_buffer)

    def len_dict(self):
        print "Length dict: " + str(self.get_dict_len())


def main():
    print("Main thread starting...")
    fbt = FifoBufferThread()
    fbt.daemon = True
    fbt.start()
    user_selection = None
    while True:
        user_selection = raw_input("Press any key to continue, 'x' (lowercase) to exit: ")
        if user_selection == "x":
            print("Exiting at user's choice.")
            break
        print(fbt.len_dict())
        # Do whatever else you might think is necessary with the dict
    print("Exiting")


if __name__ == "__main__":
    main()

现在，可以看到 main 线程与内部 (FifoBufferThread) 线程交互。

变化：

• main 函数现在是一个循环（在 启动 fbt 线程之后），以更好地说明线程在后台执行的操作（以及 fbt 如何影响主线程）。它涉及用户交互（需要按下一个键），如果该键是小写x，程序将退出。
  重要提示：当程序等待用户输入时，内部线程将继续工作，因此它会从文件中读取行
• !!!重要的一点！！！：fbt.daemon = True：线程被标记为守护进程。正如[Python]: Thread objects 所说：

  可以将线程标记为“守护线程”。这个标志的意义在于，当只剩下守护线程时，整个 Python 程序就退出了。

• get_packet：我将其修改为使用[Python]: dict.get（并返回None，而不是引发异常并终止程序）
• 添加了get_dict_len 函数（现在由len_dict 使用）。最好先获取数据，然后再将其格式化以用于打印目的

• 我不确定文件的读取速率，或者其他程序写入文件的速率，也不确定它的初始内容，但是会发生以下情况之一（假设一旦文件打开以供读取，它将在其他进程写入时“更新”其内容）：

  1. 您的应用将“吃掉”文件中的所有内容并最终到达 EOF，并且线程将停止（正常情况）

  2. 从 TCP 读取的另一个应用程序将写入足够快（不太可能），以便您的应用程序读取文件内容（并将其存储在字典中：IN MEMORY) 最终会耗尽内存

     2.1。一种味道：RAM 不会用完，但磁盘 (/media) 会

• 注意：还有更多可以改进的地方