㈠ python澶氱嚎紼嬪拰澶氳繘紼嬬殑鍖哄埆鏈夊摢浜涳紵
Python涓鐨勫氱嚎紼嬪拰澶氳繘紼嬮兘鏄鐢ㄤ簬瀹炵幇騫跺彂緙栫▼鐨勬妧鏈錛屼絾鏄瀹冧滑涔嬮棿鏈変竴浜涘尯鍒銆傚氱嚎紼嬫槸鍦ㄤ竴涓榪涚▼涓榪愯屽皹鎷嶅氫釜綰跨▼錛岃屽氳繘紼嬫槸鍦ㄤ竴涓榪涚▼涓榪愯屽氬巺鍨嬩釜紼嬪簭銆傚氱嚎紼嬪叡浜鍚屼釜鍦板潃絀洪棿銆佹墦寮鐨勬枃浠朵互鍙婂叾浠栬祫婧愶紝鑰屽氳繘紼嬪叡浜鐗╃悊鎵鍏勭寽鍐呭瓨銆佺佺洏銆佹墦鍗版満浠ュ強鍏朵粬璧勬簮銆
鍦≒ython涓錛岀敱浜庡叏灞瑙ｉ噴鍣ㄩ攣錛圙IL錛夌殑瀛樺湪錛屽氱嚎紼嬪苟涓嶈兘鐪熸ｅ湴瀹炵幇騫惰岃＄畻銆傚洜姝わ紝濡傛灉闇瑕佽繘琛屽ぇ閲忕殑璁＄畻錛屽緩璁浣跨敤澶氳繘紼嬫潵瀹炵幇銆備絾鏄濡傛灉闇瑕佽繘琛孖/O瀵嗛泦鍨嬬殑浠誨姟錛屼緥濡傜綉緇滆鋒眰銆佹枃浠惰誨啓絳夛紝鍒欏緩璁浣跨敤澶氱嚎紼嬫潵瀹炵幇銆
㈡ python中多進程+協程的使用以及為什麼要用它
前面講了為什麼python里推薦用多進程而不是多線程，但是多進程也有其自己的限制：相比線程更加笨重、切換耗時更長，並且在python的多進程下，進程數量不推薦超過CPU核心數（一個進程只有一個GIL，所以一個進程只能跑滿一個CPU），因為一個進程佔用一個CPU時能充分利用機器的性能，但是進程多了就會出現頻繁的進程切換，反而得不償失。
不過特殊情況（特指IO密集型任務）下，多線程是比多進程好用的。
舉個例子：給你200W條url，需要你把每個url對應的頁面抓取保存起來，這種時候，單單使用多進程，效果肯定是很差的。為什麼呢？
例如每次請求的等待時間是2秒，那麼如下（忽略cpu計算時間）：
1、單進程+單線程：需要2秒*200W=400W秒==1111.11個小時==46.3天，這個速度明顯是不能接受的
2、單進程+多線程：例如我們在這個進程中開了10個多線程，比1中能夠提升10倍速度，也就是大約4.63天能夠完成200W條抓取，請注意，這里的實際執行是：線程1遇見了阻塞，CPU切換到線程2去執行，遇見阻塞又切換到線程3等等，10個線程都阻塞後，這個進程就阻塞了，而直到某個線程阻塞完成後，這個進程才能繼續執行，所以速度上提升大約能到10倍（這里忽略了線程切換帶來的開銷，實際上的提升應該是不能達到10倍的），但是需要考慮的是線程的切換也是有開銷的，所以不能無限的啟動多線程（開200W個線程肯定是不靠譜的）
3、多進程+多線程：這里就厲害了，一般來說也有很多人用這個方法，多進程下，每個進程都能佔一個cpu，而多線程從一定程度上繞過了阻塞的等待，所以比單進程下的多線程又更好使了，例如我們開10個進程，每個進程里開20W個線程，執行的速度理論上是比單進程開200W個線程快10倍以上的（為什麼是10倍以上而不是10倍，主要是cpu切換200W個線程的消耗肯定比切換20W個進程大得多，考慮到這部分開銷，所以是10倍以上）。
還有更好的方法嗎？答案是肯定的，它就是：
4、協程，使用它之前我們先講講what/why/how（它是什麼/為什麼用它/怎麼使用它）
what：
協程是一種用戶級的輕量級線程。協程擁有自己的寄存器上下文和棧。協程調度切換時，將寄存器上下文和棧保存到其他地方，在切回來的時候，恢復先前保存的寄存器上下文和棧。因此：
協程能保留上一次調用時的狀態（即所有局部狀態的一個特定組合），每次過程重入時，就相當於進入上一次調用的狀態，換種說法：進入上一次離開時所處邏輯流的位置。
在並發編程中，協程與線程類似，每個協程表示一個執行單元，有自己的本地數據，與其它協程共享全局數據和其它資源。
why：
目前主流語言基本上都選擇了多線程作為並發設施，與線程相關的概念是搶占式多任務（Preemptive multitasking），而與協程相關的是協作式多任務。
不管是進程還是線程，每次阻塞、切換都需要陷入系統調用(system call)，先讓CPU跑操作系統的調度程序，然後再由調度程序決定該跑哪一個進程(線程)。
而且由於搶占式調度執行順序無法確定的特點，使用線程時需要非常小心地處理同步問題，而協程完全不存在這個問題（事件驅動和非同步程序也有同樣的優點）。
因為協程是用戶自己來編寫調度邏輯的，對CPU來說，協程其實是單線程，所以CPU不用去考慮怎麼調度、切換上下文，這就省去了CPU的切換開銷，所以協程在一定程度上又好於多線程。
how:
python裡面怎麼使用協程？答案是使用gevent，使用方法：看這里
使用協程，可以不受線程開銷的限制，我嘗試過一次把20W條url放在單進程的協程里執行，完全沒問題。
所以最推薦的方法，是多進程+協程（可以看作是每個進程里都是單線程，而這個單線程是協程化的）
多進程+協程下，避開了CPU切換的開銷，又能把多個CPU充分利用起來，這種方式對於數據量較大的爬蟲還有文件讀寫之類的效率提升是巨大的。
小例子：
[python]view plain
#-*-coding=utf-8-*-
importrequests
importgevent
fromgeventimportmonkey;monkey.patch_all()
importsys
reload(sys)
sys.setdefaultencoding('utf8')
deffetch(url):
try:
s=requests.Session()
r=s.get(url,timeout=1)#在這里抓取頁面
exceptException,e:
printe
return''
defprocess_start(url_list):
tasks=[]
forurlinurl_list:
tasks.append(gevent.spawn(fetch,url))
gevent.joinall(tasks)#使用協程來執行
deftask_start(filepath,flag=100000):#每10W條url啟動一個進程
withopen(filepath,'r')asreader:#從給定的文件中讀取url
url=reader.readline().strip()
url_list=[]#這個list用於存放協程任務
i=0#計數器，記錄添加了多少個url到協程隊列
whileurl!='':
i+=1
url_list.append(url)#每次讀取出url，將url添加到隊列
ifi==flag:#一定數量的url就啟動一個進程並執行
p=Process(target=process_start,args=(url_list,))
p.start()
url_list=[]#重置url隊列
i=0#重置計數器
url=reader.readline().strip()
ifurl_listnot[]:#若退出循環後任務隊列里還有url剩餘
p=Process(target=process_start,args=(url_list,))#把剩餘的url全都放到最後這個進程來執行
p.start()
if__name__=='__main__':
task_start('./testData.txt')#讀取指定文件
細心的同學會發現：上面的例子中隱藏了一個問題：進程的數量會隨著url數量的增加而不斷增加，我們在這里不使用進程池multiprocessing.Pool來控制進程數量的原因是multiprocessing.Pool和gevent有沖突不能同時使用，但是有興趣的同學可以研究一下gevent.pool這個協程池。
㈢ python 多進程和多線程配合
由於python的多線程中存在PIL鎖，因此python的多線程不能利用多核，那麼，由於現在的計算機是多核的，就不能充分利用計算機的多核資源。但是python中的多進程是可以跑在不同的cpu上的。因此，嘗試了多進程+多線程的方式，來做一個任務。比如：從中科大的鏡像源中下載多個rpm包。
#!/usr/bin/pythonimport reimport commandsimport timeimport multiprocessingimport threadingdef download_image(url):
print '*****the %s rpm begin to download *******' % url
commands.getoutput('wget %s' % url)def get_rpm_url_list(url):
commands.getoutput('wget %s' % url)
rpm_info_str = open('index.html').read()
regu_mate = '(?<=<a href=")(.*?)(?=">)'
rpm_list = re.findall(regu_mate, rpm_info_str)
rpm_url_list = [url + rpm_name for rpm_name in rpm_list] print 'the count of rpm list is: ', len(rpm_url_list) return rpm_url_
def multi_thread(rpm_url_list):
threads = [] # url = 'https://mirrors.ustc.e.cn/centos/7/os/x86_64/Packages/'
# rpm_url_list = get_rpm_url_list(url)
for index in range(len(rpm_url_list)): print 'rpm_url is:', rpm_url_list[index]
one_thread = threading.Thread(target=download_image, args=(rpm_url_list[index],))
threads.append(one_thread)
thread_num = 5 # set threading pool, you have put 4 threads in it
while 1:
count = min(thread_num, len(threads)) print '**********count*********', count ###25,25,...6707%25
res = [] for index in range(count):
x = threads.pop()
res.append(x) for thread_index in res:
thread_index.start() for j in res:
j.join() if not threads:
def multi_process(rpm_url_list):
# process num at the same time is 4
process = []
rpm_url_group_0 = []
rpm_url_group_1 = []
rpm_url_group_2 = []
rpm_url_group_3 = [] for index in range(len(rpm_url_list)): if index % 4 == 0:
rpm_url_group_0.append(rpm_url_list[index]) elif index % 4 == 1:
rpm_url_group_1.append(rpm_url_list[index]) elif index % 4 == 2:
rpm_url_group_2.append(rpm_url_list[index]) elif index % 4 == 3:
rpm_url_group_3.append(rpm_url_list[index])
rpm_url_groups = [rpm_url_group_0, rpm_url_group_1, rpm_url_group_2, rpm_url_group_3] for each_rpm_group in rpm_url_groups:
each_process = multiprocessing.Process(target = multi_thread, args = (each_rpm_group,))
process.append(each_process) for one_process in process:
one_process.start() for one_process in process:
one_process.join()# for each_url in rpm_url_list:# print '*****the %s rpm begin to download *******' %each_url## commands.getoutput('wget %s' %each_url)
def main():
url = 'https://mirrors.ustc.e.cn/centos/7/os/x86_64/Packages/'
url_paas = 'http://mirrors.ustc.e.cn/centos/7.3.1611/paas/x86_64/openshift-origin/'
url_paas2 ='http://mirrors.ustc.e.cn/fedora/development/26/Server/x86_64/os/Packages/u/'
start_time = time.time()
rpm_list = get_rpm_url_list(url_paas) print multi_process(rpm_list) # print multi_thread(rpm_list)
#print multi_process()
# print multi_thread(rpm_list)
# for index in range(len(rpm_list)):
# print 'rpm_url is:', rpm_list[index]
end_time = time.time() print 'the download time is:', end_time - start_timeprint main()123456789101112131415161718
代碼的功能主要是這樣的：
main（）方法中調用get_rpm_url_list（base_url）方法，獲取要下載的每個rpm包的具體的url地址。其中base_url即中科大基礎的鏡像源的地址，比如：http://mirrors.ustc.e.cn/centos/7.3.1611/paas/x86_64/openshift-origin/，這個地址下有幾十個rpm包，get_rpm_url_list方法將每個rpm包的url地址拼出來並返回。
multi_process（rpm_url_list）啟動多進程方法，在該方法中，會調用多線程方法。該方法啟動4個多進程，將上面方法得到的rpm包的url地址進行分組，分成4組，然後每一個組中的rpm包再最後由不同的線程去執行。從而達到了多進程+多線程的配合使用。
代碼還有需要改進的地方，比如多進程啟動的進程個數和rpm包的url地址分組是硬編碼，這個還需要改進，畢竟，不同的機器，適合同時啟動的進程個數是不同的。
㈣ Python實現簡單多線程任務隊列
Python實現簡單多線程任務隊列
最近我在用梯度下降演算法繪制神經網路的數據時，遇到了一些演算法性能的問題。梯度下降演算法的代碼如下（偽代碼）：
defgradient_descent(): # the gradient descent code plotly.write(X, Y)
一般來說，當網路請求 plot.ly 繪圖時會阻塞等待返回，於是也會影響到其他的梯度下降函數的執行速度。
一種解決辦法是每調用一次 plotly.write 函數就開啟一個新的線程，但是這種方法感覺不是很好。 我不想用一個像 cerely（一種分布式任務隊列）一樣大而全的任務隊列框架，因為框架對於我的這點需求來說太重了，並且我的繪圖也並不需要 redis 來持久化數據。
那用什麼辦法解決呢？我在 python 中寫了一個很小的任務隊列，它可以在一個單獨的線程中調用 plotly.write函數。下面是程序代碼。
classTaskQueue(Queue.Queue):
首先我們繼承 Queue.Queue 類。從 Queue.Queue 類可以繼承 get 和 put 方法，以及隊列的行為。
def__init__(self, num_workers=1): Queue.Queue.__init__(self) self.num_workers=num_workers self.start_workers()
初始化的時候，我們可以不用考慮工作線程的數量。
defadd_task(self, task,*args,**kwargs): args=argsor() kwargs=kwargsor{} self.put((task, args, kwargs))
我們把 task, args, kwargs 以元組的形式存儲在隊列中。*args 可以傳遞數量不等的參數，**kwargs 可以傳遞命名參數。
defstart_workers(self): foriinrange(self.num_workers): t=Thread(target=self.worker) t.daemon=True t.start()
我們為每個 worker 創建一個線程，然後在後台刪除。
下面是 worker 函數的代碼：
defworker(self): whileTrue: tupl=self.get() item, args, kwargs=self.get() item(*args,**kwargs) self.task_done()
worker 函數獲取隊列頂端的任務，並根據輸入參數運行，除此之外，沒有其他的功能。下面是隊列的代碼：
我們可以通過下面的代碼測試：
defblokkah(*args,**kwargs): time.sleep(5) print「Blokkah mofo!」 q=TaskQueue(num_workers=5) foriteminrange(1): q.add_task(blokkah) q.join()# wait for all the tasks to finish. print「Alldone!」
Blokkah 是我們要做的任務名稱。隊列已經緩存在內存中，並且沒有執行很多任務。下面的步驟是把主隊列當做單獨的進程來運行，這樣主程序退出以及執行資料庫持久化時，隊列任務不會停止運行。但是這個例子很好地展示了如何從一個很簡單的小任務寫成像工作隊列這樣復雜的程序。
defgradient_descent(): # the gradient descent code queue.add_task(plotly.write, x=X, y=Y)
修改之後，我的梯度下降演算法工作效率似乎更高了。如果你很感興趣的話，可以參考下面的代碼。 classTaskQueue(Queue.Queue): def__init__(self, num_workers=1):Queue.Queue.__init__(self)self.num_workers=num_workersself.start_workers() defadd_task(self, task,*args,**kwargs):args=argsor()kwargs=kwargsor{}self.put((task, args, kwargs)) defstart_workers(self):foriinrange(self.num_workers):t=Thread(target=self.worker)t.daemon=Truet.start() defworker(self):whileTrue:tupl=self.get()item, args, kwargs=self.get()item(*args,**kwargs)self.task_done() deftests():defblokkah(*args,**kwargs):time.sleep(5)print"Blokkah mofo!" q=TaskQueue(num_workers=5) foriteminrange(10):q.add_task(blokkah) q.join()# block until all tasks are doneprint"All done!" if__name__=="__main__":tests()
㈤ python3.7多線程代碼不執行
使用多進程代替多線程
㈥ Python 多線程傳參問題
下面是從python threading模塊摘錄下來的
class threading.Thread(group=None, target=None,
name=None, args=(), kwargs={})
args is the argument tuple for the target invocation. Defaults to
().
kwargs is a dictionary of keyword arguments for the target
invocation. Defaults to {}.
簡單的說就是，線程函數的參數要麼是tuple類型要麼是dict類型，既然python文檔上已經規定好了，那麼就是不能變了，除非重大的版本升級，棄用現在這種方式，不過這是不可能的，再者可以查看threading.Thread類的初始化代碼
classThread(_Verbose):
def__init__(self,group=None,target=None,name=None,
args=(),kwargs=None,verbose=None):
assertgroupisNone,"groupargumentmustbeNonefornow"
_Verbose.__init__(self,verbose)
ifkwargsisNone:
kwargs={}
self.__target=target
self.__name=str(nameor_newname())
self.__args=args
self.__kwargs=kwargs
self.__daemonic=self._set_daemon()
self.__ident=None
self.__started=Event()
self.__stopped=False
self.__block=Condition(Lock())
self.__initialized=True
#sys.
#sys.exc_
self.__stderr=_sys.stderr
看到上面的args和kwargs的賦值情況嗎，你不能像**kwargs這樣來傳參的。
從上面來看，你要使用多線程你就必須遵守這個約定，線程函數參數要麼是tuple,要麼是字典，沒辦法改進。非要改進那就是使用全局變數，不用傳參，但是如果是這樣的話會涉及到線程安全的問題或者叫數據同步問題，如果不加鎖或者其他互斥量的話這個問題會很嚴重。所以還是遵守規定吧
㈦ 為什麼在Python里推薦使用多進程而不是多
最近在看Python的多線程，經常我們會聽到老手說：「Python下多線程是雞肋，推薦使用多進程！」，但是為什麼這么說呢？
要知其然，更要知其所以然。所以有了下面的深入研究：
首先強調背景：
1、GIL是什麼？
GIL的全稱是Global Interpreter Lock(全局解釋器鎖)，來源是python設計之初的考慮，為了數據安全所做的決定。
2、每個CPU在同一時間只能執行一個線程（在單核CPU下的多線程其實都只是並發，不是並行，並發和並行從宏觀上來講都是同時處理多路請求的概念。但並發和並行又有區別，並行是指兩個或者多個事件在同一時刻發生；而並發是指兩個或多個事件在同一時間間隔內發生。）
在Python多線程下，每個線程的執行方式：
1、獲取GIL
2、執行代碼直到sleep或者是python虛擬機將其掛起。
3、釋放GIL
可見，某個線程想要執行，必須先拿到GIL，我們可以把GIL看作是「通行證」，並且在一個python進程中，GIL只有一個。拿不到通行證的線程，就不允許進入CPU執行。
在Python2.x里，GIL的釋放邏輯是當前線程遇見IO操作或者ticks計數達到100（ticks可以看作是Python自身的一個計數器，專門做用於GIL，每次釋放後歸零，這個計數可以通過 sys.setcheckinterval 來調整），進行釋放。
而每次釋放GIL鎖，線程進行鎖競爭、切換線程，會消耗資源。並且由於GIL鎖存在，python里一個進程永遠只能同時執行一個線程(拿到GIL的線程才能執行)，這就是為什麼在多核CPU上，python的多線程效率並不高。
那麼是不是python的多線程就完全沒用了呢？
在這里我們進行分類討論：
1、CPU密集型代碼(各種循環處理、計數等等)，在這種情況下，由於計算工作多，ticks計數很快就會達到閾值，然後觸發GIL的釋放與再競爭（多個線程來回切換當然是需要消耗資源的），所以python下的多
㈧ 如何在Python中編寫並發程序
GIL
在Python中,由於歷史原因(GIL),使得Python中多線程的效果非常不理想.GIL使得任何時刻Python只能利用一個CPU核,並
且它的調度演算法簡單粗暴:多線程中,讓每個線程運行一段時間t,然後強行掛起該線程,繼而去運行其他線程,如此周而復始,直到所有線程結束.
這使得無法有效利用計算機系統中的"局部性",頻繁的線程切換也對緩存不是很友好,造成資源的浪費.
據說Python官方曾經實現了一個去除GIL的Python解釋器,但是其效果還不如有GIL的解釋器,遂放棄.後來Python官方推出了"利
用多進程替代多線程"的方案,在Python3中也有concurrent.futures這樣的包,讓我們的程序編寫可以做到"簡單和性能兼得".
多進程/多線程+Queue
一般來說,在Python中編寫並發程序的經驗是:計算密集型任務使用多進程,IO密集型任務使用多進程或者多線程.另外,因為涉及到資源共享,所
以需要同步鎖等一系列麻煩的步驟,代碼編寫不直觀.另外一種好的思路是利用多進程/多線程+Queue的方法,可以避免加鎖這樣麻煩低效的方式.
現在在Python2中利用Queue+多進程的方法來處理一個IO密集型任務.
假設現在需要下載多個網頁內容並進行解析,單進程的方式效率很低,所以使用多進程/多線程勢在必行.
我們可以先初始化一個tasks隊列,裡面將要存儲的是一系列dest_url,同時開啟4個進程向tasks中取任務然後執行,處理結果存儲在一個results隊列中,最後對results中的結果進行解析.最後關閉兩個隊列.
下面是一些主要的邏輯代碼.
# -*- coding:utf-8 -*-
#IO密集型任務
#多個進程同時下載多個網頁
#利用Queue+多進程
#由於是IO密集型,所以同樣可以利用threading模塊
import multiprocessing
def main():
tasks = multiprocessing.JoinableQueue()
results = multiprocessing.Queue()
cpu_count = multiprocessing.cpu_count() #進程數目==CPU核數目
create_process(tasks, results, cpu_count) #主進程馬上創建一系列進程,但是由於阻塞隊列tasks開始為空,副進程全部被阻塞
add_tasks(tasks) #開始往tasks中添加任務
parse(tasks, results) #最後主進程等待其他線程處理完成結果
def create_process(tasks, results, cpu_count):
for _ in range(cpu_count):
p = multiprocessing.Process(target=_worker, args=(tasks, results)) #根據_worker創建對應的進程
p.daemon = True #讓所有進程可以隨主進程結束而結束
p.start() #啟動
def _worker(tasks, results):
while True: #因為前面所有線程都設置了daemon=True,故不會無限循環
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
總結
要是一些大型Python項目也這般編寫,那麼效率也太低了.在Python中有許多已有的框架使用,使用它們起來更加高效.