一个关于协程的 Python 面试题

之前面试遇到这么一道有趣的题目，但是只给出一个输入用例，我写的代码可以通过这个输入用例，但是其它不知道的输入用例没法通过，不知道有没有大佬有啥思路吗？

这个是题目的链接： https://drive.google.com/drive/folders/1KQ6LYQcbhdINcVB1ChayloO01oXR1hF6?usp=sharing

import asyncio

def main():
    async def do_sync_f():
        return sync_f()

    loop = asyncio.get_event_loop()

    task_g = loop.create_task(async_g())
    task_f = loop.create_task(do_sync_f())

    tasks = [
        task_g, task_f
    ]

    loop.run_until_complete(asyncio.wait(tasks))

    print(max(task_g.result(), task_f.result()))


if __name__ == '__main__':
    main()

答:有些 async 函数写的不好调用 run untill complete 你再调用 run untill complete 会出错
答:看了一遍题目，感觉你完全没理解这道题要考查什么。建议看看 https://docs.python.org/3/library/asyncio-task.html#running-in-threads ，另外，根据 python 版本不同，提供的函数不同。
答:你的这个写法并不正确，要知道 sync_f 里面的代码是会阻塞线程的，如果你 sync_f 和 async_g 都在主线程跑，那么如果 sync_f 先执行，event loop 就要等到 sync_f 解除阻塞才会执行 async_g ，那么 sync_f 和 async_g 并不 parallel ，你现在的代码之所以看起来并行，实际上是 async_g 先执行了，你可以在两个函数的开头写个 print 看看。

调用顺序对 event loop 的影响如下

import asyncio
import time

async def f():
time.sleep(3)

async def g():
await asyncio.sleep(3)

async def main():
start = time.time()
# await asyncio.gather(f(), g()) # cost 6.006543159484863
# await asyncio.gather(g(), f()) # cost 3.0021119117736816
print(f"cost {time.time()-start}")

asyncio.run(main())

所以，这题中 sync_f 必须放到另外的线程执行
答:博客欢迎赐教。或者问一下 ChatGPT 吧，它协助你，通关比较快。
答:这是 GPT4 的回答，考察的应该是：asyncio.to_thread 这个知识点。

You can use asyncio.to_thread() to run the synchronous function sync_f() in a separate thread, allowing it to run concurrently with the asynchronous function async_g(). Here's how you can achieve this:
python

Copy

import asyncio
import threading
import time

def sync_f(a=None):
lock = threading.Lock()
lock.acquire(blocking=False)
lock.acquire(timeout=3)
return 2

async def async_g(a=None):
await asyncio.sleep(3)
return 1

async def main():
# Run sync_f() in a separate thread and async_g() concurrently
sync_f_task = asyncio.to_thread(sync_f)
async_g_task = async_g()

results = await asyncio.gather(sync_f_task, async_g_task)
print(results)

if __name__ == "__main__":
asyncio.run(main())
This code will run both sync_f() and async_g() concurrently, and print their results once both are completed.