Telethon未接收到新消息时如何定时执行函数且不阻塞消息监听

Telethon 基于 asyncio 异步框架运行，只需额外创建独立的定时协程任务，和消息监听逻辑同时挂载到事件循环中即可，两者不会互相阻塞。以下是两种常见场景的实现方案：

场景1：固定每10分钟执行一次，和是否收到消息无关 #

适合不需要关联消息接收状态，只要到点就执行任务的需求：

import asyncio
from telethon import TelegramClient, events

# 替换为你自己的API参数
api_id = 你的API_ID
api_hash = '你的API_HASH'
client = TelegramClient('session_name', api_id, api_hash)

# 定时执行的自定义函数
async def periodic_task():
    while True:
        # 异步等待10分钟，不会阻塞其他任务
        await asyncio.sleep(10 * 60)
        # 这里编写你需要定时执行的操作
        print("执行10分钟定时任务")

async def main():
    me = await client.get_me()
    print(me.stringify())

    # 启动定时任务，不会阻塞后续的消息监听
    asyncio.create_task(periodic_task())

    # 原有消息监听逻辑
    @client.on(events.NewMessage('TelegramID'))
    async def new_message_listener(event):
        mensaje = event.message.message
        if mensaje is not None:
           print(mensaje)
    
    # 运行客户端直到断开
    await client.run_until_disconnected()

if __name__ == '__main__':
    client.loop.run_until_complete(main())

场景2：连续10分钟未收到消息才执行任务，收到消息重置计时 #

适合只有在消息空闲时段才触发任务的需求：

import asyncio
import time
from telethon import TelegramClient, events

api_id = 你的API_ID
api_hash = '你的API_HASH'
client = TelegramClient('session_name', api_id, api_hash)

# 记录最后一次收到消息的时间
last_msg_time = time.time()

async def idle_periodic_task():
    global last_msg_time
    while True:
        await asyncio.sleep(60) # 每分钟检查一次空闲状态
        current_time = time.time()
        # 距离上次收到消息超过10分钟则执行操作
        if current_time - last_msg_time >= 10 * 60:
            print("已10分钟未收到消息，执行指定操作")
            # 执行完更新时间，避免每分钟重复触发
            last_msg_time = current_time

async def main():
    global last_msg_time
    me = await client.get_me()
    print(me.stringify())

    asyncio.create_task(idle_periodic_task())

    @client.on(events.NewMessage('TelegramID'))
    async def new_message_listener(event):
        global last_msg_time
        # 收到新消息就更新最后接收时间，重置计时
        last_msg_time = time.time()
        mensaje = event.message.message
        if mensaje is not None:
           print(mensaje)
    
    await client.run_until_disconnected()

if __name__ == '__main__':
    client.loop.run_until_complete(main())

注意事项 #

• 所有阻塞操作（比如本地IO、同步网络请求）都要改为异步实现，或者放到线程池中运行，否则会卡住整个asyncio事件循环，导致消息监听和定时任务同时失效
• 不要使用同步的time.sleep()等待，必须用asyncio.sleep()实现异步等待，才不会阻塞其他任务的运行
• 如果需要更复杂的定时规则（比如指定时点触发、多任务调度），可以使用APScheduler的异步版本，直接挂载到当前的asyncio事件循环中即可

内容的提问来源于stack exchange，提问作者dyst