如何在Python中編寫并發(fā)程序

多進程/多線程+Queue

創(chuàng)新互聯(lián)建站成立與2013年，先為西青等服務建站，西青等地企業(yè)，進行企業(yè)商務咨詢服務。為西青企業(yè)網(wǎng)站制作PC+手機+微官網(wǎng)三網(wǎng)同步一站式服務解決您的所有建站問題。

一般來說,在Python中編寫并發(fā)程序的經(jīng)驗是:計算密集型任務使用多進程,IO密集型任務使用多進程或者多線程.另外,因為涉及到資源共享,所以需要同步鎖等一系列麻煩的步驟,代碼編寫不直觀.另外一種好的思路是利用多進程/多線程+Queue的方法,可以避免加鎖這樣麻煩低效的方式.

現(xiàn)在在Python2中利用Queue+多進程的方法來處理一個IO密集型任務.

假設現(xiàn)在需要下載多個網(wǎng)頁內(nèi)容并進行解析,單進程的方式效率很低,所以使用多進程/多線程勢在必行.

我們可以先初始化一個tasks隊列,里面將要存儲的是一系列dest_url,同時開啟4個進程向tasks中取任務然后執(zhí)行,處理結(jié)果存儲在一個results隊列中,最后對results中的結(jié)果進行解析.最后關閉兩個隊列.

下面是一些主要的邏輯代碼.

# -*- coding:utf-8 -*-

#IO密集型任務

#多個進程同時下載多個網(wǎng)頁

#利用Queue+多進程

#由于是IO密集型,所以同樣可以利用threading模塊

import multiprocessing

def main():

tasks = multiprocessing.JoinableQueue()

results = multiprocessing.Queue()

cpu_count = multiprocessing.cpu_count() #進程數(shù)目==CPU核數(shù)目

create_process(tasks, results, cpu_count) #主進程馬上創(chuàng)建一系列進程,但是由于阻塞隊列tasks開始為空,副進程全部被阻塞

add_tasks(tasks) #開始往tasks中添加任務

parse(tasks, results) #最后主進程等待其他線程處理完成結(jié)果

def create_process(tasks, results, cpu_count):

for _ in range(cpu_count):

p = multiprocessing.Process(target=_worker, args=(tasks, results)) #根據(jù)_worker創(chuàng)建對應的進程

p.daemon = True #讓所有進程可以隨主進程結(jié)束而結(jié)束

p.start() #啟動

def _worker(tasks, results):

while True: #因為前面所有線程都設置了daemon=True,故不會無限循環(huán)

try:

task = tasks.get() #如果tasks中沒有任務,則阻塞

result = _download(task)

results.put(result) #some exceptions do not handled

finally:

tasks.task_done()

def add_tasks(tasks):

for url in get_urls(): #get_urls() return a urls_list

tasks.put(url)

def parse(tasks, results):

try:

tasks.join()

except KeyboardInterrupt as err:

print "Tasks has been stopped!"

print err

while not results.empty():

_parse(results)

if __name__ == '__main__':

main()

利用Python3中的concurrent.futures包

在Python3中可以利用concurrent.futures包,編寫更加簡單易用的多線程/多進程代碼.其使用感覺和Java的concurrent框架很相似(借鑒?)

比如下面的簡單代碼示例

def handler():

futures = set()

with concurrent.futures.ProcessPoolExecutor(max_workers=cpu_count) as executor:

for task in get_task(tasks):

future = executor.submit(task)

futures.add(future)

def wait_for(futures):

try:

for future in concurrent.futures.as_completed(futures):

err = futures.exception()

if not err:

result = future.result()

else:

raise err

except KeyboardInterrupt as e:

for future in futures:

future.cancel()

print "Task has been canceled!"

print e

return result

總結(jié)

要是一些大型Python項目也這般編寫,那么效率也太低了.在Python中有許多已有的框架使用,使用它們起來更加高效.

但是自己的一些"小打小鬧"的程序這樣來編寫還是不錯的.:)

python簡單的并發(fā)問題

#!/usr/bin/envpython#-*-coding:utf-8-*-#author:ChanghuaGongimporttime,threading#fromurllib.requestimportRequest,urlopenpy3#fromurllib.errorimportURLErrorpy3importurllib2#URLreq=urllib2.Request('

#!/usr/bin/env python

# -*- coding:utf-8 -*-

# author: Changhua Gong

import time,threading

# from urllib.request import Request, urlopen py3

# from urllib.error import URLError py3

import urllib2

#URL

req = urllib2.Request('')

#

rule = {0:500,1:30}

'''

Rule規(guī)則:0:50,第一次運行不睡眠即為0,直接并發(fā)50次;1:20,第二秒,相當于睡眠1秒,然后并發(fā)20次,

如第三秒需并發(fā)500次,則rule = {0:50,1:20,1:500}

'''

