最近學(xué)習(xí)spark，我主要使用pyspark api進(jìn)行編程。

之前使用Python都是現(xiàn)學(xué)現(xiàn)用，用完就忘了也沒有理解和記憶，因此這里把Python相關(guān)的知識也彌補(bǔ)和記錄下來吧

多線程任務(wù)隊(duì)列在實(shí)際項(xiàng)目中非常有用，關(guān)鍵的地方要實(shí)現(xiàn)隊(duì)列的多線程同步問題，也即保證隊(duì)列的多線程安全

例如：可以開多個(gè)消費(fèi)者線程，每個(gè)線程上綁定一個(gè)隊(duì)列，這樣就實(shí)現(xiàn)了多個(gè)消費(fèi)者同時(shí)處理不同隊(duì)列上的任務(wù)

同時(shí)可以有多個(gè)生產(chǎn)者往隊(duì)列發(fā)送消息，實(shí)現(xiàn)異步消息處理

先復(fù)習(xí)下互斥量和條件變量的概念：

互斥量(mutex)從本質(zhì)上說是一把鎖，在訪問共享資源前對互斥量進(jìn)行加鎖，在訪問完成后釋放互斥量上的鎖。對互斥量進(jìn)行加鎖以后，任何其他試圖再次對互斥鎖加鎖的線程將會(huì)阻塞直到當(dāng)前線程釋放該互斥鎖。如果釋放互斥鎖時(shí)有多個(gè)線程阻塞，所有在該互斥鎖上的阻塞線程都會(huì)變成可運(yùn)行狀態(tài)，第一個(gè)變?yōu)檫\(yùn)行狀態(tài)的線程可以對互斥鎖加鎖，其他線程將會(huì)看到互斥鎖依然被鎖住，只能回去再次等待它重新變?yōu)榭捎谩?/p>

條件變量(cond)是在多線程程序中用來實(shí)現(xiàn)"等待--》喚醒"邏輯常用的方法。條件變量利用線程間共享的全局變量進(jìn)行同步的一種機(jī)制，主要包括兩個(gè)動(dòng)作：一個(gè)線程等待"條件變量的條件成立"而掛起；另一個(gè)線程使“條件成立”。為了防止競爭，條件變量的使用總是和一個(gè)互斥鎖結(jié)合在一起。線程在改變條件狀態(tài)前必須首先鎖住互斥量，函數(shù)pthread_cond_wait把自己放到等待條件的線程列表上，然后對互斥鎖解鎖(這兩個(gè)操作是原子操作)。在函數(shù)返回時(shí)，互斥量再次被鎖住

條件變量總是與互斥鎖一起使用的

Python的threading中定義了兩種鎖：threading.Lock和threading.RLock

兩者的不同在于后者是可重入鎖，也就是說在一個(gè)線程內(nèi)重復(fù)LOCK同一個(gè)鎖不會(huì)發(fā)生死鎖，這與POSIX中的PTHREAD_MUTEX_RECURSIVE也就是可遞歸鎖的概念是相同的, 互斥鎖的API有三個(gè)函數(shù)，分別執(zhí)行分配鎖，上鎖，解鎖操作。

Python的threading中的條件變量默認(rèn)綁定了一個(gè)RLock，也可以在初始化條件變量的時(shí)候傳進(jìn)去一個(gè)自己定義的鎖.

最后貼出我自己實(shí)現(xiàn)的簡單線程安全任務(wù)隊(duì)列

測試代碼