多线程检查队列中的成员资格并停止线程

我想使用 2 个线程迭代列表。一个来自前导，另一个来自尾随，并将元素放在一个Queue在每次迭代中。但在将值放入之前Queue我需要检查其中的值是否存在Queue（当其中一个线程将该值放入Queue），所以当发生这种情况时，我需要停止线程并返回每个线程的遍历值列表。

这是我到目前为止所尝试过的：

from Queue import Queue
from threading import Thread, Event

class ThreadWithReturnValue(Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs={}, Verbose=None):
        Thread.__init__(self, group, target, name, args, kwargs, Verbose)
        self._return = None
    def run(self):
        if self._Thread__target is not None:
            self._return = self._Thread__target(*self._Thread__args,
                                                **self._Thread__kwargs)
    def join(self):
        Thread.join(self)
        return self._return

main_path = Queue()

def is_in_queue(x, q):
   with q.mutex:
      return x in q.queue

def a(main_path,g,l=[]):
  for i in g:
    l.append(i)
    print 'a'
    if is_in_queue(i,main_path):
      return l
    main_path.put(i)

def b(main_path,g,l=[]):
  for i in g:
    l.append(i)
    print 'b'
    if is_in_queue(i,main_path):
      return l
    main_path.put(i)

g=['a','b','c','d','e','f','g','h','i','j','k','l']

t1 = ThreadWithReturnValue(target=a, args=(main_path,g))
t2 = ThreadWithReturnValue(target=b, args=(main_path,g[::-1]))
t2.start()
t1.start()
# Wait for all produced items to be consumed
print main_path.join()

I used ThreadWithReturnValue这将创建一个返回值的自定义线程。

为了进行成员资格检查，我使用了以下函数：

def is_in_queue(x, q):
   with q.mutex:
      return x in q.queue

现在如果我首先开始t1然后是t2我会得到12a然后一个b那么它不会做任何事情，我需要手动终止 python！

但如果我首先运行t2 then t1我会得到以下结果：

b
b
b
b
 ab

ab
b

b
b
 b
a
a

所以我的问题是为什么 python 在这种情况下表现不同？我怎样才能终止线程并使它们相互通信？

在我们遇到更大的问题之前，你没有使用Queue.join right.

该函数的全部要点是，将一堆项目添加到队列中的生产者可以等待，直到消费者完成对所有这些项目的处理。这是通过让消费者调用来实现的task_done当他们完成完成的每个项目后get。一旦有那么多task_done调用为put调用，队列完成。你不是在做一个get任何地方，更不用说task_done，所以队列不可能完成。所以，这就是为什么在两个线程完成后永远阻塞的原因。

这里的第一个问题是您的线程在实际同步之外几乎不做任何工作。如果他们唯一做的就是争夺队列，那么一次只有一个人能够运行。

当然，这在玩具问题中很常见，但你必须思考真正的问题：

• 如果您正在执行大量 I/O 工作（侦听套接字、等待用户输入等），则线程可以很好地工作。
• 如果您正在执行大量 CPU 工作（计算素数），则由于 GIL，线程在 Python 中无法工作，但进程可以。
• 如果您实际上主要处理同步单独的任务，那么任何一个都不会很好地工作（并且流程会更糟）。可能仍然是simpler从线程的角度思考，但这将是最慢的做事方式。您可能想研究一下协程；格雷格·尤因有一个很棒的示范如何使用yield from使用协程来构建诸如调度程序或多参与者模拟之类的东西。

接下来，正如我在上一个问题中提到的，使线程（或进程）在共享状态下高效工作需要尽可能短地持有锁。

因此，如果您必须在锁下搜索整个队列，那么最好是恒定时间搜索，而不是线性时间搜索。这就是为什么我建议使用类似的东西OrderedSet食谱而不是list，就像 stdlib 中的那样Queue.Queue。那么这个函数：

def is_in_queue(x, q):
   with q.mutex:
      return x in q.queue

…仅阻塞队列一小部分时间 - 只需足够长的时间来查找表中的哈希值，而不是足够长的时间来比较队列中的每个元素x.

最后，我试图解释你的另一个问题的竞争条件，但让我再试一次。

您需要锁定代码中每个完整的“事务”，而不仅仅是单个操作。

例如，如果您这样做：

with queue locked:
    see if x is in the queue
if x was not in the queue:
    with queue locked:
        add x to the queue

…那么当您检查时 x 总是可能不在队列中，但在您解锁它和重新锁定它之间的时间里，有人添加了它。这正是两个线程可能提前停止的原因。

要解决这个问题，您需要对整个事情加锁：

with queue locked:
    if x is not in the queue:
        add x to the queue

当然这直接违背了我之前所说的尽可能短的锁定队列的时间。事实上，简而言之，这就是多线程变得困难的原因。编写安全代码很容易，只需在可能需要的时间内锁定所有内容，但随后您的代码最终仅使用单个核心，而所有其他线程都被阻止等待锁定。编写快速代码很容易，只需尽可能简短地锁定所有内容，但这样就不安全了，你会得到垃圾值，甚至到处崩溃。弄清楚什么需要成为事务，以及如何最大程度地减少这些事务内的工作，以及如何处理您可能需要的多个锁，以使其工作而不造成死锁……这并不那么容易。