如何有效优化代码，如何使用 asyncio 实现多线程

Question

简而言之，脚本的作用以及我需要您的支持。算法运行，创建所有文件需要 10 多分钟。这样，队列会多次使用该消息。这将再次启动算法并创建 multilpe 文件，脚本最终花费的时间太长，有时需要 10-15 分钟，具体取决于文件中的数据量。您对如何优化脚本以使其运行得更快有任何想法：

   async def main(msg: func.ServiceBusMessage):
    if msg.content_type != "application/json":
        logging.info("Incorrect content type")
        return
    msg_body = json.loads(msg.get_body().decode("utf-8"))
    cc = msg_body.get("cc")
    job_id = msg_body.get("job_id")

    if cc is None or job_id is None:
        # Missing parameters
        logging.info("Required params are missing from message")
        return
    await update_status(
        job_id=job_id, cc=cc, status="EXECUTING", algorithm=ALGORITHM_NAME
    )

    # Reading input files
    azure_blob_client = AzureContainerClient(container=cc)

    # Files can be read form  API, when flag is_x set on True came from FE
    raw_csv_file = StringIO(
        await get_input_data(
            az_cl=azure_blob_client, name_part="Data", data_type="raw"
        )
    )
    ziajas_list = pd.read_csv(
        raw_csv_file
    )
   
    ziaja_json = ziajas_list.to_json(orient="records", date_format="iso")

    results = json.loads(ziaja_json)
    ziajas = {}
    for ziaja in results:
        ziaja_name = ziaja["carr"]
        if ziaja_name in ziajas:
            ziajas[ziaja_name].append(ziaja)
        else:
            ziajas[ziaja_name] = [ziaja]
        
    current_date_and_time = datetime.datetime.now()
    current_date_and_time_string = str(current_date_and_time)
    current_date_and_time_string = current_date_and_time_string.replace(' ', '_')
    ziaja_check_list = []
    ZiajaMapping = await get_parameters(parameters_name=ALGORITHM_NAME)
    all_results = []
    #Read parameter file

    for ziaja_name, ziajas_list in ziajas.items():
        try:
            if ziaja_name not in ziaja_check_list:
                user_list = get_ziaja_user_list(ziaja_name, ZiajaMapping)
                timezone = get_cc_details_timezone(cc, ZiajaMapping)
                cc_ID = get_cc_details_ccID(cc, ZiajaMapping)
                text_to_include = get_cc_details_text_to_include(cc, ZiajaMapping)
                admins = get_admins(cc, ZiajaMapping)
                meta_data = {
                "distribution": {
                    "method": "email",
                    "users": user_list,
                    "ccID": cc_ID,
                    "admin": admins
                },
                "subject": f"update for {timezone}",
                "header": {
                    "include": True, 
                    "apply_css": True
                },
                "handling": {
                    "message_style": "etxt", 
                    "title": "Lot update", 
                    "std_msg": "txt", 
                    "personalized_msg": text_to_include
                },  
                "footer": {
                    "include": True,
                    "footer_msg": "that algorithm"
                },
                "signature": {
                    "inclue": True,
                    "signature_text": "Team"
                },
                "attachments": {
                    "include": True,
                    "attachment_bytes": "base64string",
                    "attach_csv": True
                }, 
                "algorithm": {},
                "cc": {}
                }
                all_results.append(store_results(
                    result=ziajas_list,
                    cc=cc,
                    job_id=ziaja_name + "_" + current_date_and_time_string,
                    algorithm_name=ALGORITHM_NAME,
                    meta=meta_data,
                ))
                ziaja_check_list.append(ziaja_name)
        
        except Exception as e:
            logging.error(e)

    return await asyncio.gather(*all_results)

Answer 1

asyncio eventloop 在单个线程中运行并执行所有回调和任务本身。当一个任务正在运行时，没有其他任务可以同时运行。当正在运行的任务执行 await 表达式时，正在运行的任务将被挂起并执行其他任务。

因此，长期任务（例如文件操作）将阻塞事件循环，直到任务到达await（或完成）。为了防止这种情况，asycio 提供了一个高级函数来在单独的线程中异步运行函数：

asyncio.to_thread(func, /, *args, **kwargs)

import time
import asyncio

def blockingTask():
    print(f"start blockingTask at {time.strftime('%X')}")
    time.sleep(10)
    print(f"blockingTask complete at {time.strftime('%X')}")

async def task1():
    print(f"started task1 at {time.strftime('%X')}")
    for i in range(10):
        print(f'Task {i+1}/10')
        time.sleep(1)
    print(f"task1 complete at {time.strftime('%X')}")

async def main():
    print(f"started main at {time.strftime('%X')}")

    await asyncio.gather(
        to_thread(blockingTask),
        task1())
    print(f"finished main at {time.strftime('%X')}")

asyncio.run(main())

##### Output:
# blockingTask complete at 12:41:57
# task1 complete at 12:41:58
# finished main at 12:41:58

如您所见，这两个任务使用time.sleep() 导致线程冻结。即使阻塞了两个任务同时完成的线程，这是因为在不同的线程中运行

注意事项：

1. to_thread 仅适用于 python 3.9+

2. 将延迟设置为 0 可提供优化路径以允许其他任务运行。这可以由长时间运行的函数使用，以避免在函数调用的整个持续时间内阻塞事件循环。

3. 如果您使用的是旧版本，您可以自己声明函数to_thread。请参阅此问题的公认答案： Python module 'asyncio' has no attribute 'to_thread'（我已经对其进行了测试并工作了）。如果您自己声明函数，请确保导入 contextvars 和 functools。
4. 请注意，当您调用 to_thread(blockingTask) 时，被包裹的函数没有括号