为什么我在使用threading.Lock和'with‘时会出现线程匮乏？(Python 3.6.6)

Question

tl;dr

使用threading.Lock和with会导致线程匮乏(一个线程完成全部或大部分工作)：

with lock:
    next(generator)

但是手动调用acquire()和release()可以在线程之间更均匀地分配工作：

lock.acquire()
next(generator)
lock.release()

完整的代码和输出示例

使用threading.Lock()和with进行同步

import threading
import queue
import time
import random

class ThreadedGen:
  def __init__(self, generator):
    self._generator = generator
    self._queue = queue.Queue(10)
    self._lock  = threading.Lock()

  def _run(self):
    while True:
      with self._lock: # <==== use with, no explicit call to acuire()/release()
        try:
          # get the next item from the generator
          # put the thread name and item as a tuple on the queue
          n = next(self._generator)
          self._queue.put((threading.current_thread().name, n))
        except Exception as e:
          print(threading.current_thread().name, e)

  def start(self):
    for _ in range(3): # spawn three threads
      threading.Thread(target=self._run).start()

  def get(self):
    while True:
      yield self._queue.get()

if __name__ == '__main__':
  def gen():
    n = 0
    while n < 100:
      s = random.randint(0,5)
      time.sleep(s) # simulate work
      yield n
      n += 1

  t = ThreadedGen(gen())
  t.start()
  outputgen = t.get()
  for _ in range(100):
    (tname, n) = next(outputgen)
    print(f'got {n} from {tname}')

输出：

got 0 from Thread-1
got 1 from Thread-1
got 2 from Thread-1
got 3 from Thread-1
got 4 from Thread-1
got 5 from Thread-1
got 6 from Thread-1
got 7 from Thread-1
got 8 from Thread-1
got 9 from Thread-1
got 10 from Thread-1
got 11 from Thread-1
got 12 from Thread-1
got 13 from Thread-1
got 14 from Thread-1
got 15 from Thread-1
got 16 from Thread-1
got 17 from Thread-1
got 18 from Thread-1
got 19 from Thread-1
got 20 from Thread-1
got 21 from Thread-1
got 22 from Thread-1
got 23 from Thread-1
got 24 from Thread-1
got 25 from Thread-1
got 26 from Thread-1
got 27 from Thread-1
got 28 from Thread-1
got 29 from Thread-1
got 30 from Thread-1
got 31 from Thread-1
got 32 from Thread-1
got 33 from Thread-1
got 34 from Thread-1
got 35 from Thread-1
got 36 from Thread-1
got 37 from Thread-1
got 38 from Thread-1
got 39 from Thread-1
got 40 from Thread-1
got 41 from Thread-1
got 42 from Thread-1
got 43 from Thread-1
got 44 from Thread-1
got 45 from Thread-1
got 46 from Thread-1
got 47 from Thread-3
got 48 from Thread-3
got 49 from Thread-3
got 50 from Thread-3
...

使用threading.Lock()和显式acquire() release()进行同步

import threading
import queue
import time
import random

class ThreadedGen:
  def __init__(self, generator):
    self._generator = generator
    self._queue = queue.Queue(10)
    self._lock  = threading.Lock()

  def _run(self):
    while True:
      try:
        # get the next item from the generator
        # put the thread name and item as a tuple on the queue
        self._lock.acquire() # <==== explicit acquire() call
        n = next(self._generator)
        self._lock.release() # <==== explicit release() call
        self._queue.put((threading.current_thread().name, n))
      except Exception as e:
        self._lock.release()
        print(threading.current_thread().name, e)

  def start(self):
    for _ in range(3): # spawn three threads
      threading.Thread(target=self._run).start()

  def get(self):
    while True:
      yield self._queue.get()

if __name__ == '__main__':
  def gen():
    n = 0
    while n < 100:
      s = random.randint(0,5)
      time.sleep(s) # simulate work
      yield n
      n += 1

  t = ThreadedGen(gen())
  t.start()
  outputgen = t.get()
  for _ in range(100):
    (tname, n) = next(outputgen)
    print(f'got {n} from {tname}')

输出：

got 0 from Thread-1
got 1 from Thread-2
got 2 from Thread-3
got 3 from Thread-1
got 4 from Thread-2
got 5 from Thread-3
got 6 from Thread-1
got 7 from Thread-2
got 8 from Thread-3
got 9 from Thread-1
got 10 from Thread-1
got 11 from Thread-3
got 12 from Thread-2
got 13 from Thread-1
got 14 from Thread-3
got 15 from Thread-2
got 16 from Thread-1
got 17 from Thread-3
got 18 from Thread-2
got 19 from Thread-1
got 20 from Thread-3
got 21 from Thread-2
got 22 from Thread-1
got 23 from Thread-3
got 24 from Thread-2
got 25 from Thread-1
got 26 from Thread-3
got 27 from Thread-2
got 28 from Thread-1
got 29 from Thread-3
got 30 from Thread-2
got 31 from Thread-1
got 32 from Thread-3
got 33 from Thread-2
got 34 from Thread-2
got 35 from Thread-2
got 36 from Thread-3
got 37 from Thread-1
got 38 from Thread-2
got 39 from Thread-3
got 40 from Thread-1
got 41 from Thread-2
got 42 from Thread-2
got 43 from Thread-1
got 44 from Thread-3
got 45 from Thread-2
got 46 from Thread-1
got 47 from Thread-3
got 48 from Thread-2
got 49 from Thread-1
got 50 from Thread-1

Answer 1

当我输入我的问题时，我想出了这个问题(非常感谢！)问题是，在调用next(self._generator)之前，我并没有将with放在同一位置

错误：

  with self._lock:
    try:
      # get the next item from the generator
      # put the thread name and item as a tuple on the queue
      n = next(self._generator)
      self._queue.put((threading.current_thread().name, n))
    except Exception as e:
      print(threading.current_thread().name, e)

右图：

    try:
      # get the next item from the generator
      # put the thread name and item as a tuple on the queue
      with self._lock:
        n = next(self._generator)
      self._queue.put((threading.current_thread().name, n))
    except Exception as e:
      print(threading.current_thread().name, e)

如果队列已满，则默认情况下queue.put()会阻塞。这意味着锁不会被立即释放，从而阻止其他线程访问生成器，并在第一个线程可以重新获取锁之前竞相从第一个线程获取锁(所以这是一个竞争条件)。

摘要：

将with尽可能靠近同步的调用，以防止其他阻塞调用延迟其他线程对资源的访问