python中多線程（1）

一多線程的概念介紹

threading模塊介紹

threading模塊和multiprocessing模塊在使用層面，有很大的類似性。

2、開啓多線程的兩種方式

1.建立線程的開銷比建立進程的開銷小，於是建立線程的速度快
 2 from multiprocessing import Process
 3 from threading import Thread
 4 import os
 5 import time
 6 def work():
 7     print('<%s> is running'%os.getpid())
 8     time.sleep(2)
 9     print('<%s> is done'%os.getpid())
10 
11 if __name__ == '__main__':
12     t=Thread(target=work,)
13     # t= Process(target=work,)
14     t.start()
15     print('主',os.getpid())

from threading import Thread
 2 import time
 3 class Work(Thread):
 4     def __init__(self,name):
 5         super().__init__()
 6         self.name = name
 7     def run(self):
 8         # time.sleep(2)
 9         print('%s say hell'%self.name)
10 if __name__ == '__main__':
11     t = Work('egon')
12     t.start()
13     print('主')

在一個進程下開啓多個線程與在一個進程下開啓多個子進程的區別

from  multiprocessing import Process
 2 from threading import Thread
 3 import time
 4 def work():
 5     time.sleep(2)
 6     print('hello')
 7 if __name__ == '__main__':
 8     t = Thread(target=work)#若是等上幾秒，他會在開啓的過程當中先打印主，若是不等會先打印hello
 9     # t = Process(target=work) #子進程會先打印主，
10     t.start()
11     print('主')

# 2.----------
 2 from  multiprocessing import Process
 3 from threading import Thread
 4 import os
 5 def work():
 6     print('hello',os.getpid())
 7 if __name__ == '__main__':
 8     #在主進程下開啓多個線程，每一個線程都跟主進程的pid同樣
 9     t1= Thread(target=work)
10     t2 = Thread(target=work)
11     t1.start()
12     t2.start()
13     print('主線程pid',os.getpid())
14 
15     #來多個進程，每一個進程都有不一樣的pid
16     p1 = Process(target=work)
17     p2 = Process(target=work)
18     p1.start()
19     p2.start()
20     print('主進程pid', os.getpid())

 

from  threading import Thread
 2 from multiprocessing import  Process
 3 import os
 4 def work():
 5     global n
 6     n-=1
 7     print(n)  #因此被改爲99了
 8 n = 100
 9 if __name__ == '__main__':
10     # p = Process(target=work)
11     p = Thread(target=work)  #當開啓的是線程的時候，由於同一進程內的線程之間共享進程內的數據
12                             #因此打印的n爲99
13     p.start()
14     p.join()
15     print('主',n) #毫無疑問子進程p已經將本身的全局的n改爲了0,
16     # 但改的僅僅是它本身的,查看父進程的n仍然爲100

進程之間是互相隔離的，不共享。須要藉助第三方來完成共享（藉助隊列，管道，共享數據）

3、練習

練習一：多線程實現併發

from socket import *
 2 from threading import Thread
 3 s = socket(AF_INET,SOCK_STREAM)
 4 s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) #端口重用
 5 s.bind(('127.0.0.1',8081))
 6 s.listen(5)
 7 print('start running...')
 8 def talk(coon,addr):
 9     while True:  # 通訊循環
10         try:
11             cmd = coon.recv(1024)
12             print(cmd.decode('utf-8'))
13             if not cmd: break
14             coon.send(cmd.upper())
15             print('發送的是%s'%cmd.upper().decode('utf-8'))
16         except Exception:
17             break
18     coon.close()
19 if __name__ == '__main__':
20     while True:#連接循環
21         coon,addr = s.accept()
22         print(coon,addr)
23         p =Thread(target=talk,args=(coon,addr))
24         p.start()
25     s.close()

from socket import *
 2 c = socket(AF_INET,SOCK_STREAM)
 3 c.connect(('127.0.0.1',8081))
 4 while True:
 5     cmd = input('>>:').strip()
 6     if not cmd:continue
 7     c.send(cmd.encode('utf-8'))
 8     data = c.recv(1024)
 9     print('接受的是%s'%data.decode('utf-8'))
10 c.close()

 

練習二：三個任務，一個接收用戶輸入，一個將用戶輸入的內容格式化成大寫，一個將格式化後的結果存入文件

