如何让aiogram中图像检测与消息处理函数并发运行？

我开发了一个基于Python asyncio和aiogram的Telegram机器人，包含image_detection()和handle_messages()两个核心函数：

• image_detection()：通过OpenCV和pyautogui捕获屏幕，执行模板匹配检测特定图像，检测到后通过aiogram发送截图给用户。
• handle_messages()：处理用户消息并回复确认信息。

当前问题：两个函数无法并发运行，当image_detection()执行时，handle_messages()无法回复用户消息。尝试用asyncio.gather()实现并发但未解决。

原代码如下：

import asyncio
import numpy as np
from aiogram import *
import pyautogui
import cv2
import os

current_directory = os.path.dirname(__file__)
image_directory = os.path.join(current_directory, 'Images')
template_image = cv2.imread(
    image_directory + "\\image.pnh", cv2.IMREAD_GRAYSCALE)

bot_token = 'MY_BOT_TOKEN'
chat_id = MY_CHAT_ID
bot = Bot(token=bot_token)
dp = Dispatcher(bot)

def template_match(template, x = 0.8):
    screenshot = pyautogui.screenshot()
    screenshot = np.array(screenshot)  
    gray_screenshot = cv2.cvtColor(screenshot, cv2.COLOR_BGR2GRAY)
    res = cv2.matchTemplate(gray_screenshot, template, cv2.TM_CCOEFF_NORMED)
    threshold = x
    loc = np.where(res >= threshold)
    if len(loc[0]) > 0:
        return True
    else:
        return False

@dp.message_handler()
async def handle_messages(message: types.Message):
    await message.reply("I received your message.")
    # Going to add some code to handle user messages

async def image_detection():
    print("imageDetection() is executing...")
    while True:
        if(template_match(template_image)):
            screenshot = pyautogui.screenshot()
            screenshot.save('captured_image.jpg')
            with open('captured_image.jpg', 'rb') as photo:
                await bot.send_photo(chat_id, photo)
                await bot.send_message(chat_id, 'Image detected') # send detected image to user

async def main():
    await asyncio.gather(
        dp.start_polling(),
        image_detection()
    )

if __name__ == '__main__':
    asyncio.run(main())

问题根源 #

template_match是同步阻塞函数，其中pyautogui.screenshot()、OpenCV的图像转换和模板匹配都是CPU密集型的同步操作，会长时间占用asyncio的事件循环，导致消息处理任务被阻塞，无法及时响应。

修改方案 #

1. 将同步阻塞的template_match和pyautogui.screenshot()操作放到asyncio.to_thread()中，把任务移到线程池执行，避免阻塞事件循环。
2. 在image_detection的循环中添加await asyncio.sleep(0.1)，主动给事件循环让出时间片，确保消息处理任务能被调度。
3. 优化截图保存逻辑，避免多个线程同时写入同一个文件（比如使用临时文件或唯一文件名）。

修改后的完整代码 #

import asyncio
import numpy as np
from aiogram import *
import pyautogui
import cv2
import os
import tempfile  # 用于创建临时文件

current_directory = os.path.dirname(__file__)
image_directory = os.path.join(current_directory, 'Images')
# 修正文件名拼写错误（原pnh应为png，否则模板图像无法加载）
template_image = cv2.imread(
    os.path.join(image_directory, "image.png"), cv2.IMREAD_GRAYSCALE)

bot_token = 'MY_BOT_TOKEN'
chat_id = MY_CHAT_ID
bot = Bot(token=bot_token)
dp = Dispatcher(bot)

def template_match(template, x = 0.8):
    screenshot = pyautogui.screenshot()
    screenshot = np.array(screenshot)  
    gray_screenshot = cv2.cvtColor(screenshot, cv2.COLOR_BGR2GRAY)
    res = cv2.matchTemplate(gray_screenshot, template, cv2.TM_CCOEFF_NORMED)
    threshold = x
    loc = np.where(res >= threshold)
    return len(loc[0]) > 0

@dp.message_handler()
async def handle_messages(message: types.Message):
    await message.reply("I received your message.")
    # 后续添加消息处理逻辑

async def image_detection():
    print("imageDetection() is executing...")
    while True:
        # 将同步的模板匹配操作放到线程池执行
        detected = await asyncio.to_thread(template_match, template_image)
        if detected:
            # 将截图操作也放到线程池
            screenshot = await asyncio.to_thread(pyautogui.screenshot)
            # 使用临时文件保存截图，避免文件冲突
            with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as temp_file:
                screenshot.save(temp_file.name)
                with open(temp_file.name, 'rb') as photo:
                    await bot.send_photo(chat_id, photo)
                    await bot.send_message(chat_id, 'Image detected')
            # 删除临时文件
            os.unlink(temp_file.name)
        # 让出时间片，让事件循环处理其他任务
        await asyncio.sleep(0.1)

async def main():
    await asyncio.gather(
        dp.start_polling(),
        image_detection()
    )

if __name__ == '__main__':
    asyncio.run(main())

关键修改点说明 #

• asyncio.to_thread()：把同步的CPU密集型任务（截图、模板匹配）移到单独的线程中执行，asyncio事件循环可以在等待线程结果时处理其他任务（比如消息回复）。
• asyncio.sleep(0.1)：每次循环后暂停0.1秒，强制事件循环切换到其他任务，确保消息处理不会被长时间阻塞。
• 临时文件：避免多个检测周期同时写入同一个captured_image.jpg导致文件损坏或冲突。
• 文件名修正：原代码中image.pnh应为image.png（大概率是拼写错误），否则模板图像无法加载。

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