Pengurus konteks dan benang daemon

Kandungan soalan

Saya memulakan utas daemon daripada pengurus konteks yang sepatutnya menghantar degupan jantung setiap saat tetapi memandangkan ia berjalan dalam utas, ia tidak menamatkan pengurus konteks jika pengecualian berlaku. Bagaimana untuk menimbulkan pengecualian dalam pengurus konteks apabila degupan jantung berhenti?

from contextlib import contextmanager
from threading import Thread, Event
from time import sleep


@contextmanager
def plc():
    stop_event = Event()

    try:
        # Send heartbeat every second
        hb_t = Thread(target=heartbeat_task,
                      args=(stop_event,),
                      daemon=True)
        hb_t.start()

        yield
    except Exception:
        raise
    finally:
        stop_event.set()
        hb_t.join()
        print("Heartbeat stopped")


def heartbeat_task(stop_event):

    value = False
    
    while not stop_event.is_set():

        value = not value

        print("Heartbeat: " + str(value))

        sleep(1)


def main():

    with plc():

        while True:

            print("Program running")

            sleep(5)

if __name__ == '__main__':
    main()

Saya menghadapi masalah mencari contoh perkara ini.

Terima kasih atas bantuan anda!

Jawapan betul

dikemas kini

Saya telah mengubah suai kod supaya lebih konsisten dengan kod yang anda siarkan. Tetapi:

Kod yang anda berikan tidak konsisten: heartbeat_task 传递了一个事件，如果设置该事件将导致函数返回。但只有当使用 with plc(): 创建的函数 main 中的上下文管理器退出时才会设置它，而这是永远不会的。如果您希望 heartbeat_task 抛出的任何异常将强制上下文管理器退出，然后在函数 plc 中捕获，那么调用 stop_event.set() 的意义何在？如果根据定义，我们仅在 heartbeat_task Adakah anda tiba di sini hanya apabila ia tidak lagi wujud kerana pengecualian?

Jadi sama ada anda mahu heartbeat_task 无限期地运行，直到引发异常（在这种情况下，没有“停止”事件的意义），要么您希望能够在存在某些条件时停止 heartbeat_task，但没有这样做的代码。出于演示目的，我假设 main 将有权访问 stop_event 事件，并在某些情况下对其进行设置。否则，它会一直运行，直到检测到 heartbeat_task tidak akan berjalan lagi, mungkin kerana ia melemparkan pengecualian (ia melaksanakan gelung tak terhingga, jadi bagaimana mungkin ia ditamatkan jika acara berhenti belum ditetapkan lagi?). Selebihnya ialah sebab anda perlu menggunakan pengurus konteks. Saya akan mencadangkan alternatif kemudian.

Jika anda menggunakan kumpulan berbilang benang (kami hanya memerlukan satu utas dalam kumpulan), maka menjadi mudah untuk utas utama untuk menangkap pengecualian yang dilemparkan oleh tugasan yang diserahkan kepada kumpulan: apabila kaedah multiprocessing.pool.threadpool.apply_async 被调用时返回 multiprocessing.pool.asyncresult 实例，表示未来的完成。当在此实例上调用 get 方法时，您可以从辅助函数 (heartbeat_task) 获取返回值，或者重新引发辅助函数引发的任何异常。但是我们也可以使用 wait 方法来等待提交任务的完成或经过的时间。然后我们可以使用 ready menguji sama ada tugasan diserahkan selepas menunggu 5 saat sebenarnya Selesai (kerana pengecualian atau kembali). Jika tugas itu masih berjalan maka kita boleh menyuruhnya berhenti. Dalam demo ini, saya memaksa tugas untuk membuang pengecualian selepas kira-kira 7 saat:

from contextlib import contextmanager
from threading import event
from multiprocessing.pool import threadpool
from time import sleep


@contextmanager
def plc():
    stop_event = event()
    pool = threadpool(1)

    # send heartbeat every second
    async_result = pool.apply_async(heartbeat_task, args=(stop_event,))
    yield stop_event, async_result
    # we only return here if the task is no longer running
    try:
        # see if task threw an exception and if so, catch it:
        async_result.get()
    except exception as e:
        print("got exception:", e)
    finally:
        pool.close()
        pool.join()
        print("heartbeat stopped")


def heartbeat_task(stop_event):
    # for demo purposes, we will force an exception to occur
    # after approximately 7 seconds:
    value = false

    n = 0
    while not stop_event.is_set():
        value = not value
        print("heartbeat: " + str(value))
        sleep(1)
        n += 1
        if n == 7:
            raise exception('oops!')


def main():
    with plc() as tpl:
        stop_event, async_result = tpl
        # this function could forcibly cause the heartbeat_task
        # to complete by calling stop_event.set()

        # loop while the task is still running
        while not async_result.ready():
            """
            if some_condition:
                stop_event.set()
                break
            """
            print("program running")
            # sleep for 5 seconds or until heartbeat_task terminates:
            async_result.wait(5)

if __name__ == '__main__':
    main()

Cetakan:

program running
heartbeat: true
heartbeat: false
heartbeat: true
heartbeat: false
heartbeat: true
program running
heartbeat: false
heartbeat: true
got exception: oops!
heartbeat stopped

Alternatif untuk menggunakan pengurus konteks

from threading import Event
from multiprocessing.pool import ThreadPool
from time import sleep


def heartbeat_task(stop_event):
    value = False

    n = 0
    while not stop_event.is_set():
        value = not value
        print("Heartbeat: " + str(value))
        sleep(1)
        n += 1
        if n == 7:
            raise Exception('Oops!')

def main():
    stop_event = Event()
    pool = ThreadPool(1)
    async_result = pool.apply_async(heartbeat_task, args=(stop_event,))

    # Run as long as heartbeat_task is running:
    while not async_result.ready():
        """
        if some_condition:
            stop_event.set()
            break
        """
        print("Program running")
        # Sleep for 5 seconds or until heartbeat_task terminates:
        async_result.wait(5)

    # Any exception thrown in heartbeat_task will be rethrown and caught here:
    try:
        async_result.get()
    except Exception as e:
        print("Got exception:", e)
    finally:
        pool.close()
        pool.join()

if __name__ == '__main__':
    main()

Atas ialah kandungan terperinci Pengurus konteks dan benang daemon. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!