from threading import Thread
 2 import os
 3 input_l = []
 4 format_l = []
 5 def talk(): #監聽輸入任務
 6     while True:
 7         cmd = input('>>:').strip()
 8         if not cmd:continue
 9         input_l.append(cmd)
10 
11 def format():
12      while True:
13          if input_l:
14              res = input_l.pop()#取出來
15              format_l.append(res.upper()) #取出來後變大寫
16 def save():
17     while True:
18         if format_l:  #若是format_l不爲空
19             with open('db','a') as f:
20                 f.write(format_l.pop()+'\n') #寫進文件
21                 f.flush()
22 if __name__ == '__main__':
23     t1=Thread(target=talk)
24     t2=Thread(target=format)
25     t3=Thread(target=save)
26     t1.start()
27     t2.start()
28     t3.start()

4、多線程共享同一個進程內的地址空間 

from threading import Thread
 2 from multiprocessing import Process
 3 import os
 4 n = 100
 5 def talk():
 6     global n
 7     n-=100
 8     print(n)
 9 if __name__ == '__main__':
10     t = Thread(target=talk) #若是開啓的是線程的話，n=0
11     # t = Process(target=talk) #若是開啓的是進程的話，n=100
12     t.start()
13     t.join()
14     print('主',n)

5、線程對象的其餘屬性和方法

# isAlive(): 返回線程是否活動的。
  # getName(): 返回線程名。
  # setName(): 設置線程名。

threading模塊提供的一些方法：
  # threading.currentThread(): 返回當前的線程變量。
  # threading.enumerate(): 返回一個包含正在運行的線程的list。正在運行指線程啓動後、結束前，不包括啓動前和終止後的線程。
  # threading.activeCount(): 返回正在運行的線程數量，與len(threading.enumerate())有相同的結果。

 from  threading import Thread
 2 from multiprocessing import Process
 3 import time,os,threading
 4 def work():
 5     time.sleep(2)
 6     print('%s is running' % threading.currentThread().getName())
 7     print(threading.current_thread()) #其餘線程
 8     print(threading.currentThread().getName()) #獲得其餘線程的名字
 9 if __name__ == '__main__':
10     t = Thread(target=work)
11     t.start()
12 
13     print(threading.current_thread().getName())  #主線程的名字
14     print(threading.current_thread()) #主線程
15     print(threading.enumerate()) #連同主線程在內有兩個運行的線程
16     time.sleep(2)
17     print(t.is_alive()) #判斷線程是否存活
18     print(threading.activeCount())
19     print('主')

 

6、join與守護線程

主進程等全部的非守護的子進程結束他才結束（回收它子進程的資源）：（有父子關係）
主線程等非守護線程全都結束它才結束： （沒父子關係）

from  threading import Thread
 2 import time,os
 3 def talk():
 4     time.sleep(3)
 5     print('%s is running..'%os.getpid())
 6 if __name__ == '__main__':
 7     t = Thread(target=talk)
 8     t.start()
 9     t.join()  #主進程在等子進程結束
10     print('主')

守護線程與守護進程的區別

1.守護進程：主進程會等到全部的非守護進程結束，才銷燬守護進程。也就是說（主進程運行完了被守護的那個就幹掉了）

2.守護線程：主線程運行完了守護的那個尚未乾掉，主線程等非守護線程全都結束它才結束

from  multiprocessing import Process
 2 from threading import Thread,currentThread
 3 import time,os
 4 def talk1():
 5     time.sleep(2)
 6     print('hello')
 7 def talk2():
 8     time.sleep(2)
 9     print('you see see')
10 if __name__ == '__main__':
11     t1 = Thread(target=talk1)
12     t2 = Thread(target=talk2)
13     # t1 = Process(target=talk1)
14     # t2 = Process(target=talk2)
15     t1.daemon = True
16     t1.start()
17     t2.start()
18     print('主線程',os.getpid())

 

 #3 --------迷惑人的例子
 2 from threading import Thread
 3 import time
 4 def foo():
 5     print(123)
 6     # time.sleep(10) #若是這個等的時間大於下面等的時間，就把不打印end123了
 7     time.sleep(2)  #若是這個等的時間小於下面等的時間，就把end123也打印了
 8     print('end123')
 9 def bar():
