如何将Python顺序执行的for循环改为并行运行及问题排查

一、multiprocessing无输出的常见原因 #

• 缺少进程入口守卫：Windows系统下使用multiprocessing必须加if __name__ == '__main__'判断，否则进程创建逻辑会被重复导入执行，导致程序卡住无输出。
• print输出缓冲区未刷新：多进程环境下标准输出默认会缓冲内容，没有flush的话不会实时打印到控制台。
• 无用的StreamConn实例化：你的代码中创建了StreamConn对象但没有实际使用，该对象是长连接实例，异常初始化可能会阻塞进程执行。

二、修复后的多进程代码 #

import multiprocessing
import alpaca_trade_api as tradeapi

keys = [['apikey1','secretkey1'],['apikey2','secretkey2'],['apikey3','secretkey3']]

def printstats(key):
    print(key, flush=True)
    api = tradeapi.REST(key[0],key[1])
    # 删掉无用的StreamConn初始化，需要用流的时候再加
    account = api.get_account()
    print(float(account.equity), flush=True)
    print(float(account.last_equity), flush=True)
    print(float(account.equity) - float(account.last_equity), flush=True)
    # 返回结果方便后续处理
    return {
        "key": key,
        "equity": float(account.equity),
        "last_equity": float(account.last_equity),
        "diff": float(account.equity) - float(account.last_equity)
    }

if __name__ == '__main__':
    # 可以指定进程数，默认是CPU核心数，你只有3个账户指定3就够
    a_pool = multiprocessing.Pool(processes=3)
    result = a_pool.map(printstats, keys)
    a_pool.close()
    a_pool.join()
    print(result)

三、是否应该改用async实现 #

完全可以，你的场景属于IO密集型任务，大部分耗时是在等待Alpaca接口返回数据，用async协程比多进程资源开销小很多，执行效率更高。

四、多进程和async的核心区别 #

• 多进程：属于真正的并行执行，每个进程有独立的内存空间和GIL，适合CPU密集型运算场景，创建和切换进程的开销很大，你当前场景用属于资源浪费。
• async：属于单线程内的协程调度，只有遇到IO等待（比如网络请求、文件读写）时才会切换到其他任务，不需要多进程/多线程的上下文切换开销，非常适合大量IO操作的场景，代码维护也更简单。

五、async实现示例 #

alpaca官方SDK已经提供了异步客户端，直接调用即可：

import asyncio
import alpaca_trade_api as tradeapi

keys = [['apikey1','secretkey1'],['apikey2','secretkey2'],['apikey3','secretkey3']]

async def printstats(key):
    print(key, flush=True)
    # 使用异步REST客户端
    api = tradeapi.AsyncREST(key[0], key[1])
    account = await api.get_account()
    equity = float(account.equity)
    last_equity = float(account.last_equity)
    diff = equity - last_equity
    print(equity, flush=True)
    print(last_equity, flush=True)
    print(diff, flush=True)
    await api.close()
    return {
        "key": key,
        "equity": equity,
        "last_equity": last_equity,
        "diff": diff
    }

async def main():
    tasks = [printstats(key) for key in keys]
    results = await asyncio.gather(*tasks)
    print(results)

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

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