python - 当一个线程正在运行时阻塞其他线程

Question

假设我有两种类型的线程，

1. 每 x 分钟运行一次的单线程。我们称之为线程

2. 多线程一直在运行。当 A 线程 do_something() 时，B 线程我希望所有 B 线程等待 A 完成然后恢复它们。我不知道该用什么。

我尝试使用threading.Condition、wait()/notifyAll()但它没有按我想要的方式工作。一旦我放入条件，它就会像同步线程或其他东西一样进行 1 1 的处理。我想让他们自由地奔跑。

这是我尝试将它们放入 wait() 的示例代码，然后通知它们，但它像 join() 一样 1 接 1 地执行。不知道对我们来说是什么。

class ...
check = True
def xxx(self,g,con):
  for i in range(3):
    with con:
      if self.check:
        con.wait()
      self.check = False
      time.sleep(3)
      print(g)

con = threading.Condition()
threading.Thread(target=xxx,args=('a',con,)).start()
threading.Thread(target=xxx,args=('b',con,)).start()
threading.Thread(target=xxx,args=('c',con,)).start()
time.sleep(2)
con.notifyAll()

Answer 1

Question: Blocking other Threads while one Thread is running

此示例使用 threading.Barrier(...)，而不是使用 threading.Condition()。

<小时/>

使用来自 docs.python.org 的模块:

• module-threading
• event-objects
• barrier-objects

<小时/>

import time, threading
from threading import BrokenBarrierError

def worker_A(g, terminate, barrier):
    # Counter to simulate conditional workload
    do_something = 3

    while not terminate.is_set():
        if do_something == 0:
            # Reset the barrier and wait until n_waiting == 2
            barrier.reset()
            while not terminate.is_set() and barrier.n_waiting < 2:
                time.sleep(0.5)

            # Now the other Threads waiting at the barrier
            # Simulate worklaod ...
            print('worker_A barrier.broken={} n_waiting={}'
                .format(barrier.broken, barrier.n_waiting))
            time.sleep(3)

            # Call the third barrier.wait to release the barrier
            try:
                barrier.wait()
            except BrokenBarrierError:
                pass

            # Reset counter to restart simulate conditional workload
            do_something = 3
        else:
            # Count down and give the other threads a timeslice
            do_something -= 1
            time.sleep(0.5)

def worker_B(g, terminate, barrier):
    while not terminate.is_set():
        # Simulate workload ...
        print('worker_B({})'.format(g))
        time.sleep(1)

        # Block at barrier.wait() if the barrier is NOT in the broken state
        try:
            barrier.wait()
        except BrokenBarrierError:
            pass

if __name__ == "__main__":
    # Event to terminate all Threads save
    terminate = threading.Event()

    # Barrier to block worker_B Threads
    # We use 3 Threads, therefore init with parties=3
    barrier = threading.Barrier(3)
    barrier.abort()

    # Create and start the Threads
    threads = []
    for t in [(worker_A, 'a'), (worker_B, 'b'), (worker_B, 'c'), ]:
        threads.append(threading.Thread(target=t[0], args=(t[1], terminate, barrier,)))
        threads[-1].start()
        time.sleep(0.2)

    # Simulating MAIN Thread
    time.sleep(20)

    # Set the `terminate` Event to True, 
    # and abort the barrier to force all Threads to terminate
    print('Terminate...')
    terminate.set()
    barrier.abort()

    # Wait until all Threads terminated
    for t in threads:
        t.join()

    print('EXIT MAIN')

使用 Python 测试:3.5