10     print(456)
11     # time.sleep(5)
12     time.sleep(10)
13     print('end456')
14 if __name__ == '__main__':
15     t1 = Thread(target=foo)
16     t2 = Thread(target=bar)
17     t1.daemon = True #主線程運行完了守護的那個尚未乾掉，
18     # 主線程等非守護線程全都結束它才結束
19     t1.start()
20     t2.start()
21     print('main---------')

7、GIL與Lock

1.python GIL（Global Interpreter Lock) #全局的解釋器鎖

2.鎖的目的：犧牲了效率，保證了數據的安全
3.保護不一樣的數據加不一樣的鎖（）
4.python自帶垃圾回收

5.誰拿到GIL鎖就讓誰獲得Cpython解釋器的執行權限

6.GIT鎖保護的是Cpython解釋器數據的安全，而不會保護你本身程序的數據的安全
7.GIL鎖當遇到阻塞的時候，就被迫的吧鎖給釋放了，那麼其餘的就開始搶鎖了，搶到
後吧值修改了，可是第一個拿到的還在本來拿到的那個數據的那停留着呢，當再次拿
到鎖的時候，數據已經修改了，而你還拿的原來的，這樣就混亂了，因此也就保證不了
數據的安全了。
8.那麼怎麼解決數據的安全ne ?
本身再給加吧鎖：mutex=Lock()

 同步鎖

GIL 與Lock是兩把鎖，保護的數據不同，前者是解釋器級別的（固然保護的就是解釋器級別的數據，好比垃圾回收的數據），後者是保護用戶本身開發的應用程序的數據，很明顯GIL不負責這件事，只能用戶自定義加鎖處理，即Lock

 

過程分析：全部線程搶的是GIL鎖，或者說全部線程搶的是執行權限

 

　　線程1搶到GIL鎖，拿到執行權限，開始執行，而後加了一把Lock，尚未執行完畢，即線程1還未釋放Lock，有可能線程2搶到GIL鎖，開始執行，執行過程當中發現Lock尚未被線程1釋放，因而線程2進入阻塞，被奪走執行權限，有可能線程1拿到GIL，而後正常執行到釋放Lock。。。這就致使了串行運行的效果

　　既然是串行，那咱們執行

　　t1.start()

　　t1.join

　　t2.start()

　　t2.join()

　　這也是串行執行啊，爲什麼還要加Lock呢，需知join是等待t1全部的代碼執行完，至關於鎖住了t1的全部代碼，而Lock只是鎖住一部分操做共享數據的代碼。

 

 由於Python解釋器幫你自動按期進行內存回收，你能夠理解爲python解釋器裏有一個獨立的線程，每過一段時間它起wake up作一次全局輪詢看看哪些內存數據是能夠被清空的，此時你本身的程序 裏的線程和 py解釋器本身的線程是併發運行的，假設你的線程刪除了一個變量，py解釋器的垃圾回收線程在清空這個變量的過程當中的clearing時刻，可能一個其它線程正好又從新給這個還沒來及得清空的內存空間賦值了，結果就有可能新賦值的數據被刪除了，爲了解決相似的問題，python解釋器簡單粗暴的加了鎖，即當一個線程運行時，其它人都不能動，這樣就解決了上述的問題， 這能夠說是Python早期版本的遺留問題。　 

from threading import Thread,Lock
 2 import time
 3 n=100
 4 def work():
 5     mutex.acquire()
 6     global n
 7     temp=n
 8     time.sleep(0.01)
 9     n=temp-1
10     mutex.release()
11 if __name__ == '__main__':
12     mutex=Lock()
13     t_l=[]
14     s=time.time()
15     for i in range(100):
16         t=Thread(target=work)
17         t_l.append(t)
18         t.start()
19     for t in t_l:
20         t.join()
21     print('%s:%s' %(time.time()-s,n))

鎖一般被用來實現對共享資源的同步訪問。爲每個共享資源建立一個Lock對象，當你須要訪問該資源時，調用acquire方法來獲取鎖對象（若是其它線程已經得到了該鎖，則當前線程需等待其被釋放），待資源訪問完後，再調用release方法釋放鎖：

import threading
2 mutex = threading.Lock()
3 mutex.aquire()
4 '''
5 對公共數據的操做
6 '''
7 mutex.release()

