
编程开发
切片copy
当我们需要在Go语言中复制一个切片时,我们可以使用内置的copy函数。copy函数可以将一个切片的元素复制到另一个切片中。下面是一个示例: 在上面的示例中,我们首先创建了一个源切片
1、异步程序可以在等待其最终结果的同时“暂停”并让其他程序同时运行。 2、通过上述机制,异步代码有助于并发执行。换句话说,异步代码表现出了并发的特点。异步IO:一种与语言无关的范例(模型) ,很多编程语言都有这种实现,它是一种单线程,单进程设计:它使用协作多任务处理,尽管在单个进程中使用单个线程,异步 IO 仍具有并发的感觉。async/await:两个用于定义协程的新 Python 关键字。asyncio:Python 标准库,为运行和管理协程提供了基础和 API。
#!/usr/bin/env python3
#~~~~www.wulaoer.org 吴老二个人博客~~~~
# countasync.py
import asyncio
async def count() :
print("One")
await asyncio.sleep(1)
print("Two")
async def main() :
await asyncio.gather(count() , count() , count())
if __name__ == "__main__":
import time
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"{__file__} executed in {elapsed:0.2f} seconds.")
执行结果
One One One Two Two Two e:\countasync.py executed in 1.01 seconds.这里使用 time.sleep() 和 asyncio.sleep() 是有区别的,time.sleep() 可以表示任何耗时的阻塞函数调用,而 asyncio.sleep 不阻塞,可将 CPU 的控制权交给下一个协程。
async def g() :
# Pause here and come back to g() when f() is ready
r = await f()
return r
这里的 await 挂起了本次协程的执行。如果 Python 在 g() 范围内遇到 await f() 表达式,那就意味着,“暂停 g() 的执行,直到我等待f() 返回结果。同时,让其他协程运行。”。当然也有一些规则要求什么地方可以使用 async/await 关键字,什么地方不能用:
async def f(x) :
y = await z(x) # OK - `await` and `return` allowed in coroutines
return y
async def g(x) :
yield x # OK - this is an async generator
async def m(x) :
yield from gen(x) # No - SyntaxError
def m(x) :
y = await z(x) # Still no - SyntaxError (no `async def` here)
return y
言归正传,让我们来看一个更复杂的例子:给定一个产生随机数的函数 makerandom,它产生一个 0-9 之间的随机数,直到超过给定数据后结束,让此协程的多个调用无需等待彼此即可执行,代码如下
import asyncio
import random
#~~~~www.wulaoer.org 吴老二个人博客~~~~
# ANSI colors
c = (
"\033[0m", # End of color
"\033[36m", # Cyan
"\033[91m", # Red
"\033[35m", # Magenta
)
async def makerandom(idx: int, threshold: int = 6) -> int:
print(c[idx + 1] + f"Initiated makerandom({idx}) .")
i = random.randint(0, 10)
while i <= threshold:
print(c[idx + 1] + f"makerandom({idx}) == {i} too low; retrying.")
await asyncio.sleep(idx + 1)
i = random.randint(0, 10)
print(c[idx + 1] + f"---> Finished: makerandom({idx}) == {i}" + c[0])
return i
async def main() :
res = await asyncio.gather(*(makerandom(i, 10 - i - 1) for i in range(3)) )
return res
if __name__ == "__main__":
random.seed(444)
r1, r2, r3 = asyncio.run(main())
print()
print(f"r1: {r1}, r2: {r2}, r3: {r3}")
执行结果:
Initiated makerandom(0) .
makerandom(0) == 4 too low; retrying.
Initiated makerandom(1) .
makerandom(1) == 4 too low; retrying.
Initiated makerandom(2) .
makerandom(2) == 0 too low; retrying.
makerandom(0) == 4 too low; retrying.
makerandom(1) == 7 too low; retrying.
makerandom(0) == 4 too low; retrying.
makerandom(2) == 4 too low; retrying.
makerandom(0) == 8 too low; retrying.
---> Finished: makerandom(1) == 10
makerandom(0) == 7 too low; retrying.
makerandom(0) == 8 too low; retrying.
makerandom(2) == 4 too low; retrying.
makerandom(0) == 7 too low; retrying.
makerandom(0) == 1 too low; retrying.
makerandom(0) == 6 too low; retrying.
---> Finished: makerandom(2) == 9
makerandom(0) == 3 too low; retrying.
makerandom(0) == 9 too low; retrying.
makerandom(0) == 7 too low; retrying.
---> Finished: makerandom(0) == 10
r1: 10, r2: 10, r3: 9
该程序使用一个主协程 makerandom() ,并在 3 个不同的输入上同时运行它。大多数程序将包含小型模块化协程和一个包装器功能,用于将每个较小的协程链接在一起。main() 通过在一些可迭代或池中映射中央协程来收集任务。在此示例中,池为 range(3) 。尽管“制作随机整数”(比CPU绑定更多的东西) 可能不是 asyncio 的最佳选择,但在示例中正是 asyncio.sleep() 旨在模仿 IO 绑定的过程等待时间不确定的地方。例如,asyncio.sleep() 调用可以表示在消息应用程序的两个客户端之间发送和接收的延迟。
#!/usr/bin/env python3
#~~~~www.wulaoer.org 吴老二个人博客~~~~
# chained.py
import asyncio
import random
import time
async def part1(n: int) -> str:
i = random.randint(0, 10)
print(f"part1({n}) sleeping for {i} seconds.")
await asyncio.sleep(i)
result = f"result{n}-1"
print(f"Returning part1({n}) == {result}.")
return result
async def part2(n: int, arg: str) -> str:
i = random.randint(0, 10)
print(f"part2{n, arg} sleeping for {i} seconds.")
await asyncio.sleep(i)
result = f"result{n}-2 derived from {arg}"
print(f"Returning part2{n, arg} == {result}.")
return result
async def chain(n: int) -> None:
start = time.perf_counter()
p1 = await part1(n)
p2 = await part2(n, p1)
end = time.perf_counter() - start
print(f"-->Chained result{n} => {p2} (took {end:0.2f} seconds) .")
async def main(*args) :
await asyncio.gather(*(chain(n) for n in args))
if __name__ == "__main__":
import sys
random.seed(444)
args = [1, 2, 3] if len(sys.argv) == 1 else map(int, sys.argv[1:])
start = time.perf_counter()
asyncio.run(main(*args))
end = time.perf_counter() - start
print(f"Program finished in {end:0.2f} seconds.")
注意观察输出,part1() 睡眠一段时间,part2() 在结果可用时开始处理它们:
$ python3 chained.py 9 6 3
part1(9) sleeping for 4 seconds.
part1(6) sleeping for 4 seconds.
part1(3) sleeping for 0 seconds.
Returning part1(3) == result3-1.
part2(3, 'result3-1') sleeping for 4 seconds.
Returning part1(9) == result9-1.
part2(9, 'result9-1') sleeping for 7 seconds.
Returning part1(6) == result6-1.
part2(6, 'result6-1') sleeping for 4 seconds.
Returning part2(3, 'result3-1') == result3-2 derived from result3-1.
-->Chained result3 => result3-2 derived from result3-1 (took 4.00 seconds) .
Returning part2(6, 'result6-1') == result6-2 derived from result6-1.
-->Chained result6 => result6-2 derived from result6-1 (took 8.01 seconds) .
Returning part2(9, 'result9-1') == result9-2 derived from result9-1.
-->Chained result9 => result9-2 derived from result9-1 (took 11.01 seconds) .
Program finished in 11.01 seconds.
在以上代码中,main() 的运行时间将等于它收集在一起的协程任务的最大运行时间。
#!/usr/bin/env python3
#~~~~www.wulaoer.org 吴老二个人博客~~~~
# asyncq.py
import asyncio
import itertools as it
import os
import random
import time
async def makeitem(size: int = 5) -> str:
return os.urandom(size) .hex()
async def randsleep(a: int = 1, b: int = 5, caller=None) -> None:
i = random.randint(0, 10)
if caller:
print(f"{caller} sleeping for {i} seconds.")
await asyncio.sleep(i)
async def produce(name: int, q: asyncio.Queue) -> None:
n = random.randint(0, 10)
for _ in it.repeat(None, n) : # Synchronous loop for each single producer
await randsleep(caller=f"Producer {name}")
i = await makeitem()
t = time.perf_counter()
await q.put((i, t))
print(f"Producer {name} added <{i}> to queue.")
async def consume(name: int, q: asyncio.Queue) -> None:
while True:
await randsleep(caller=f"Consumer {name}")
i, t = await q.get()
now = time.perf_counter()
print(f"Consumer {name} got element <{i}>"
f" in {now-t:0.5f} seconds.")
q.task_done()
async def main(nprod: int, ncon: int) :
q = asyncio.Queue()
producers = [asyncio.create_task(produce(n, q)) for n in range(nprod) ]
consumers = [asyncio.create_task(consume(n, q)) for n in range(ncon) ]
await asyncio.gather(*producers)
await q.join() # Implicitly awaits consumers, too
for c in consumers:
c.cancel()
if __name__ == "__main__":
import argparse
random.seed(444)
parser = argparse.ArgumentParser()
parser.add_argument("-p", "--nprod", type=int, default=5)
parser.add_argument("-c", "--ncon", type=int, default=10)
ns = parser.parse_args()
start = time.perf_counter()
asyncio.run(main(**ns.__dict__))
elapsed = time.perf_counter() - start
print(f"Program completed in {elapsed:0.5f} seconds.")
第一个协程是辅助函数,它们返回一个随机字符串。生产者将 1 到 5 个项目放入队列中。每个项目都是(i,t) 的元组,其中 i 是随机字符串,t 是生产者尝试将元组放入队列的时间。消费者将商品拉出时,它仅使用放入商品的时间戳来计算商品在队列中的经过时间。请记住,asyncio.sleep() 用于模仿其他一些更复杂的协程,如果这是常规的阻止函数,则会消耗时间并阻止所有其他执行。这是由两个生产者和五个消费者进行的测试结果:
$ python3 asyncq.py -p 2 -c 5
Producer 0 sleeping for 3 seconds.
Producer 1 sleeping for 3 seconds.
Consumer 0 sleeping for 4 seconds.
Consumer 1 sleeping for 3 seconds.
Consumer 2 sleeping for 3 seconds.
Consumer 3 sleeping for 5 seconds.
Consumer 4 sleeping for 4 seconds.
Producer 0 added <377b1e8f82> to queue.
Producer 0 sleeping for 5 seconds.
Producer 1 added <413b8802f8> to queue.
Consumer 1 got element <377b1e8f82> in 0.00013 seconds.
Consumer 1 sleeping for 3 seconds.
Consumer 2 got element <413b8802f8> in 0.00009 seconds.
Consumer 2 sleeping for 4 seconds.
Producer 0 added <06c055b3ab> to queue.
Producer 0 sleeping for 1 seconds.
Consumer 0 got element <06c055b3ab> in 0.00021 seconds.
Consumer 0 sleeping for 4 seconds.
Producer 0 added <17a8613276> to queue.
Consumer 4 got element <17a8613276> in 0.00022 seconds.
Consumer 4 sleeping for 5 seconds.
Program completed in 9.00954 seconds.
在这种情况下,生产或消费的每一个处理过程将在几分之一秒内完成处理。延迟可能有两个原因:
$ cat urls.txt
https://regex101.com/
https://docs.python.org/3/this-url-will-404.html
https://www.nytimes.com/guides/
https://www.mediamatters.org/
https://1.1.1.1/
https://www.politico.com/tipsheets/morning-money
https://www.bloomberg.com/markets/economics
https://www.ietf.org/rfc/rfc2616.txt
列表中的第二个 URL 应该返回 404 响应,你需要对其进行适当处理。在实际使用中,可能需要处理比这更棘手的问题,例如服务器断开连接和无止尽的重定向。请求本身应使用单个会话发出,以充分利用会话的内部连接池。让我们看一下完整的程序:
#!/usr/bin/env python3
#~~~~www.wulaoer.org 吴老二个人博客~~~~
# areq.py
"""Asynchronously get links embedded in multiple pages' HMTL."""
import asyncio
import logging
import re
import sys
from typing import IO
import urllib.error
import urllib.parse
import aiofiles
import aiohttp
from aiohttp import ClientSession
logging.basicConfig(
format="%(asctime) s %(levelname) s:%(name) s: %(message) s",
level=logging.DEBUG,
datefmt="%H:%M:%S",
stream=sys.stderr,
)
logger = logging.getLogger("areq")
logging.getLogger("chardet.charsetprober") .disabled = True
HREF_RE = re.compile(r'href="(.*?) "')
async def fetch_html(url: str, session: ClientSession, **kwargs) -> str:
"""GET request wrapper to fetch page HTML.
kwargs are passed to `session.request() `.
"""
resp = await session.request(method="GET", url=url, **kwargs)
resp.raise_for_status()
logger.info("Got response [%s] for URL: %s", resp.status, url)
html = await resp.text()
return html
async def parse(url: str, session: ClientSession, **kwargs) -> set:
"""Find HREFs in the HTML of `url`."""
found = set()
try:
html = await fetch_html(url=url, session=session, **kwargs)
except (
aiohttp.ClientError,
aiohttp.http_exceptions.HttpProcessingError,
) as e:
logger.error(
"aiohttp exception for %s [%s]: %s",
url,
getattr(e, "status", None) ,
getattr(e, "message", None) ,
)
return found
except Exception as e:
logger.exception(
"Non-aiohttp exception occured: %s", getattr(e, "__dict__", {})
)
return found
else:
for link in HREF_RE.findall(html) :
try:
abslink = urllib.parse.urljoin(url, link)
except (urllib.error.URLError, ValueError) :
logger.exception("Error parsing URL: %s", link)
pass
else:
found.add(abslink)
logger.info("Found %d links for %s", len(found) , url)
return found
async def write_one(file: IO, url: str, **kwargs) -> None:
"""Write the found HREFs from `url` to `file`."""
res = await parse(url=url, **kwargs)
if not res:
return None
async with aiofiles.open(file, "a") as f:
for p in res:
await f.write(f"{url}\t{p}\n")
logger.info("Wrote results for source URL: %s", url)
async def bulk_crawl_and_write(file: IO, urls: set, **kwargs) -> None:
"""Crawl & write concurrently to `file` for multiple `urls`."""
async with ClientSession() as session:
tasks = []
for url in urls:
tasks.append(
write_one(file=file, url=url, session=session, **kwargs)
)
await asyncio.gather(*tasks)
if __name__ == "__main__":
import pathlib
import sys
assert sys.version_info >= (3, 7) , "Script requires Python 3.7+."
here = pathlib.Path(__file__) .parent
with open(here.joinpath("urls.txt")) as infile:
urls = set(map(str.strip, infile))
outpath = here.joinpath("foundurls.txt")
with open(outpath, "w") as outfile:
outfile.write("source_url\tparsed_url\n")
asyncio.run(bulk_crawl_and_write(file=outpath, urls=urls))
如果你想了解更多内容,请在 https://github.com/realpython/materials/tree/master/asyncio-walkthrough 查看。这是所有执行过程的结果,areq.py 在一秒钟之内即可获取,解析并保存9个网址的结果:
$ python3 areq.py
21:33:22 DEBUG:asyncio: Using selector: KqueueSelector
21:33:22 INFO:areq: Got response [200] for URL: https://www.mediamatters.org/
21:33:22 INFO:areq: Found 115 links for https://www.mediamatters.org/
21:33:22 INFO:areq: Got response [200] for URL: https://www.nytimes.com/guides/
21:33:22 INFO:areq: Got response [200] for URL: https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Got response [200] for URL: https://www.ietf.org/rfc/rfc2616.txt
21:33:22 ERROR:areq: aiohttp exception for https://docs.python.org/3/this-url-will-404.html [404]: Not Found
21:33:22 INFO:areq: Found 120 links for https://www.nytimes.com/guides/
21:33:22 INFO:areq: Found 143 links for https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Wrote results for source URL: https://www.mediamatters.org/
21:33:22 INFO:areq: Found 0 links for https://www.ietf.org/rfc/rfc2616.txt
21:33:22 INFO:areq: Got response [200] for URL: https://1.1.1.1/
21:33:22 INFO:areq: Wrote results for source URL: https://www.nytimes.com/guides/
21:33:22 INFO:areq: Wrote results for source URL: https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Got response [200] for URL: https://www.bloomberg.com/markets/economics
21:33:22 INFO:areq: Found 3 links for https://www.bloomberg.com/markets/economics
21:33:22 INFO:areq: Wrote results for source URL: https://www.bloomberg.com/markets/economics
21:33:23 INFO:areq: Found 36 links for https://1.1.1.1/
21:33:23 INFO:areq: Got response [200] for URL: https://regex101.com/
21:33:23 INFO:areq: Found 23 links for https://regex101.com/
21:33:23 INFO:areq: Wrote results for source URL: https://regex101.com/
21:33:23 INFO:areq: Wrote results for source URL: https://1.1.1.1/
>>> import asyncio
>>> async def coro(seq) -> list:
... """'IO' wait time is proportional to the max element."""
... await asyncio.sleep(max(seq))
... return list(reversed(seq))
...
>>> async def main():
... # This is a bit redundant in the case of one task
... # We could use `await coro([3, 2, 1])` on its own
... t = asyncio.create_task(coro([3, 2, 1])) # Python 3.7+
... await t
... print(f't: type {type(t)}')
... print(f't done: {t.done()}')
...
>>> t = asyncio.run(main())
t: type <class '_asyncio.Task'>
t done: True
如果不在 main() 中使用 await ,它可能会在 main() 本身完成之前就结束了。因为 asyncio.run(main()) 调用 loop.run_until_complete(main()) ,所以事件循环仅关注 main() 完成,而不关注在 main() 中创建的任务是否完成。如果不等待,循环的其他任务可能会在完成之前被取消。如果需要获取当前待处理任务的列表,则可以使用 asyncio.Task.all_tasks() 。另外,还有 asyncio.gather() 。尽管它没有做任何特别的事情,但是 collect() 的目的是将协程的集合整齐地放在一个 future。结果,它返回一个 future 的对象,如果你等待 asyncio.gather() 并指定多个任务或协程,则你正在等待所有这些任务或协程完成。这与我们前面的示例有点类似 queue.join() ,collect() 的结果将是输入结果的列表:
>>> import time
>>> async def main():
... t = asyncio.create_task(coro([3, 2, 1]))
... t2 = asyncio.create_task(coro([10, 5, 0])) # Python 3.7+
... print('Start:', time.strftime('%X'))
... a = await asyncio.gather(t, t2)
... print('End:', time.strftime('%X')) # Should be 10 seconds
... print(f'Both tasks done: {all((t.done(), t2.done()))}')
... return a
...
>>> a = asyncio.run(main())
Start: 16:20:11
End: 16:20:21
Both tasks done: True
>>> a
[[1, 2, 3], [0, 5, 10]]
你可能已经注意到,collect() 等待传递给它的 Future 或协程的整个结果集。你可以遍历 asyncio.as_completed() 以按完成顺序获取任务结果。该函数返回一个迭代器,该迭代器在完成任务时产生任务。下面,在coro([10,5,0]) 完成之前,可使用coro([3,2,1]) 的结果,而用 collect() 则不是这样:
>>> async def main():
... t = asyncio.create_task(coro([3, 2, 1]))
... t2 = asyncio.create_task(coro([10, 5, 0]))
... print('Start:', time.strftime('%X'))
... for res in asyncio.as_completed((t, t2)):
... compl = await res
... print(f'res: {compl} completed at {time.strftime("%X")}')
... print('End:', time.strftime('%X'))
... print(f'Both tasks done: {all((t.done(), t2.done()))}')
...
>>> a = asyncio.run(main())
Start: 09:49:07
res: [1, 2, 3] completed at 09:49:10
res: [0, 5, 10] completed at 09:49:17
End: 09:49:17
Both tasks done: True
#!/usr/bin/env python3
#~~~~www.wulaoer.org 吴老二个人博客~~~~
# screenshot.py
import asyncio
from pyppeteer import launch
async def main():
browser = await launch()
page = await browser.newPage()
await page.goto('http://example.com')
await page.screenshot({'path': 'example.png'})
await browser.close()
asyncio.run(main())
上面代码中,启动浏览器、打开新 Tab、访问网址、截图、关闭浏览器,这一系列操作都是异步任务,使用 await 命令写起来非常自然简单。执行这个脚本,当前目录下就会生成截图文件 example.png。如果脚本执行时报错 No usable sandbox!,可以参考这里。另外,第一次执行这个脚本,会下载安装 Puppeteer,可能需要等待较长时间,但是此后的执行就会很快。Pyppeteer 的官网还有其他实例,比如向网页注入 JavaScript 代码。
评论 0