探索 Flask 对 asyncio 的支持

 1import asyncio
 2import time
 3import threading
 4
 5
 6async def compute(cost):
 7    compute_start = time.time()
 8    await asyncio.sleep(cost)
 9    print(f'{cost}: {time.time() - compute_start} - {threading.current_thread().name}')
10    return cost + 1
11
12
13tasks = [asyncio.ensure_future(compute(cc)) for cc in [2, 4]]
14
15start = time.time()
16loop = asyncio.get_event_loop()
17try:
18    results = loop.run_until_complete(asyncio.gather(*tasks))
19    # loop.run_until_complete(asyncio.wait(tasks))  # 或者这种方式
20    print('results: ', results)
21finally:
22    loop.close()
23print(f'Total elapsed time {time.time() - start}')

1tasks = [compute(cc) for cc in [2, 4]]
2
3async def main():
4    return await asyncio.gather(*tasks)
5
6start = time.time()
7results = asyncio.run(main())   # main() 就是一个 coroutine 对象
8print('results:', results)
9print(f'Total elapsed time {time.time() - start}')

 1def run(main, *, debug=False):
 2    if events._get_running_loop() is not None:
 3        raise RuntimeError(
 4            "asyncio.run() cannot be called from a running event loop")
 5
 6    if not coroutines.iscoroutine(main):
 7        raise ValueError("a coroutine was expected, got {!r}".format(main))
 8
 9    loop = events.new_event_loop()
10    try:
11        events.set_event_loop(loop)
12        loop.set_debug(debug)
13        return loop.run_until_complete(main)
14    finally:
15        try:
16            _cancel_all_tasks(loop)
17            loop.run_until_complete(loop.shutdown_asyncgens())
18        finally:
19            events.set_event_loop(None)
20            loop.close()

 1import asyncio
 2import requests
 3from flask import Flask
 4
 5app = Flask(__name__)
 6
 7
 8async def fetch(url):
 9    return requests.get(url).text
10
11async def main():
12   tasks = [fetch(url) for url in ["https://google.com", "https://bing.com", "https://yanbin.blog"]]
13   return await asyncio.gather(*tasks)
14
15@app.route("/")
16def index():
17    responses = asyncio.run(main())
18    return f'response sizes: {[len(res) for res in responses]}\n'
19
20if __name__ == "__main__":
21    app.run(debug=False, use_reloader=False)

 1import asyncio
 2import requests
 3from flask import Flask
 4
 5app = Flask(__name__)
 6
 7async def fetch(url):
 8    return requests.get(url).text
 9
10@app.route("/")
11def index():
12
13    try:
14        loop = asyncio.get_event_loop()
15    except RuntimeError as er:
16        print(er.args[0], 'create a new EventLoop')
17        loop = asyncio.new_event_loop()
18        asyncio.set_event_loop(loop)
19
20    responses = loop.run_until_complete(asyncio.gather(
21        fetch("https://google.com"),
22        fetch("https://bing.com"),
23        fetch("https://yanbin.blog")
24    ))
25    return f'response sizes: {[len(res) for res in responses]}'
26
27if __name__ == "__main__":
28    app.run(debug=False, use_reloader=False)

 1import asyncio
 2from flask import Flask
 3
 4async def abar(a):
 5    print(a)
 6
 7loop = asyncio.get_event_loop()  # 获得主线程上的 EventLoop
 8app = Flask(__name__)
 9
10@app.route("/")
11def notify():
12    loop.run_until_complete(abar("abar"))  # 异步调用使用主线程的 EventLoop
13    return "OK"
14
15if __name__ == "__main__":
16    app.run(debug=False, use_reloader=False)

 1from flask import Flask, jsonify, has_request_context, copy_current_request_context
 2from functools import wraps
 3from concurrent.futures import Future, ThreadPoolExecutor
 4import asyncio
 5
 6def run_async(func):
 7    @wraps(func)
 8    def _wrapper(*args, **kwargs):
 9        call_result = Future()
10        def _run():
11            loop = asyncio.new_event_loop()
12            try:
13                result = loop.run_until_complete(func(*args, **kwargs))
14            except Exception as error:
15                call_result.set_exception(error)
16            else:
17                call_result.set_result(result)
18            finally:
19                loop.close()
20
21        loop_executor = ThreadPoolExecutor(max_workers=1)
22        if has_request_context():
23            _run = copy_current_request_context(_run)
24        loop_future = loop_executor.submit(_run)
25        loop_future.result()
26        return call_result.result()
27
28    return _wrapper
29
30
31app = Flask(__name__)
32
33
34@app.route('/')
35@run_async
36async def index():
37    return jsonify('hello')
38
39app.run()

1@app.route('/use-external-api')
2@async
3def use_external_api():
4    response = yield from aiohttp.request(
5        'GET', 'https://api.example.com/data/1')
6    data = yield from response.read()
7
8    return data

 1from asyncio import coroutine, sleep
 2from flask import Flask, request
 3
 4app = Flask(__name__)
 5
 6@app.route("/hello/<string:name>")
 7@coroutine
 8def say_hi(name):
 9    yield from sleep(2)
10    return "it worked %s" % request.args.get("name", name)
11
12app.run()

1import asyncio
2import async_timeout
3
4async def fetch(url):
5    async with aiohttp.ClientSession() as session, async_timeout.timeout(10):
6        async with session.get(url) as response:
7            return await response.text()

1async def fetch(url):
2    async with aiohttp.ClientSession() as session, async_timeout.timeout(10):
3        async with session.get(url) as response:
4            response = await response.text()
5            async with session.get(f'http://localhost:5000/ping/{len(response)}') as res:
6                return await res.text()