如何整合异步定时器代码与事件监听死循环代码，实现二者独立无干扰运行？

看起来你已经有了两个能单独工作的模块——异步定时器和同步的事件监听循环，但现在要让它们“和平共处”对吧？问题的核心在于那个while True的事件监听是同步阻塞式的，会把整个线程占住，导致asyncio的事件循环没法处理定时器任务，反过来也一样。下面给你两种靠谱的整合方案，你可以根据项目需求选择：

方案1：线程分离法（改动最小，快速上手） #

这个方法最直接——把你的同步事件监听循环丢到一个单独的子线程里运行，主线程继续处理asyncio的定时器任务，两者在不同线程完全独立，互不干扰。

修改后的完整代码 #

import asyncio
import threading
import selectors
import evdev
from evdev import ecodes

# 你的Timer类代码，原样保留
class Timer:
    def __init__(self, interval, first_immediately, timer_name, context, callback):
        self._interval = interval
        self._first_immediately = first_immediately
        self._name = timer_name
        self._context = context
        self._callback = callback
        self._is_first_call = True
        self._ok = True
        self._task = asyncio.ensure_future(self._job())
        print(timer_name + " init done")

    async def _job(self):
        try:
            while self._ok:
                if not self._is_first_call or not self._first_immediately:
                    await asyncio.sleep(self._interval)
                await self._callback(self._name, self._context, self)
                self._is_first_call = False
        except Exception as ex:
            print(ex)

    def cancel(self):
        self._ok = False
        self._task.cancel()

# 定时器回调函数，原样保留
async def timer1(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))

async def timer2(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))

# 把事件监听循环包装成线程函数，支持优雅停止
running = True
def run_event_listener():
    global running
    # 这里替换成你实际的输入设备路径
    device = evdev.InputDevice('/dev/input/event0')
    selector = selectors.DefaultSelector()
    selector.register(device, selectors.EVENT_READ)
    
    while running:
        # 给select加timeout，每隔0.5秒检查一次是否要停止
        for key, mask in selector.select(timeout=0.5):
            device = key.fileobj
            for event in device.read():
                if event.type == ecodes.EV_KEY:
                    match event.code:
                        case 114:
                            print("Vol down")
                        case 115:
                            print("Vol up")
    # 优雅清理
    selector.unregister(device)
    selector.close()
    device.close()

if __name__ == "__main__":
    # 初始化定时器
    t1 = Timer(interval=1, first_immediately=True, timer_name="Timer Channel", context={'count': 0}, callback=timer1)
    t2 = Timer(interval=3, first_immediately=False, timer_name="Timer Scroll", context={'count': 0}, callback=timer2)

    # 启动事件监听子线程（daemon=True：主线程退出时自动终止子线程）
    listener_thread = threading.Thread(target=run_event_listener, daemon=True)
    listener_thread.start()

    try:
        loop = asyncio.get_event_loop()
        loop.run_forever()
    except KeyboardInterrupt:
        # 优雅停止事件监听
        running = False
        # 停止定时器
        t1.cancel()
        t2.cancel()
        # 等待子线程退出（可选）
        listener_thread.join(timeout=1)
        print("clean up done")

方案说明 #

• 我们把原来的while True事件循环包装成了线程函数run_event_listener，通过global running变量实现优雅停止
• 用daemon=True创建线程，确保主线程退出时子线程自动结束，避免僵尸线程
• 给selector.select()加了timeout=0.5，这样线程会定期检查running状态，收到停止信号后能及时退出

方案2：纯异步改造（优雅，适合长期维护） #

如果你的项目想完全基于asyncio生态，推荐把事件监听逻辑改成异步风格，利用asyncio的事件循环直接处理设备IO，无需线程，更符合异步设计理念。

修改后的完整代码 #

import asyncio
import evdev
from evdev import ecodes

# 你的Timer类代码，原样保留
class Timer:
    def __init__(self, interval, first_immediately, timer_name, context, callback):
        self._interval = interval
        self._first_immediately = first_immediately
        self._name = timer_name
        self._context = context
        self._callback = callback
        self._is_first_call = True
        self._ok = True
        self._task = asyncio.ensure_future(self._job())
        print(timer_name + " init done")

    async def _job(self):
        try:
            while self._ok:
                if not self._is_first_call or not self._first_immediately:
                    await asyncio.sleep(self._interval)
                await self._callback(self._name, self._context, self)
                self._is_first_call = False
        except Exception as ex:
            print(ex)

    def cancel(self):
        self._ok = False
        self._task.cancel()

# 定时器回调函数，原样保留
async def timer1(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))

async def timer2(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))

# 异步事件处理函数
def handle_event(device):
    for event in device.read():
        if event.type == ecodes.EV_KEY:
            match event.code:
                case 114:
                    print("Vol down")
                case 115:
                    print("Vol up")

# 初始化事件监听
async def setup_event_listener(loop):
    # 这里替换成你实际的输入设备路径
    device = evdev.InputDevice('/dev/input/event0')
    # 告诉asyncio事件循环：当设备有数据可读时，调用handle_event
    loop.add_reader(device.fileno(), handle_event, device)
    return device

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    # 初始化事件监听
    device = loop.run_until_complete(setup_event_listener(loop))
    # 初始化定时器
    t1 = Timer(interval=1, first_immediately=True, timer_name="Timer Channel", context={'count': 0}, callback=timer1)
    t2 = Timer(interval=3, first_immediately=False, timer_name="Timer Scroll", context={'count': 0}, callback=timer2)

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        # 停止定时器
        t1.cancel()
        t2.cancel()
        # 移除事件监听
        loop.remove_reader(device.fileno())
        device.close()
        print("clean up done")

方案说明 #

• 用loop.add_reader()把设备的读事件注册到asyncio事件循环中，当设备有输入时自动触发handle_event
• 所有任务（定时器+事件监听）都在同一个异步事件循环中运行，没有线程切换的开销
• 代码更统一，符合asyncio的设计哲学，适合长期维护的异步项目

总结 #

• 如果你只是想快速把两个模块拼起来用，**方案1（线程分离）**是最佳选择，改动极小，几乎不用动原有逻辑
• 如果你正在做一个纯异步的项目，或者希望代码更优雅、无线程问题，**方案2（纯异步改造）**更合适

备注：内容来源于stack exchange，提问作者Kaloschke