分析：
2 　　1.100個線程去搶GIL鎖，即搶執行權限
3      2. 確定有一個線程先搶到GIL（暫且稱爲線程1），而後開始執行，一旦執行就會拿到lock.acquire()
4      3. 極有可能線程1還未運行完畢，就有另一個線程2搶到GIL，而後開始運行，但線程2發現互斥鎖lock還未被線程1釋放，因而阻塞，被迫交出執行權限，即釋放GIL
5     4.直到線程1從新搶到GIL，開始從上次暫停的位置繼續執行，直到正常釋放互斥鎖lock，而後其餘的線程再重複2 3 4的過程

若是不加鎖：併發執行，速度快，數據不安全。

加鎖：串行執行，速度慢，數據安全。

#不加鎖:併發執行,速度快,數據不安全
  2 from threading import current_thread,Thread,Lock
  3 import os,time
  4 def task():
  5     global n
  6     print('%s is running' %current_thread().getName())
  7     temp=n
  8     time.sleep(0.5)
  9     n=temp-1
 10 
 11 
 12 if __name__ == '__main__':
 13     n=100
 14     lock=Lock()
 15     threads=[]
 16     start_time=time.time()
 17     for i in range(100):
 18         t=Thread(target=task)
 19         threads.append(t)
 20         t.start()
 21     for t in threads:
 22         t.join()
 23 
 24     stop_time=time.time()
 25     print('主:%s n:%s' %(stop_time-start_time,n))
 26 
 27 '''
 28 Thread-1 is running
 29 Thread-2 is running
 30 ......
 31 Thread-100 is running
 32 主:0.5216062068939209 n:99
 33 '''
 34 
 35 
 36 #不加鎖:未加鎖部分併發執行,加鎖部分串行執行,速度慢,數據安全
 37 from threading import current_thread,Thread,Lock
 38 import os,time
 39 def task():
 40     #未加鎖的代碼併發運行
 41     time.sleep(3)
 42     print('%s start to run' %current_thread().getName())
 43     global n
 44     #加鎖的代碼串行運行
 45     lock.acquire()
 46     temp=n
 47     time.sleep(0.5)
 48     n=temp-1
 49     lock.release()
 50 
 51 if __name__ == '__main__':
 52     n=100
 53     lock=Lock()
 54     threads=[]
 55     start_time=time.time()
 56     for i in range(100):
 57         t=Thread(target=task)
 58         threads.append(t)
 59         t.start()
 60     for t in threads:
 61         t.join()
 62     stop_time=time.time()
 63     print('主:%s n:%s' %(stop_time-start_time,n))
 64 
 65 '''
 66 Thread-1 is running
 67 Thread-2 is running
 68 ......
 69 Thread-100 is running
 70 主:53.294203758239746 n:0
 71 '''
 72 
 73 #有的同窗可能有疑問:既然加鎖會讓運行變成串行,那麼我在start以後當即使用join,就不用加鎖了啊,也是串行的效果啊
 74 #沒錯:在start以後馬上使用jion,確定會將100個任務的執行變成串行,毫無疑問,最終n的結果也確定是0,是安全的,但問題是
 75 #start後當即join:任務內的全部代碼都是串行執行的,而加鎖,只是加鎖的部分即修改共享數據的部分是串行的
 76 #單從保證數據安全方面,兩者均可以實現,但很明顯是加鎖的效率更高.
 77 from threading import current_thread,Thread,Lock
 78 import os,time
 79 def task():
 80     time.sleep(3)
 81     print('%s start to run' %current_thread().getName())
 82     global n
 83     temp=n
 84     time.sleep(0.5)
 85     n=temp-1
 86 
 87 
 88 if __name__ == '__main__':
 89     n=100
 90     lock=Lock()
 91     start_time=time.time()
 92     for i in range(100):
 93         t=Thread(target=task)
 94         t.start()
 95         t.join()
 96     stop_time=time.time()
 97     print('主:%s n:%s' %(stop_time-start_time,n))
 98 
 99 '''
100 Thread-1 start to run
101 Thread-2 start to run
102 ......
103 Thread-100 start to run
104 主:350.6937336921692 n:0 #耗時是多麼的恐怖
105 '''