暂停两个Python线程，而第三个线程执行任务（使用锁？）

Question

我不熟悉并发编程。在

我想重复执行三个任务。前两个应该一直运行，第三个大约每小时运行一次。前两个任务可以并行运行，但我总是想在第三个任务运行时暂停它们。在

以下是我尝试过的方法：

import threading
import time

flock = threading.Lock()
glock = threading.Lock()

def f():
    while True:
        with flock:
            print 'f'
            time.sleep(1)

def g():
    while True:
        with glock:
            print 'g'
            time.sleep(1)

def h():
    while True:
        with flock:
            with glock:
                print 'h'
        time.sleep(5)

threading.Thread(target=f).start()
threading.Thread(target=g).start()
threading.Thread(target=h).start()

我希望这段代码每秒钟输出一个f和一个g，大约每五秒钟打印一个h。但是，当我运行它时，大约需要12个f和12个g才开始看到一些h。看起来前两个线程不断地释放并重新获取它们的锁，而第三个线程则不在循环中。在

1. 为什么？当第三个线程试图获取一个当前持有的锁，然后它被释放时，获取是否应该立即成功，而不是第一个/第二个线程立即再次获取它？我可能误解了什么。在
2. 什么是实现我想要的东西的好方法？在

注意：在这个简单的例子中，将time.sleep(1)调用移出with flock/glock块是可行的，但显然不适用于我的实际应用程序，因为线程大部分时间都在执行实际操作。在每次执行循环体之后，当前两个线程睡眠一秒钟时，释放锁，第三个任务仍然无法执行。在

Answer 1

用threading.Events做这个怎么样：

import threading
import time
import logging

logger=logging.getLogger(__name__)

def f(resume,is_waiting,name):
    while True:
        if not resume.is_set():
            is_waiting.set()
            logger.debug('{n} pausing...'.format(n=name))
            resume.wait()
            is_waiting.clear()
        logger.info(name)
        time.sleep(1)

def h(resume,waiters):
    while True:
        logger.debug('halt') 
        resume.clear()
        for i,w in enumerate(waiters):
            logger.debug('{i}: wait for worker to pause'.format(i=i))
            w.wait()
        logger.info('h begin')
        time.sleep(2)
        logger.info('h end')        
        logger.debug('resume')
        resume.set()
        time.sleep(5)

logging.basicConfig(level=logging.DEBUG,
                    format='[%(asctime)s %(threadName)s] %(message)s',
                    datefmt='%H:%M:%S')

# set means resume; clear means halt
resume = threading.Event()
resume.set()

waiters=[]
for name in 'fg':
    is_waiting=threading.Event()
    waiters.append(is_waiting)
    threading.Thread(target=f,args=(resume,is_waiting,name)).start()    
threading.Thread(target=h,args=(resume,waiters)).start()

收益率

^{pr2}$

---

（针对注释中的一个问题）此代码尝试测量h-线程从其他工作线程获取每个锁所需的时间。在

这似乎表明，即使h正在等待获取锁，另一个工作线程也可能以相当高的概率释放并重新获取该锁。 没有给h优先权，因为它等待的时间更长。在

davidbeazley在PyCon上介绍了与线程和GIL相关的问题。这是一个pdf of the slides。这是一本引人入胜的读物，可能也有助于解释这一点。在

import threading
import time
import logging

logger=logging.getLogger(__name__)

def f(lock,n):
    while True:
        with lock:
            logger.info(n)
            time.sleep(1)

def h(locks):
    while True:
        t=time.time()
        for n,lock in enumerate(locks):
            lock.acquire()
            t2=time.time()
            logger.info('h acquired {n}: {d}'.format(n=n,d=t2-t))
            t=t2
        t2=time.time()
        logger.info('h {d}'.format(d=t2-t))
        t=t2
        for lock in locks:
            lock.release()
        time.sleep(5)

logging.basicConfig(level=logging.DEBUG,
                    format='[%(asctime)s %(threadName)s] %(message)s',
                    datefmt='%H:%M:%S')

locks=[]
N=5
for n in range(N):
    lock=threading.Lock()
    locks.append(lock)
    t=threading.Thread(target=f,args=(lock,n))
    t.start()

threading.Thread(target=h,args=(locks,)).start()

Answer 2

最简单的方法是使用3个Python进程。如果您在Linux上执行此操作，则每小时进程可以发送一个信号以使其他任务暂停，或者您甚至可以杀死它们，然后在每小时一次的任务完成后重新启动。不需要螺纹。在

但是，如果您决定使用线程，那么尝试在线程之间共享NO数据，只需来回发送消息（也称为数据复制，而不是数据共享）。穿线很难做好。在

但是，多个进程迫使您什么都不共享，因此更容易正确地执行。如果您使用像0MQhttp://www.zeromq.org这样的库来传递消息，那么从线程模型迁移到多进程模型是很容易的。在

Answer 3

使用通信进行同步：

#!/usr/bin/env python
import threading
import time
from Queue import Empty, Queue

def f(q, c):
    while True:
        try: q.get_nowait(); q.get() # get PAUSE signal      
        except Empty: pass  # no signal, do our thing
        else: q.get()       # block until RESUME signal
        print c,
        time.sleep(1)

def h(queues):
    while True:
        for q in queues:
            q.put_nowait(1); q.put(1) # block until PAUSE received
        print 'h'
        for q in queues:
            q.put(1) # put RESUME
        time.sleep(5)

queues = [Queue(1) for _ in range(2)]
threading.Thread(target=f, args=(queues[0], 'f')).start()
threading.Thread(target=f, args=(queues[1], 'g')).start()
threading.Thread(target=h, args=(queues,)).start()

从你的表现来看，它可能不是最佳的，但我发现它更容易遵循。在

输出

^{pr2}$