python多线程、异步

import asyncio
import time

# 一定义一个阻塞同步函数
def blocking_task(seconds: int):
    time.sleep(seconds)
    print(f"同步阻塞任务：睡眠了 {seconds} 秒")

# 定义一个异步函数
async def say_hello(name: str) -> None:
    # 模拟异步等待
    await asyncio.sleep(1)
    print(f"你好，{name}！")

# 异步调用异步函数
async def say_hello_2():
    # 直接 await 一个异步函数
    await say_hello("小易")

# 异步调用同步
async def blocking_task_async():
    # 直接调用，但是会阻塞2秒
    blocking_task(2)

    # 异步运行同步函数，且不阻塞,会将blocking_task函数运行在另一个线程中，不会阻塞当前线程
    await asyncio.to_thread(blocking_task, 2)

# 运行异步程序
asyncio.run(say_hello_2())

# 并发执行异步函数，返回task对象，可以用它停止
task = asyncio.create_task(say_hello())

# 停止异步线程
task.cancel()

import concurrent.futures
from concurrent.futures import ThreadPoolExecutor
import threading

# 处理单个人设的函数
def process_single_persona(persona):
	# 这里写处理单个人设
    pass

# 运行账号主线编排任务
def run_main_workflow():
    """
    使用线程池处理大量persona，不收集结果统计
    """
    # 查所有人设
    personas = db.query_all_personas()
    total_count = len(personas)
    print(f"总共需要处理 {total_count} 个人设")
    
    if total_count == 0:
        print("没有找到任何人设")
        return
    
    # 根据数据库连接池大小动态调整线程数
    max_workers = min(32, max(4, total_count // 1000))  # 动态调整，最少4个，最多32个
    print(f"使用 {max_workers} 个线程进行处理")
    
    # 使用线程池处理
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 遍历personas
        for persona in personas:
			# 将单个persona提交给线程池使用process_single_persona函数处理
            executor.submit(process_single_persona, persona)

import asyncio

# 异步处理单个人设的函数
async def process_single_persona(persona):
    # 这里写处理单个人设的异步逻辑
    # 例如：await some_io_bound_operation(persona)
    pass  # TODO: 替换为实际逻辑

# 异步运行账号主线编排任务
async def run_main_workflow():
    """
    使用 asyncio 协程处理大量 persona，不使用线程池
    """
    # 如果你的 db 库支持异步查询，建议改为 await db.query_all_personas()
    personas = db.query_all_personas()  # 同步获取所有人设
    total_count = len(personas)
    print(f"总共需要处理 {total_count} 个人设")  # 中文注释：输出总数
    
    if total_count == 0:
        print("没有找到任何人设")
        return

    # 根据总量动态计算最大并发协程数：最少4，最多32
    max_concurrency = min(32, max(4, total_count // 1000))
    print(f"使用 {max_concurrency} 个协程并发处理")  # 中文注释：输出并发量

    # 用信号量限制并发数
    semaphore = asyncio.Semaphore(max_concurrency)

    async def sem_wrapper(persona):
        # 每次只有拿到信号量的协程才能执行
        async with semaphore:
            await process_single_persona(persona)

    # 创建所有任务
    tasks = [asyncio.create_task(sem_wrapper(p)) for p in personas]

    # 等待所有任务完成
    await asyncio.gather(*tasks)

    print("全部完成")  # 中文注释：所有人设处理完毕

if __name__ == "__main__":
    # 在主线程里启动事件循环
    asyncio.run(run_main_workflow())