python - 从python中的另一个线程从 sleep 中唤醒特定线程

Question

几天来我一直在追逐这个，我快疯了。我是一个完全的业余爱好者，对 Python 完全陌生，所以请原谅我的愚蠢。

我的主线程反复检查数据库中的条目，然后为它在数据库中找到的每个新条目启动线程。

它启动的线程基本上轮询数据库以获取一个值，如果找不到该值，它会做一些事情，然后休眠 60 秒并重新开始。

启动线程的简化非代码

while True:
    stop = _Get_a_Value_From_Database_for_Exit() #..... a call to DBMS 
    If stop = 0:
        Do_stuff()
        time.sleep(60)
    else:
        break

在任何给定时间可能有许多这样的线程在运行。我想要做的是让主线程检查数据库中的另一个位置的特定值，然后可以在启动的特定线程中中断上面示例中的 sleep 。目标是退出一个特定的线程，如上面列出的线程，而不必等待剩余的 sleep 持续时间。所有这些线程都可以通过共享的数据库 ID 来引用。我已经看到对 event.wait() 和 event.set() 的引用，一直试图弄清楚我如何用它替换 time.sleep()，但我不知道如何使用它来唤醒特定线程而不是所有线程。

这就是我无知的地方:有没有一种方法可以让我根据数据库 id 为 event.wait 做一些事情(比如 12345.wait(60) 在启动线程中，12345.set() 在主线程中(所有动态都基于不断变化的数据库 id)。

感谢您的帮助!

Answer 1

这个项目有点复杂，这是我的版本。

• 扫描数据库文件/tmp/db.dat，预填两个词

• manager:为每个单词创建一个线程；默认是一个“威士忌”线程和一个“糖浆”线程

• 如果单词以 _stop 结尾，例如 syrup_stop，通过设置其停止事件告诉该线程结束

• 每个线程扫描数据库文件并在看到单词stop 时退出。如果设置了停止事件，它也会退出。

• 请注意，如果 Manager 线程设置了 worker 的 stop_event，则该 worker 将立即退出。每个线程做一些事情，但大部分时间都花在 stop_ev.wait() 调用上。因此，当事件确实设置时，它不必等待，它可以立即退出。

服务器很有趣!启动它，然后通过向数据库添加行来向它发送命令。尝试以下每一项:

$ echo pie >> /tmp/db.dat  # start new thread

$ echo pie_stop >> /tmp/db.dat # stop thread by event

$ echo whiskey_stop >> /tmp/db.dat # stop another thread "

$ echo stop >> /tmp/db.dat # stop all threads

来源

import logging, sys, threading, time

STOP_VALUE = 'stop'

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)-4s %(threadName)s %(levelname)s %(message)s", 
    datefmt="%H:%M:%S",
    stream=sys.stderr,
)


class Database(list):
    PATH = '/tmp/db.dat'
    def __init__(self):
        super(Database,self).__init__()
        self._update_lock = threading.Lock()

    def update(self):
        with self._update_lock:
            self[:] = [ line.strip() for line in open(self.PATH) ]

db = Database()

def spawn(events, key):
    events[key] = threading.Event()
    th = threading.Thread(
        target=search_worker,
        kwargs=dict(stop_ev=events[key]),
        name='thread-{}'.format(key),
    )
    th.daemon = True
    th.start()

def search_worker(stop_ev):
    """
    scan database until "stop" found, or our event is set
    """
    logging.info('start')
    while True:
        logging.debug('scan')
        db.update()
        if STOP_VALUE in db:
            logging.info('stopvalue: done')
            return
        if stop_ev.wait(timeout=10):
            logging.info('event: done')
            return 

def manager():
    """
    scan database
    - word: spawn thread if none already
    - word_stop: tell thread to die by setting its stop event
    """
    logging.info('start')
    events = dict()
    while True:
        db.update()
        for key in db:
            if key == STOP_VALUE:
                continue
            if key in events:
                continue
            if key.endswith('_stop'):
                key = key.split('_')[0]
                if key not in events:
                    logging.error('stop: missing key=%s!', key)
                else:
                    # signal thread to stop
                    logging.info('stop: key=%s', key)
                    events[key].set()
                    del events[key]
            else:
                spawn(events, key)
                logging.info('spawn: key=%s', key)
        time.sleep(2)


if __name__=='__main__':

    with open(Database.PATH, 'w') as dbf:
        dbf.write(
        'whiskey\nsyrup\n'
        )

    db.update()
    logging.info('start: db=%s -- %s', db.PATH, db)

    manager_t = threading.Thread(
        target=manager,
        name='manager',
    )
    manager_t.start()
    manager_t.join()