如何让异步main函数仅处理新数据项，避免重复发送？

核心问题分析 #

1. 未将已发送的项标记为已检查，导致每次循环重复发送
2. 用同步的time.sleep()阻塞事件循环，无法精准响应新数据添加

修改方案 #

1. 优化已处理项的检查逻辑

用集合存储已处理ID，避免每次遍历列表，提升查找效率；发送后立即标记为已处理，防止重复发送。

2. 用异步事件实现新数据等待

替换同步sleep，使用asyncio.Event实现异步等待，只有当新数据添加时才触发处理逻辑，避免无效循环。

完整修改代码 #

import asyncio

# 假设这是你的全局数据和已检查列表
data = []
checked = {'items': []}
# 初始化已处理ID集合，快速判断是否已发送
processed_ids = {item['id'] for item in checked['items']}

async def sendItem(item):
    # 这里是你的发送逻辑
    print(f"发送数据：{item}")

async def main(new_data_event):
    global data, processed_ids, checked
    while True:
        # 等待新数据事件触发，无新数据时挂起协程，不阻塞事件循环
        await new_data_event.wait()
        
        # 遍历处理所有未发送的项
        for item in data:
            item_id = item['id']
            if item_id in processed_ids:
                continue
            
            # 发送数据
            await sendItem(item)
            # 标记为已处理，同时更新checked集合
            processed_ids.add(item_id)
            checked['items'].append({'id': item_id})
        
        # 重置事件，等待下一次新数据添加
        new_data_event.clear()

def add_new_item(new_item):
    """添加新数据的函数，调用此函数时触发处理逻辑"""
    global data, new_data_event
    data.append(new_item)
    # 触发事件，通知main协程有新数据需要处理
    new_data_event.set()

if __name__ == "__main__":
    new_data_event = asyncio.Event()
    # 启动异步事件循环
    asyncio.run(main(new_data_event))
    
    # 示例：添加新数据时调用add_new_item
    # add_new_item({'id': '1', 'content': '新数据1'})

关键说明 #

• 已处理标记：用processed_ids集合存储已发送的ID，O(1)时间复杂度判断是否已处理，比原列表遍历效率更高
• 异步等待：asyncio.Event让协程在无新数据时挂起，不占用CPU资源，新数据添加时立即唤醒处理
• 触发机制：所有新数据通过add_new_item函数添加，确保每次添加都会触发事件，不会遗漏数据

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