#Open url

def geturl():

time_b = time.time()

try:

response = urllib2.urlopen(req)

print(response.read().decode("utf-8")) # 打印輸出內(nèi)容

except urllib2.URLError as e:

if hasattr(e, 'reason'):

print('We failed to reach a server.')

print('Reason: ', e.reason)

elif hasattr(e, 'code'):

print('The server couldn/'t fulfill the request.')

print('Error code: ', e.code)

time_e = time.time()

print("Thread %s runned for %ss" % (threading.current_thread().name, (time_e - time_b))) #線程訪問時效

if __name__=='__main__':

for k in rule:

time.sleep(k)

for i in range(rule[k]):

t = threading.Thread(target=geturl)

t.start()

python并發(fā)編程-Gevent包介紹

Gevent 是一個第三方庫，可以輕松通過gevent實現(xiàn)并發(fā)同步或異步編程，在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協(xié)程。 Greenlet全部運行在主程序操作系統(tǒng)進程的內(nèi)部，但它們被協(xié)作式地調(diào)度。

遇到IO阻塞時會自動切換任務

上例gevent.sleep(2)模擬的是gevent可以識別的io阻塞,

而time.sleep(2)或其他的阻塞,gevent是不能直接識別的需要用下面一行代碼,打補丁,就可以識別了

from gevent import monkey;monkey.patch_all()必須放到被打補丁者的前面，如time，socket模塊之前

我們可以用threading.current_thread().getName()來查看每個g1和g2，查看的結(jié)果為DummyThread-n，即假線程

通過gevent實現(xiàn)單線程下的socket并發(fā)（from gevent import monkey;monkey.patch_all()一定要放到導入socket模塊之前，否則gevent無法識別socket的阻塞）

服務端

客戶端

多線程并發(fā)多個客戶端

python并發(fā)編程-進程池

在利用Python進行系統(tǒng)管理的時候，特別是同時操作多個文件目錄，或者遠程控制多臺主機，并行操作可以節(jié)約大量的時間。多進程是實現(xiàn)并發(fā)的手段之一，需要注意的問題是：

例如當被操作對象數(shù)目不大時，可以直接利用multiprocessing中的Process動態(tài)成生多個進程，十幾個還好，但如果是上百個，上千個。。。手動的去限制進程數(shù)量卻又太過繁瑣，此時可以發(fā)揮進程池的功效。

我們就可以通過維護一個進程池來控制進程數(shù)目，比如httpd的進程模式，規(guī)定最小進程數(shù)和最大進程數(shù)..

ps： 對于遠程過程調(diào)用的高級應用程序而言，應該使用進程池，Pool可以提供指定數(shù)量的進程，供用戶調(diào)用，當有新的請求提交到pool中時，如果池還沒有滿，那么就會創(chuàng)建一個新的進程用來執(zhí)行該請求；但如果池中的進程數(shù)已經(jīng)達到規(guī)定最大值，那么該請求就會等待，直到池中有進程結(jié)束，就重用進程池中的進程。

創(chuàng)建進程池的類：如果指定numprocess為3，則進程池會從無到有創(chuàng)建三個進程，然后自始至終使用這三個進程去執(zhí)行所有任務，不會開啟其他進程

參數(shù)介紹：

方法介紹：

主要方法：

其他方法（了解部分）

應用：

發(fā)現(xiàn)：并發(fā)開啟多個客戶端，服務端同一時間只有3個不同的pid，干掉一個客戶端，另外一個客戶端才會進來，被3個進程之一處理

回調(diào)函數(shù)：

需要回調(diào)函數(shù)的場景：進程池中任何一個任務一旦處理完了，就立即告知主進程：我好了額，你可以處理我的結(jié)果了。主進程則調(diào)用一個函數(shù)去處理該結(jié)果，該函數(shù)即回調(diào)函數(shù)

我們可以把耗時間（阻塞）的任務放到進程池中，然后指定回調(diào)函數(shù)（主進程負責執(zhí)行），這樣主進程在執(zhí)行回調(diào)函數(shù)時就省去了I/O的過程，直接拿到的是任務的結(jié)果。

如果在主進程中等待進程池中所有任務都執(zhí)行完畢后，再統(tǒng)一處理結(jié)果，則無需回調(diào)函數(shù)

Python中的并行和并發(fā)是什么

并行和并發(fā)

無論是并行還是并發(fā)，在用戶看來都是'同時'運行的，不管是進程還是線程，都只是一個任務而已，真是干活的是cpu，cpu來做這些任務，而一個cpu同一時刻只能執(zhí)行一個任務。

并發(fā)是偽并行，即看起來是同時運行。單個cpu+多道技術就可以實現(xiàn)并發(fā)，（并行也屬于并發(fā)），簡單的可以理解為快速在多個線程來回切換，感覺好像同時在做多個事情。

只有具備多個cpu才能實現(xiàn)并行，單核下，可以利用多道技術，多個核，每個核也都可以利用多道技術（多道技術是針對單核而言的）。? 有四個核，六個任務，這樣同一時間有四個任務被執(zhí)行，假設分別被分配給了cpu1，cpu2，cpu3，cpu4，一旦任務1遇到I/O就被迫中斷執(zhí)行，此時任務5就拿到cpu1的時間片去執(zhí)行，這就是單核下的多道技術 ，而一旦任務1的I/O結(jié)束了，操作系統(tǒng)會重新調(diào)用它(需知進程的調(diào)度、分配給哪個cpu運行，由操作系統(tǒng)說了算)，可能被分配給四個cpu中的任意一個去執(zhí)行。

相關推薦：《Python視頻教程》

多道技術：內(nèi)存中同時存入多道（多個）程序，cpu從一個進程快速切換到另外一個，使每個進程各自運行幾十或幾百毫秒，這樣，雖然在某一個瞬間，一個cpu只能執(zhí)行一個任務，但在1秒內(nèi)，cpu卻可以運行多個進程，這就給人產(chǎn)生了并行的錯覺，即偽并發(fā)，以此來區(qū)分多處理器操作系統(tǒng)的真正硬件并行（多個cpu共享同一個物理內(nèi)存）。

同步執(zhí)行：一個進程在執(zhí)行某個任務時，另外一個進程必須等待其執(zhí)行完畢，才能繼續(xù)執(zhí)行。

異步執(zhí)行：一個進程在執(zhí)行某個任務時，另外一個進程無需等待其執(zhí)行完畢，就可以繼續(xù)執(zhí)行，當有消息返回時，系統(tǒng)會通知后者進行處理，這樣可以提高執(zhí)行效率。

舉個例子，打電話時就是同步通信，發(fā)短息時就是異步通信。

相關推薦：

Python如何實現(xiàn)線程間同步

Python中級精華-并發(fā)之啟動和停止線程

為了讓代碼能夠并發(fā)執(zhí)行，向創(chuàng)建線程并在核實的時候銷毀它。

由于目的比較單純，只是講解基礎的線程創(chuàng)建方法，所以可以直接使用threading庫中的Thread類來實例化一個線程對象。

例子，用戶輸入兩個數(shù)字，并且求其兩個數(shù)字的四則運算的結(jié)果：

除了以上的一些功能以外，在python線程

中沒有其他的諸如給線程發(fā)信號、設置線程調(diào)度屬性、執(zhí)行任何其他高級操作的功能了，如果需要這些功能，就需要手工編寫了。

另外，需要注意的是，由于GIL（全局解釋器鎖）的存在，限制了在python解釋器當中只允許運行一個線程。基于這個原因，不停該使用python線程來處理計算密集型的任務，因為在這種任務重我們希望在多個CPU核心上實現(xiàn)并行處理。Python線程更適合于IO處理以及設計阻塞操作的并發(fā)執(zhí)行任務（即等待IO響應或等待數(shù)據(jù)庫取出結(jié)果等）。

如何判斷線程是否已經(jīng)啟動？

目的：我們加載了一個線程，但是想要知道這個線程什么時候才會開始運行？

方法：

線程的核心特征我認為就是不確定性，因為其什么時候開始運行，什么時候被打斷，什么時候恢復執(zhí)行，這不是程序員能夠控制的，而是有系統(tǒng)調(diào)度

來完成的。如果遇到像某個線程的運行依托于其他某個線程運行到某個狀態(tài)時該線程才能開始運行，那么這就是線程同步

問題，同樣這個問題非常棘手。要解決這類問題我們要借助threading中的Event對象。

Event其實和條件標記類似，勻速線程

等待某個時間發(fā)生。初始狀態(tài)時事件被設置成0。如果事件沒有被設置而線程正在等待該事件，那么線程就會被阻塞，直到事件被設置位置，當有線程設置了這個事件之后，那么就會喚醒正在等待事件的線程，如果線程等待的事件已經(jīng)設置了，那么線程會繼續(xù)執(zhí)行。

一個例子：

如上能夠確定的是，主線程會在線程t運行結(jié)束時再運行。

新聞名稱：python函數(shù)并發(fā),python并發(fā)處理
網(wǎng)站網(wǎng)址：http://chinadenli.net/article34/dsicpse.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站建設、軟件開發(fā)、App開發(fā)、標簽優(yōu)化、營銷型網(wǎng)站建設、ChatGPT

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)