Title: python-asynchronous execution library asyncio
categories: python
tags: [python, asyncio, asynchronous, parallel]
date: 2020-09-28 14:45:34
comments: false
mathjax: true
toc: true
When writing the tool, it needs to request data for dozens of times, synchronous sequential execution, the speed is a little slow, so it is much easier to change to asynchronous parallel execution. Similarly, other designs to IO that will block can be solved by asynchronous parallel execution. Similarly, file IO can also be handled asynchronously.
Asyncio (built-in) + AIoHTTP /aiofiles (requires PIP installation) is used
code
Tool class Async_util.py (simply wrap it up)
# -*- coding: utf-8 -*-
import aiofiles
import aiohttp
import asyncio
import json
import sys
import traceback
import threading
from typing import List
from tool import utils
class CReqInfo:
def __init__(self):
self.url = None
self.method = "POST"
self.data = None
self.extA = None
class CRspInfo:
def __init__(self):
self.code: int = 0
self.text = None
self.extA = None
class CFileInfo:
def __init__(self, path, encoding="utf-8"):
self.path = path
self.encoding = encoding
self.content = None
self.error = None
self.extA = None
class CCmdInfo:
def __init__(self, cmd):
self.code = 0
self.msg = None
self.cmd = cmd
self.extA = None
class CThreadInfo:
def __init__(self, target, args=()):
self.target = target
self.args = args
self.result = None
class CInnerThread(threading.Thread):
# def __init__(self, autoid, target, args=()):
def __init__(self, autoid, ti: CThreadInfo):
super(CInnerThread, self).__init__()
self.autoid = autoid
self.target = ti.target
self.args = ti.args
self.ti: CThreadInfo = ti
def run(self):
try:
self.ti.result = self.target(*self.args)
except Exception as e:
self.ti.result = e
traceback.print_stack()
def get_result(self):
return self.autoid, self.ti
class CAsyncHttp:
async def request(self, reqInfo: CReqInfo):
if isinstance(reqInfo.data, dict):
reqInfo.data = json.dumps(reqInfo.data)
rspInfo = CRspInfo()
try:
async with aiohttp.request(method=reqInfo.method, url=reqInfo.url, data=reqInfo.data) as response:
rspInfo.code = int(response.status)
rspInfo.extA = reqInfo.extA
rspInfo.text = await response.text()
except Exception as e:
rspInfo.code = -999
rspInfo.text = e
finally:
return rspInfo
def doReq(self, *reqArr) -> List[CRspInfo]:
return CAsyncTask().doTask(*[self.request(reqInfo) for reqInfo in reqArr])
class CAsyncFileRead:
async def read(self, fi: CFileInfo):
try:
async with aiofiles.open(fi.path, mode="rb") as fd:
content = await fd.read()
fi.content = fi.encoding is None and content or str(content, encoding=fi.encoding, errors="ignore")
except Exception as e:
fi.error = e
finally:
return fi
def doRead(self, *fileArr) -> List[CFileInfo]:
return CAsyncTask().doTask(*[self.read(fi) for fi in fileArr])
class CAsyncFileWrite:
async def write(self, fi: CFileInfo):
utils.createDirForFile(fi.path)
try:
async with aiofiles.open(fi.path, mode="wb") as fd:
bts = fi.encoding is None and fi.content or fi.content.encode(
encoding=fi.encoding)
await fd.write(bts)
except Exception as e:
fi.error = e
finally:
return fi
def doWrite(self, *fileArr) -> List[CFileInfo]:
return CAsyncTask().doTask(*[self.write(fi) for fi in fileArr])
class CAsyncCmd:
async def run(self, ci: CCmdInfo):
proc = await asyncio.create_subprocess_shell(
ci.cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE)
stdout, stderr = await proc.communicate()
bts = stdout or stderr
ci.code = proc.returncode
ci.msg = bts is not None and str(bts, encoding="utf-8", errors="ignore")
return ci
def doCmd(self, *cmdArr) -> List[CCmdInfo]:
return CAsyncTask().doTask(*[self.run(ci) for ci in cmdArr])
class CAsyncTask:
def __init__(self):
self.isStopProgress = False
async def progress(self):
symbol = ["/", "ᅳ", "\\", "|"]
total = len(symbol)
cnt = 0
while not self.isStopProgress:
sys.stdout.write(f"------ processing {symbol[cnt % total]}\r")
sys.stdout.flush()
await asyncio.sleep(0.1)
cnt += 1
print("------ processing 100%")
async def start(self, *taskArr):
first = asyncio.gather(*taskArr)
second = asyncio.create_task(self.progress())
retVal = await first
self.isStopProgress = True
await second
return retVal
def doTask(self, *taskArr):
loop = asyncio.get_event_loop()
res = loop.run_until_complete(self.start(*taskArr))
# loop.close() # https 会报错: RuntimeError: Event loop is closed
return res
class CAsyncThread:
def doRun(self, *threadArr) -> List[CThreadInfo]:
thdInsArr = []
autoid = 1
for ti in threadArr:
thd = CInnerThread(autoid=autoid, ti=ti)
autoid += 1
thdInsArr.append(thd)
thd.start()
retDct = {}
for thd in thdInsArr:
thd.join()
aid, ti = thd.get_result()
retDct[aid] = ti
sorted(retDct.items(), key=lambda x: x[0], reverse=False)
return list(retDct.values())
def doTask(*taskArr):
return CAsyncTask().doTask(*taskArr)
def doReq(*reqArr):
return CAsyncHttp().doReq(*reqArr)
def doRead(*fileArr):
return CAsyncFileRead().doRead(*fileArr)
def doWrite(*fileArr):
return CAsyncFileWrite().doWrite(*fileArr)
def doCmd(*cmdArr):
return CAsyncCmd().doCmd(*cmdArr)
def doRun(*threadArr):
return CAsyncThread().doRun(*threadArr)
The test case
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import sys
import os
import asyncio, aiohttp, aiofiles
import json
from datetime import datetime, timedelta
from time import ctime, sleep
import time
import unittest
from tool import utils, async_util
SelfPath: str = os.path.abspath(os.path.dirname(__file__))
class Test_Async(unittest.TestCase):
def setUp(self):
print("\n\n------------------ test result ------------------")
def test_gather(self):
async def count(num):
print(f"One - {num}")
await asyncio.sleep(1)
print(f"Two - {num}")
async def main():
await asyncio.gather(count(1), count(2), count(3)) # gather Execute concurrently, returning sequential results.
asyncio.run(main())
print("--- finished")
def test_createTask(self):
async def count(num):
print("One")
await asyncio.sleep(num)
print("Two")
async def main():
first = asyncio.create_task(count(2)) # Start executing it when you create it.
second = asyncio.create_task(count(1))
await first
print(f"finished first")
await second
print(f"finished second")
asyncio.run(main())
print("--- finished")
def test_progress(self):
from tool.async_util import CAsyncTask, CRspInfo
# Tasks to be performed
async def reqFn(num):
url = "http://149.129.147.44:8305/hotupdate"
reqInfo = {
"Plat": 8,
"Os": 2,
"Appid": 3,
"Uid": '123123',
"Version": '0.0.0.1',
"Deviceid": 'wolegequ',
}
rspInfo = CRspInfo()
try:
async with aiohttp.request(method="POST", url=url, data=json.dumps(reqInfo)) as rsp:
print(f"--- idx: {num} code: {rsp.status}")
rspInfo.code = num
rspInfo.text = await rsp.text()
except:
rspInfo.code = -999
finally:
return rspInfo
async def reqFn01():
print("--- start reqFn01")
await asyncio.sleep(20)
return "hello01"
async def reqFn02():
print("--- start reqFn02")
await asyncio.sleep(10)
return "hello02"
async def reqFn03():
print("--- start reqFn03")
await asyncio.sleep(30)
return "hello03"
taskArr = [reqFn(idx) for idx in range(30)]
res = CAsyncTask().doTask(reqFn01(), reqFn02(), reqFn03(), *taskArr)
print(f"--- finished, res: {utils.beautyJson(res)}")
# asynchronous io http
def test_concurrencyReq(self):
url = "http://149.129.147.44:8305/hotupdate" # Test
# url = "https://www.baidu.com" # Test
reqInfo = {
"Plat": 8,
"Os": 2,
"Appid": 3,
"Uid": '123123',
"Version": '0.0.0.1',
"Deviceid": 'wolegequ',
}
# code, rspDct = utils.httpPost(url, utils.objToJson(reqInfo))
# print(f"--- code: {code}, rsp: {utils.beautyJson(rspDct)}")
# return
async def reqFn(idx):
try:
# async with aiohttp.request(method="GET", url=url) as rsp:
async with aiohttp.request(method="POST", url=url, data=json.dumps(reqInfo)) as rsp:
print(f"--- idx: {idx} code: {rsp.status}")
# response.request_info
res = await rsp.text()
# print(f"--- res: {res}")
return res
except:
return "--- error"
# create task 方式
async def main01():
taskArr = []
for idx in range(5):
task = asyncio.create_task(reqFn(idx)) # Start executing it when you create it.
taskArr.append(task)
resArr = []
for task in taskArr: # Waiting for all requests to complete
res = await task
resArr.append(res)
return resArr
# gather 方式
async def main02():
taskArr = []
for idx in range(5):
task = reqFn(idx)
taskArr.append(task)
return await asyncio.gather(*taskArr)
# True
loop = asyncio.get_event_loop()
resArr = loop.run_until_complete(main02()) # Complete the event loop until the end of the last task
# # Error: RuntimeError: Event loop is closed
# resArr = asyncio.run(main02())
print("--- finished")
print(f"--- resArr: {utils.beautyJson(resArr)}")
def test_compare_http(self):
url = "http://149.129.147.44:8305/hotupdate"
# url = "https://www.baidu.com"
reqCnt = 1
dct = {
"Plat": 8,
"Os": 2,
"Appid": 3,
"Uid": '123123',
"Version": '0.0.0.1',
"Deviceid": 'wolegequ',
}
@utils.call_cost
def syncFn():
print("--- syncFn start")
for idx in range(reqCnt):
code, rspDct = utils.httpPost(url, dct)
print("--- syncFn end")
@utils.call_cost
def asyncFn():
print("--- asyncFn start")
reqArr = []
for idx in range(reqCnt):
ri = async_util.CReqInfo()
ri.url = url
ri.data = dct # dict or json string
ri.method = "POST"
ri.extA = f"extra data {idx}"
reqArr.append(ri)
resArr = async_util.doReq(*reqArr)
print("--- type: {}, len: {}".format(type(resArr), len(resArr)))
# print(f"--- finished, resArr: {utils.beautyJson(resArr)}")
print("--- asyncFn end")
sync_cc = syncFn()
print("sync: {}".format(sync_cc))
print()
async_cc = asyncFn()
print("async: {}".format(async_cc))
def test_asyncFile(self):
async def dealFile(filePath):
print("--- dealFile:", filePath)
async with aiofiles.open(filePath, mode="r") as fd: # read
txt = await fd.read()
print("--- read:", txt)
async with aiofiles.open(filePath, mode="w") as fd: # write
await fd.write("wolegequ")
return "done!!"
path = utils.getDesktop("test_io2/aaa.txt")
res = async_util.doTask(dealFile(path))
print("--- res:", res)
# Asynchronous io file, read by line
def test_asyncLine(self):
async def dealFile(filePath):
print("--- dealFile:", filePath)
async with aiofiles.open(filePath, mode="rb") as fd: # write
async for line in fd:
# print("--- line:", line.decode(encoding="utf-8", errors="ignore"))
print("--- line:", str(line, encoding="utf-8", errors="ignore"))
path = utils.getDesktop("a_temp.lua")
res = async_util.doTask(dealFile(path))
print("--- res:", res)
# Compare file reads, synchronous, asynchronous, time-consuming.
def test_compare_readFile(self):
dstDir = utils.getDesktop("test_io")
fileArr = utils.getFiles(dstDir, ["*.*"])
print("--- fileArr len: {}".format(len(fileArr)))
@utils.call_cost
def syncFn():
print("--- syncFn start")
for file in fileArr:
# time.sleep(0.5)
utils.readFileBytes(file)
print("--- syncFn end")
@utils.call_cost
def asyncFn():
print("--- asyncFn start")
fiArr = [async_util.CFileInfo(file) for file in fileArr]
res = async_util.doRead(*fiArr)
# print("--- res:", utils.beautyJson(res))
# 换个目录写进去
# for fi in fiArr:
# fi.path = fi.path.replace("test_io2", "test_io3")
# async_util.doWrite(*fiArr)
print("--- asyncFn end")
sync_cc = syncFn()
print("sync: {}".format(sync_cc))
print()
async_cc = asyncFn()
print("async: {}".format(async_cc))
# Asynchronous parallel execution of system commands
def test_subprocess(self):
# official document: https://docs.python.org/3/library/asyncio-subprocess.html
async def run(cmd):
proc = await asyncio.create_subprocess_shell(
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE)
stdout, stderr = await proc.communicate()
print(f'[{cmd!r} exited with {proc.returncode}]')
if stdout:
print(f'[stdout]\n{stdout.decode(errors="ignore")}')
if stderr:
print(f'[stderr]\n{stderr.decode(errors="ignore")}')
cmd = "git status"
asyncio.run(run(cmd))
def test_compare_subprocess(self):
cnt = 5
# cmd = "git status"
cmd = "call {}".format(utils.getDesktop("aaa.exe"))
@utils.call_cost
def asyncFn():
cmdArr = []
for i in range(cnt):
ci = async_util.CCmdInfo(cmd)
ci.extA = i
cmdArr.append(ci)
res = async_util.doCmd(*cmdArr)
# print("--- res:", utils.beautyJson(res))
@utils.call_cost
def syncFn():
async def run(command):
return utils.cmdToString(command)
res = async_util.doTask(*[run(cmd) for i in range(cnt)])
# print("--- res:", utils.beautyJson(res))
dt1 = syncFn()
print("--- syncFn cost time:", dt1) # --- syncFn cost time: 00:00:45
dt2 = asyncFn()
print("--- asyncFn cost time:", dt2) # --- asyncFn cost time: 00:00:09
# True multi-threaded parallelism.
def test_multi_thread(self):
def fn001(name):
# print("--- hello 111, name: {}".format(name))
# time.sleep(5)
# print("--- hello 222, name: {}".format(name))
# error
# assert False, "--- wolegequ"
# arr = []
# b = arr[1]
utils.execute("call {}".format(utils.getDesktop("aaa.exe")))
return "world-{}".format(name)
# res Sequential return value
res = async_util.doRun(*[async_util.CThreadInfo(target=fn001, args=(i,)) for i in range(3)])
# print("--- end, res: {}".format(utils.beautyJson([str(ti.result) for ti in res])))
for ti in res:
print("--- result is error:", utils.isError(ti.result))
# print("--- exmsg:", utils.exmsg(ti.result))
if __name__ == "__main__":
ins = Test_Async()
ins.test_multi_thread()
