主页 / 问题 / 875976
Accepted
E1mir
Asked:2020-09-01 01:55:06 +0000 UTC2020-09-01 01:55:06 +0000 UTC

Python 3 线程中的队列

  • 772

研究多线程，我写了一个小代码，理论上应该是依次启动线程，写了这么一个小脚本：

import queue
import threading

exit_flag = False


class MyThread(threading.Thread):

    def __init__(self, thread_id, name, q):
        threading.Thread.__init__(self)
        self.thread_id = thread_id
        self.name = name
        self.q = q

    def run(self):
        print("Starting {}".format(self.name))
        process_data(self.name, self.q)
        print("Exiting {}".format(self.name))


def process_data(working_thread, q):
    while not exit_flag:
        queue_lock.acquire()
        if not work_queue.empty():
            data = q.get()
            queue_lock.release()
            print("{} processing {}".format(working_thread, data))


thread_list = ["Thread-1", "Thread-2", "Thread-3"]
name_list = ["One", "Two", "Three", "Four", "Five"]

queue_lock = threading.Lock()
work_queue = queue.Queue(10)
threads = []
t_id = 1

for t_name in thread_list:
    thread = MyThread(t_id, t_name, work_queue)
    thread.start()
    threads.append(thread)
    t_id += 1

queue_lock.acquire()
for word in name_list:
    work_queue.put(word)
queue_lock.release()

while not work_queue.empty():
    pass

exit_flag = True

for t in threads:
    t.join()

print("Main thread end!")

最有趣的是，在调试模式下工作时，一切正常，但如果你只是让它运行，那么什么都不会发生，只有线程正在运行的消息。你能告诉我哪里出错了吗？

python

3 个回答

  1. Best Answer
    import queue, threading, time

ResultDt = {}
ResultLock = threading.Lock()


class MyThread(threading.Thread):
    def __init__(self, q, name):
        threading.Thread.__init__(self, name=name, daemon=True)
        self.q = q

    def run(self):
        print("Starting {}".format(self.name))
        while True:  # циклически получать задачи в потоке
            try:
                data = self.q.get(timeout=2)  # ожидать задачу в течение 2 сек
                print("{} processing {}".format(self.name, data))
            except queue.Empty:
                print('timeout выход', self.name)
                break
            else:  # из очереди получена задача
                try:
                    if data is None:
                        print('None выход', self.name)
                        break
                    else:
                        r = worker(data)  # выполнение задачи
                        ResultLock.acquire()  # заблокировать
                        ResultDt['run_{}'.format(data)] = r  # сохр. результат выполнения
                        ResultLock.release()  # освободить
                finally:  # task_done, при любом исходе
                    self.q.task_done()
        print("Exiting {}".format(self.name))


def worker(arg):
    """выполнение задач"""
    time.sleep(.5)
    return 'done_{}'.format(arg)


if __name__ == '__main__':
    work_queue = queue.Queue()  # очередь задач
    threads = [MyThread(work_queue, t_name) for t_name in ("Thread-1", "Thread-2", "Thread-3")]  # потоки
    for thread in threads:
        thread.start()

    for word in ["One", None, "Two", "Three", "Four", "Five"]:  # задачи
        work_queue.put(word)
    work_queue.join()  # заблокировать до выполнения всех задач

    time.sleep(3)  # ожидать более timeout
    print(ResultDt)
    print("Main thread end!")

    出去

    Starting Thread-1
Starting Thread-2
Starting Thread-3
Thread-1 processing One
Thread-2 processing None
None выход Thread-2
Thread-3 processing Two
Exiting Thread-2
Thread-1 processing Three
Thread-3 processing Four
Thread-1 processing Five
timeout выход Thread-3
Exiting Thread-3
timeout выход Thread-1
Exiting Thread-1
{'run_One': 'done_One', 'run_Two': 'done_Two', 'run_Three': 'done_Three', 'run_Four': 'done_Four', 'run_Five': 'done_Five'}
Main thread end!
    • 2

  2. 根据接受的答案，您想启动3一个线程并处理5其中的单词：

    #!/usr/bin/env python
import time
from multiprocessing.pool import ThreadPool

def handle_word(word):
    time.sleep(5)
    return 'run_' + word, 'done_' + word

result = dict(ThreadPool(3).map(handle_word, ["One", "Two", "Three", "Four", "Five"]))

    线程池本身内部使用了一个队列，提供了一个简单的接口。

    • 2

  3. 与其阻塞，threading.Lock()不如使用以下可能性Queue：

    Queue.task_done()

    指示以前排队的任务已完成。由队列消费者线程使用。对于用于获取任务的每个 get()，对 task_done() 的后续调用会告诉队列该任务的处理已完成。

    如果 join() 当前处于阻塞状态，它将在处理完所有项目后恢复（这意味着对于已将 put() 放入队列的每个项目都收到了 task_done() 调用）。

    如果调用的次数多于队列中放置的项目，则引发 ValueError。

    文档：https ://docs.python.org/3.6/library/queue.html

    还有例子

    • 1

相关问题