python協(xié)程庫asyncio(異步io)問題
介紹
- 異步IO：就是發(fā)起一個IO操作（如：網(wǎng)絡請求，文件讀寫等），這些操作一般是比較耗時的，不用等待它結束，可以繼續(xù)做其他事情，結束時會發(fā)來通知。
- 協(xié)程：又稱為微線程，在一個線程中執(zhí)行，執(zhí)行函數(shù)時可以隨時中斷，由程序（用戶）自身控制，執(zhí)行效率極高，與多線程比較，沒有切換線程的開銷和多線程鎖機制。
asyncio中幾個重要概念
1.事件循環(huán)
事件循環(huán)是每個 asyncio 應用的核心，管理所有的事件，在整個程序運行過程中不斷循環(huán)執(zhí)行并追蹤事件發(fā)生的順序將它們放在隊列中，空閑時調用相應的事件處理者來處理這些事件。
創(chuàng)建事件循環(huán)
loop = asyncio.get_event_loop()
獲取當前事件循環(huán)。
如果當前 OS 線程沒有設置當前事件循環(huán)并且 set_event_loop() 還沒有被調用，asyncio 將創(chuàng)建一個新的事件循環(huán)并將其設置為當前循環(huán)。
另起一個線程創(chuàng)建事件循環(huán)
from threading import Thread
import asyncio
def start_thread_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
new_loop = asyncio.new_event_loop()
loop_thread = Thread(target=start_thread_loop, args=(new_loop,))
loop_thread.setDaemon(True) # 守護線程
loop_thread.start()
2.Future
Future對象表示尚未完成的計算，還未完成的結果，它和task上沒有本質上的區(qū)別
3.Task
是Future的子類，作用是在運行某個任務的同時可以并發(fā)的運行多個任務。asyncio.Task用于實現(xiàn)協(xié)作式多任務的庫，且Task對象不能用戶手動實例化，通過下面2個函數(shù)創(chuàng)建：loop.create_task() 或 asyncio.ensure_future()
- loop.create_task() ,要在定義loop對象之后，調用將方法對象轉化成了task的對象
- asyncio.ensure_future() 直接調用asyncio 的ensure_future() 方法，返回的也是task 對象（我們還沒有聲明 loop 也可以提前定義好 task 對象）
4.async/await 關鍵字
asyncio實現(xiàn)了TCP、UDP、SSL等協(xié)議，async定義一個協(xié)程，await用于掛起阻塞的異步調用接口。對于異步io你需要知道的重點，要注意的是，await語法只能出現(xiàn)在通過async修飾的函數(shù)中，否則會報SyntaxError錯誤。而且await后面的對象需要是一個Awaitable，或者實現(xiàn)了相關的協(xié)議。
注意
- 所有需要異步執(zhí)行的函數(shù)，都需要asyncio中的輪詢器去輪詢執(zhí)行，如果函數(shù)阻塞，輪詢器就會去執(zhí)行下一個函數(shù)。所以所有需要異步執(zhí)行的函數(shù)都需要加入到這個輪詢器中。
- 若在協(xié)程中需要有延時操作，應該使用 await asyncio.sleep()，而不是使用time.sleep()，因為使用time.sleep()后會釋放GIL，阻塞整個主線程，從而阻塞整個事件循環(huán)。
創(chuàng)建一個協(xié)程
使用async可以定義協(xié)程對象，使用await可以針對耗時的操作進行掛起，就像生成器里的yield一樣，函數(shù)讓出控制權。協(xié)程遇到await，事件循環(huán)將會掛起該協(xié)程，執(zhí)行別的協(xié)程，直到其他的協(xié)程也掛起或者執(zhí)行完畢，再進行下一個協(xié)程的執(zhí)行
耗時的操作一般是一些IO操作，例如網(wǎng)絡請求，文件讀取等。我們使用asyncio.sleep函數(shù)來模擬IO操作。協(xié)程的目的也是讓這些IO操作異步化。
簡單例子
import asyncio
async def execute(x):
print('Number:', x)
coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')
loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine)
print('After calling loop')
首先我們引入了 asyncio 這個包，這樣我們才可以使用 async 和 await，然后我們使用 async 定義了一個 execute() 方法，方法接收一個數(shù)字參數(shù)，方法執(zhí)行之后會打印這個數(shù)字。
隨后我們直接調用了這個方法，然而這個方法并沒有執(zhí)行，而是返回了一個 coroutine 協(xié)程對象。隨后我們使用 get_event_loop() 方法創(chuàng)建了一個事件循環(huán) loop，并調用了 loop 對象的 run_until_complete() 方法將協(xié)程注冊到事件循環(huán) loop 中，然后啟動。最后我們才看到了 execute() 方法打印了輸出結果。
可見，async 定義的方法就會變成一個無法直接執(zhí)行的 coroutine 對象，必須將其注冊到事件循環(huán)中才可以執(zhí)行。
進階例子
多個任務，定義一個task列表，使用asyncio.gather(*tasks) 或 asyncio.wait(tasks) 接收
import asyncio
import time
now = lambda: time.time()
"""
asyncio.gather主要集中在收集結果上。它等待一堆task并按給定的順序返回結果。
asyncio.wait等待task。而不是直接給你結果，它提供完成和待處理的任務。你必須手工收集結果。
asyncio.wait(tasks) ps:asyncio.wait([1,2,3]) 也可以使用 asyncio.gather(*tasks) ps: asyncio.gather(1,2,3),前者接受一個task列表，后者接收一堆task。
"""
# 定義一個異步任務
async def do_some_work(x):
print("waiting:", x)
# 模擬io阻塞
await asyncio.sleep(x)
return "Done after {}s".format(x)
async def main(loop):
"""
:param loop: loop.create_task（需要傳進loop參數(shù)）
:return: None
"""
coroutine1 = do_some_work(1)
coroutine2 = do_some_work(2)
coroutine3 = do_some_work(4)
# asyncio.ensure_future
tasks = [
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3)
]
# loop.create_task（需要傳進loop參數(shù)）
# tasks = [
# loop.create_task(coroutine1),
# loop.create_task(coroutine2),
# loop.create_task(coroutine3)
# ]
# 返回 完成的 task object
dones, pendings = await asyncio.wait(tasks)
print(dones, pendings)
for task in dones:
print("Task ret:", task.result())
# 返回 task 方法的 返回值
# results = await asyncio.gather(*tasks)
# for result in results:
# print("Task ret:",result)
start = now()
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
print("Time:", now() - start)
gather和wait 的區(qū)別
把多個協(xié)程注冊進一個事件循環(huán)中的兩種方法
使用方式區(qū)別
1.使用asyncio.wait()
loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks))
2.使用asyncio.gather()
loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.gather(*tasks)) # *接收args參數(shù)
接收參數(shù)區(qū)別
asyncio.wait
參數(shù)必須是list對象 ，list 對象存放多個 task object
用asyncio.ensure_future轉為task對象
tasks=[
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3)
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
不轉為task對象
loop = asyncio.get_event_loop()
tasks=[
coroutine1,
coroutine2,
coroutine3
]
loop.run_until_complete(asyncio.wait(tasks))
asyncio.gather
必須用 * 來接收 list 對象
tasks=[
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3)
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(*tasks))
返回結果區(qū)別
asyncio.wait
asyncio.wait返回dones和pendings
- dones：表示已經(jīng)完成的任務
- pendings：表示未完成的任務
我們需要手動去獲取結果
dones, pendings = await asyncio.wait(tasks)
print(dones, pendings)
for task in dones:
print("Task ret:", task.result())
asyncio.gather
它的返回值就是 return的結果 ，不用再task.result() 來獲取
# 返回 task 方法的 返回值
results = await asyncio.gather(*tasks)
for result in results:
print("Task ret:",result)
另 asyncio.wait 帶有控制功能
【控制運行任務數(shù)】：運行第一個任務就返回
- FIRST_COMPLETED ：第一個任務完全返回
- FIRST_EXCEPTION：產(chǎn)生第一個異常返回
- ALL_COMPLETED：所有任務完成返回 （默認選項）
import asyncio
import random
async def coro(tag):
print(">", tag)
await asyncio.sleep(random.uniform(0.5, 5))
print("<", tag)
return tag
loop = asyncio.get_event_loop()
tasks = [coro(i) for i in range(1, 11)]
# 第一次wait 完成情況
print("Get first result:")
finished, unfinished = loop.run_until_complete(
asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)) # 第一個任務完全返回
for task in finished:
print(task.result())
print("unfinished:", len(unfinished))
# 繼續(xù)第一次未完成任務
print("Get more results in 2 seconds:")
finished2, unfinished2 = loop.run_until_complete(
asyncio.wait(unfinished, timeout=2)) # 超時2s 返回
for task in finished2:
print(task.result())
print("unfinished2:", len(unfinished2))
# 繼續(xù)第2次未完成任務
print("Get all other results:")
finished3, unfinished3 = loop.run_until_complete(asyncio.wait(unfinished2)) # ALL_COMPLETED：所有任務完成返回 （默認項）
for task in finished3:
print(task.result())
loop.close()
動態(tài)添加協(xié)程
很多時候，我們的事件循環(huán)用于注冊協(xié)程，而有的協(xié)程需要動態(tài)的添加到事件循環(huán)中。一個簡單的方式就是使用多線程。當前線程創(chuàng)建一個事件循環(huán)，然后在新建一個線程，在新線程中啟動事件循環(huán)。當前線程不會被block
相關函數(shù)介紹：
- loop.call_soon_threadsafe() ：與 call_soon()類似，等待此函數(shù)返回后馬上調用回調函數(shù)，返回值是一個 asyncio.Handle 對象，此對象內(nèi)只有一個方法為 cancel()方法，用來取消回調函數(shù)。
- loop.call_soon() ： 與call_soon_threadsafe()類似，call_soon_threadsafe() 是線程安全的
- loop.call_later()：延遲多少秒后執(zhí)行回調函數(shù)
- loop.call_at()：在指定時間執(zhí)行回調函數(shù)，這里的時間統(tǒng)一使用 loop.time() 來替代 time.sleep()
- asyncio.run_coroutine_threadsafe()： 動態(tài)的加入?yún)f(xié)程，參數(shù)為一個回調函數(shù)和一個loop對象，返回值為future對象，通過future.result()獲取回調函數(shù)返回值
動態(tài)添加協(xié)程同步方式通過調用 call_soon_threadsafe()函數(shù)，傳入一個回調函數(shù)callback和一個位置參數(shù)
注意：同步方式，回調函數(shù) more_work()為普通函數(shù)
import asyncio
from threading import Thread
import time
now = lambda: time.time()
def start_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
def more_work(x):
print('More work {}'.format(x))
time.sleep(x)
print('Finished more work {}'.format(x))
start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
print('TIME: {}'.format(time.time() - start))
new_loop.call_soon_threadsafe(more_work, 6)
new_loop.call_soon_threadsafe(more_work, 3)
print('here')
啟動上述代碼之后，當前線程不會被block，新線程中會按照順序執(zhí)行call_soon_threadsafe方法注冊的more_work方法， 后者因為time.sleep操作是同步阻塞的，因此運行完畢more_work需要大致6 + 3
異步方式
import asyncio
import time
from threading import Thread
now = lambda: time.time()
def start_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
async def do_some_work(x):
print('Waiting {}'.format(x))
await asyncio.sleep(x)
print('Done after {}s'.format(x))
start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
print('TIME: {}'.format(time.time() - start))
asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)
asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)
上述的例子，主線程中創(chuàng)建一個new_loop，然后在另外的子線程中開啟一個無限事件循環(huán)。
主線程通過run_coroutine_threadsafe新注冊協(xié)程對象。這樣就能在子線程中進行事件循環(huán)的并發(fā)操作，同時主線程又不會被block。一共執(zhí)行的時間大概在6s左右。
協(xié)程的停止
future對象有幾個狀態(tài)：
PendingRunningDoneCacelled
創(chuàng)建future的時候，task為pending，事件循環(huán)調用執(zhí)行的時候當然就是running，調用完畢自然就是done，如果需要停止事件循環(huán)，就需要先把task取消?？梢允褂胊syncio.Task獲取事件循環(huán)的task
import asyncio
import time
now = lambda: time.time()
async def do_some_work(x):
print("Waiting:", x)
await asyncio.sleep(x)
return "Done after {}s".format(x)
coroutine1 = do_some_work(1)
coroutine2 = do_some_work(2)
coroutine3 = do_some_work(2)
tasks = [
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3),
]
start = now()
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(asyncio.wait(tasks))
except KeyboardInterrupt as e:
print(asyncio.Task.all_tasks())
for task in asyncio.Task.all_tasks():
print(task.cancel())
loop.stop()
loop.run_forever()
finally:
loop.close()
print("Time:", now() - start)
啟動事件循環(huán)之后，馬上ctrl+c，會觸發(fā)run_until_complete的執(zhí)行異常 KeyBorardInterrupt。然后通過循環(huán)asyncio.Task取消future。
True表示cannel成功，loop stop之后還需要再次開啟事件循環(huán)，最后在close，不然還會拋出異常
循環(huán)task，逐個cancel是一種方案，可是正如上面我們把task的列表封裝在main函數(shù)中，main函數(shù)外進行事件循環(huán)的調用。這個時候，main相當于最外出的一個task，那么處理包裝的main函數(shù)即可。
協(xié)程中生產(chǎn)-消費模型設計
通過上面的動態(tài)添加協(xié)程的思想，我們可以設計一個生產(chǎn)-消費的模型，至于中間件（管道）是什么無所謂，下面以內(nèi)置隊列和redis隊列來舉例說明。
提示：若想主線程退出時，子線程也隨之退出，需要將子線程設置為守護線程，函數(shù) setDaemon(True)
import asyncio
from threading import Thread
from collections import deque
import random
import time
def start_thread_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
def consumer():
while True:
if dq:
msg = dq.pop()
if msg:
asyncio.run_coroutine_threadsafe(thread_example('Zarten'+ msg), new_loop)
async def thread_example(name):
print('正在執(zhí)行name:', name)
await asyncio.sleep(2)
return '返回結果：' + name
dq = deque()
new_loop = asyncio.new_event_loop()
loop_thread = Thread(target= start_thread_loop, args=(new_loop,))
loop_thread.setDaemon(True)
loop_thread.start()
consumer_thread = Thread(target= consumer)
consumer_thread.setDaemon(True)
consumer_thread.start()
while True:
i = random.randint(1, 10)
dq.appendleft(str(i))
time.sleep(2)
redis隊列模型
生產(chǎn)者代碼：
import redis
conn_pool = redis.ConnectionPool(host='127.0.0.1')
redis_conn = redis.Redis(connection_pool=conn_pool)
redis_conn.lpush('coro_test', '1')
redis_conn.lpush('coro_test', '2')
redis_conn.lpush('coro_test', '3')
redis_conn.lpush('coro_test', '4')
消費者代碼：
import asyncio
from threading import Thread
import redis
def get_redis():
conn_pool = redis.ConnectionPool(host= '127.0.0.1')
return redis.Redis(connection_pool= conn_pool)
def start_thread_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
async def thread_example(name):
print('正在執(zhí)行name:', name)
await asyncio.sleep(2)
return '返回結果：' + name
redis_conn = get_redis()
new_loop = asyncio.new_event_loop()
loop_thread = Thread(target= start_thread_loop, args=(new_loop,))
loop_thread.setDaemon(True)
loop_thread.start()
#循環(huán)接收redis消息并動態(tài)加入?yún)f(xié)程
while True:
msg = redis_conn.rpop('coro_test')
if msg:
asyncio.run_coroutine_threadsafe(thread_example('Zarten' + bytes.decode(msg, 'utf-8')), new_loop)
asyncio在aiohttp中的應用
aiohttp是一個異步庫，分為客戶端和服務端，下面只是簡單對客戶端做個介紹以及一個經(jīng)常遇到的異常情況。
aiohttp客戶端為異步網(wǎng)絡請求庫
import asyncio
import aiohttp
count = 0
async def get_http(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as res:
global count
count += 1
print(count, res.status)
def main():
loop = asyncio.get_event_loop()
url = 'https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&ch=&tn=baiduerr&bar=&wd={0}'
tasks = [get_http(url.format(i)) for i in range(10)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
if __name__ == '__main__':
main()
aiohttp并發(fā)量太大的異常解決方案
在使用aiohttp客戶端進行大量并發(fā)請求時，程序會拋出 ValueError: too many file descriptors in select() 的錯誤。
異常代碼示例
說明：測試機器為windows系統(tǒng)
import asyncio
import aiohttp
count = 0
async def get_http(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as res:
global count
count += 1
print(count, res.status)
def main():
loop = asyncio.get_event_loop()
url = 'https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&ch=&tn=baiduerr&bar=&wd={0}'
tasks = [get_http(url.format(i)) for i in range(600)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
if __name__ == '__main__':
main()
原因分析：使用aiohttp時，python內(nèi)部會使用select()，操作系統(tǒng)對文件描述符最大數(shù)量有限制，linux為1024個，windows為509個。
解決方案：
最常見的解決方案是：限制并發(fā)數(shù)量（一般500），若并發(fā)的量不大可不作限制。其他方案這里不做介紹，如windows下使用loop = asyncio.ProactorEventLoop() 以及使用回調方式等
限制并發(fā)數(shù)量方法
提示：此方法也可用來作為異步爬蟲的限速方法（反反爬）
使用semaphore = asyncio.Semaphore(500) 以及在協(xié)程中使用 async with semaphore: 操作
具體代碼如下：
import asyncio
import aiohttp
async def get_http(url):
async with semaphore:
async with aiohttp.ClientSession() as session:
async with session.get(url) as res:
global count
count += 1
print(count, res.status)
if __name__ == '__main__':
count = 0
semaphore = asyncio.Semaphore(500)
loop = asyncio.get_event_loop()
url = 'https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&ch=&tn=baiduerr&bar=&wd={0}'
tasks = [get_http(url.format(i)) for i in range(600)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
在線程或進程池中執(zhí)行代碼
在《流暢的python》中有這樣一段話。
函數(shù)(例如io讀寫，requests網(wǎng)絡請求)阻塞了客戶代碼與asycio事件循環(huán)的唯一線程，因此在執(zhí)行調用時，整個應用程序都會凍結。這個問題的解決方法是，使用事件循環(huán)對象的 run_in_executor方法。
asyncio的事件循環(huán)在背后維護著一個ThreadPoolExecutor對象，我們可以調用run_in_executor方法，把可調用對象發(fā)給它執(zhí)行。
import asyncio
from time import sleep, strftime
from concurrent import futures
executor = futures.ThreadPoolExecutor(max_workers=5)
async def blocked_sleep(name, t):
print(strftime('[%H:%M:%S]'), end=' ')
print('sleep {} is running {}s'.format(name, t))
loop = asyncio.get_event_loop()
await loop.run_in_executor(executor, sleep, t)
print(strftime('[%H:%M:%S]'), end=' ')
print('sleep {} is end'.format(name))
return t
async def main():
future = (blocked_sleep(i, i) for i in range(1, 6))
fs = asyncio.gather(*future)
return await fs
loop = asyncio.get_event_loop()
results = loop.run_until_complete(main())
print('results: {}'.format(results))
在同一個線程里，兩個 event loop 無法同時 run，但這不能阻止您用兩個線程分別跑兩個 event loop,
其次再說 ThreadPoolExecutor。您也可以看到，它根本不是 asyncio 庫的東西。當您創(chuàng)建一個 ThreadPoolExecutor 對象時，您實際上是創(chuàng)建了一個線程池。
僅此而已，與 asyncio、event loop 并無瓜葛。而當您明確使用一個 event loop 的 run_in_executor() 方法時，其實底層做的只有兩件事：
1,用線程池執(zhí)行給定函數(shù)，與 asyncio 毫無關系；
2,給線程池執(zhí)行結果增加一個回調，該回調會在 event loop 的下一次循環(huán)中保存執(zhí)行結果。
所以 run_in_executor() 只是將傳統(tǒng)的線程池結果拉回到給定 event loop 中，以便進一步處理而已，不存在誰共享誰的關系，指定誰是誰。您可以嘗試一下，在多個線程中跑多個 event loop，然后都向同一個線程池扔任務，然后返回結果：
import asyncio
import threading
import time
from concurrent.futures.thread import ThreadPoolExecutor
e = ThreadPoolExecutor()
def worker(index):
print(index, 'before:', time.strftime('%X'))
time.sleep(1)
print(index, 'after:', time.strftime('%X'))
return index
def main(index):
loop = asyncio.new_event_loop()
res = loop.run_until_complete(loop.run_in_executor(e, worker, index))
print('Thread', index, 'got result', res)
threads = []
for i in range(5):
t = threading.Thread(target=main, args=(i,))
t.start()
threads.append(t)
for t in threads:
t.join()
不同于上面的方法，這里是把阻塞的方法放到新的線程里跑。
總結
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
基于python實現(xiàn)MQTT發(fā)布訂閱過程原理解析
這篇文章主要介紹了基于python實現(xiàn)MQTT發(fā)布訂閱過程原理解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-07-07
今天小編就為大家分享一篇對Django外鍵關系的描述，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-07-07
Python實現(xiàn)統(tǒng)計代碼行的方法分析
這篇文章主要介紹了Python實現(xiàn)統(tǒng)計代碼行的方法,結合實例形式分析了Python針對代碼行數(shù)的計算實現(xiàn)步驟與操作技巧,需要的朋友可以參考下2017-07-07
使用pyecharts在jupyter notebook上繪圖
這篇文章主要介紹了使用pyecharts在jupyter notebook上繪圖，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2017-07-07
這篇文章主要介紹了pytorch模型的保存和加載、checkpoint操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-06-06
Python re.split方法分割字符串的實現(xiàn)示例
本文主要介紹了Python re.split方法分割字符串的實現(xiàn)示例，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-08-08
python使用SimpleXMLRPCServer實現(xiàn)簡單的rpc過程
這篇文章主要介紹了python使用SimpleXMLRPCServer實現(xiàn)簡單的rpc過程，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-06-06
這篇文章主要為大家詳細介紹了python游戲地圖最短路徑的求解，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-01-01
python re.sub()替換正則的匹配內(nèi)容方法
今天小編就為大家分享一篇python re.sub()替換正則的匹配內(nèi)容方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-07-07
Python實戰(zhàn)之能監(jiān)控文件變化的神器—看門狗
這篇文章主要介紹了Python實戰(zhàn)之能監(jiān)控文件變化的神器—看門狗,文中有非常詳細的圖文及代碼示例,對正在學習python的小伙伴們有非常好的幫助,需要的朋友可以參考下2021-05-05