002---線程

線程

什麼是線程

線程： 線程是操做系統調度的最小單位，它包含在進程中。

比喻：一條流水線工做的流程，一條流水線必須屬於一個車間，一個車間的工做過程是一個進程，車間負責把資源整合到一塊兒，是一個資源單位，而一個車間內至少有一個流水線，流水線的工做須要電源，電源就至關於cpu

線程與進程的區別

建立一個進程，就是建立一個車間，涉及到申請空間，並且在該空間內建至少一條流水線，但建立線程，就只是在一個車間內造一條流水線，無需申請空間，因此建立開銷小。

• 同一個進程內的多個線程共享該進程內的地址資源
• 建立線程的開銷要遠小於建立進程的開銷
• 線程的開啓速度更快

開啓線程的兩種方式

• 普通方式

  from threading import Thread
  
  
  def func(*args, **kwargs):
      print(args, kwargs)
      print('子線程啓動啦')
  
  
  t = Thread(target=func, args=(1, 2, 3), kwargs={"name": 'jiangwei'})
  t.start()
  print('開啓了一個線進程')

• 類建立

  from threading import Thread
  
  
  class MyThread(Thread):
      def __init__(self, i):
          super(MyThread, self).__init__()
          self.i = i
  
      def run(self):
          time.sleep(1)
          print(self.i)
  
  
  t = MyThread(10)
  t.start()

線程相關方法和屬性

Thread對象的屬性和方法

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

threading模塊提供的一些方法：

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

守護線程

不管是進程仍是線程，都遵循：守護xxx會等待主xxx運行完畢以後被銷燬

注意：運行完畢並不是終止運行

• 對進程來講：指主進程代碼運行完畢
• 對線程來講：指主線程所在的進程內全部的非守護線程都運行完畢，主線程纔算運行完畢，由於主線程的結束表明主進程的結束，進程總體的資源都將被回收，而進程必須保證非守護線程都運行完畢才能結束。

def f1():
    print('f1', 666)
    time.sleep(3)
    print('你都沒死，我咋能先死，你先把其餘人殺了再說')
    time.sleep(2.5)
    print('我發現我已經死了')


def f2():
    print('f2', 999)
    time.sleep(5)
    print('哈哈哈，他把我殺了')


t1 = Thread(target=f1)
t2 = Thread(target=f2)
t1.daemon = True  # 守護線程t1
t1.start()
t2.start()
print('主')       # 主線程結束(等待其餘非守護進程結束，也就是f2運行完畢)、守護進程隨之結束(f1不會執行到print)   
"""
f1 666
f2 999
主
你都沒死，我咋能先死，你先把其餘人殺了再說
哈哈哈，他把我殺了
"""

GIL全局解釋器鎖

• 本質也是一個互斥鎖，不是Python的特性，是Cpython解釋器的特性
• 在同一個進程下開啓的多線程，同一時刻只能有一個線程能拿到那把鎖，只有一個線程而後去執行
• 不一樣的鎖保護不一樣的數據，自定義的互斥鎖，保護程序內的數據，而GIL鎖是解釋器級別的，與垃圾回收的數據有關
• 開啓多線程以後，先搶GIL,後搶自定義鎖

爲何要有這把鎖

解釋器的代碼是共享的，若是程序中有一個線程是修改全局變量n=100,而解釋器裏的垃圾回收線程執行回收n=100的操做，這就致使了數據混亂，不安全。

疑問

有了GIL的存在，同一時刻同一進程中只有一個線程被執行。那麼，進程能夠利用多核，但開銷大，而線程開銷小，卻沒法利用多核？

其實否則：
我開了一個工廠。假設個人原材料充足的狀況下，個人工人越多越好，效率快。相似於cpu作計算.可是，若是個人原料不充足，要從西伯利亞運過來，那麼個人工人越多越好嗎？無濟於事。相似於cpu作io讀寫操做。

結論

• 多線程用於IO密集型，如socket，爬蟲，web
• 多進程用於計算密集型，如金融分析，科學計算

死鎖與遞歸鎖

死鎖

兩個或兩個以上的進程或線程在執行過程當中，由於爭奪資源而形成的互相等待的現象。若無外力做用，他們將沒法推動下去，會進入死鎖狀態

遞歸鎖Rlock

能夠屢次acquire,內部維護一個Lock和counter變量。counter記錄acquire的次數，每加鎖一次就+1。直到該線程的全部鎖被釋放，其餘線程才能搶到

import time
from threading import Lock, Thread, RLock

# l1 = Lock()
# l2 = Lock()


# 遞歸鎖  能夠連續acquire屢次，每acquire一次，計數器+1  只要計數不爲0，就不能被其餘線程搶到
l1 = l2  = RLock()

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        l1.acquire()
        print('%s拿到了l1' % self.name)
        l2.acquire()
        print('%s拿到了l2' % self.name)
        l2.release()
        l1.release()

    def f2(self):
        l2.acquire()
        print('%s拿到了l1' % self.name)
        time.sleep(0.1)
        l1.acquire()
        print('%s拿到了l2' % self.name)
        l1.release()
        l2.release()


if __name__ == '__main__':
    for i in range(10):
        t = MyThread()
        t.start()

信號量

也是一把鎖，放多個鑰匙。只不過再也不是家裏的衛生間，而是公共廁所，每次能進入多人。

import time, random
from threading import Thread, Semaphore, currentThread

sm = Semaphore(5)


def task():
    with sm:
        print('%s ing ' % currentThread().getName())
        time.sleep(random.randint(1, 3))


if __name__ == '__main__':
    for i in range(10):
        t = Thread(target=task)
        t.start()

事件

主要是根據狀態來控制線程

經常使用方法

• e.isSet()/e.is_set()：返回event的狀態值；
• e.wait()：若是 event.isSet()==False將阻塞線程；
• e.set()： 設置event的狀態值爲True，全部阻塞池的線程激活進入就緒狀態，等待操做系統調度；
• e.clear()：恢復event的狀態值爲False；

import time
from threading import Thread, Event

event = Event()


def student(name):
    print('%s 正在聽課' % name)
    event.wait()
    # event.wait(2)  # 時間
    print('%s 課間活動' % name)


def teacher(name):
    print('%s 正在講課' % name)
    time.sleep(7)

    event.set()  # 改變狀態爲True，默認false event.clear()設爲false


if __name__ == '__main__':
    t = Thread(target=teacher, args=('jw',))
    s1 = Thread(target=student, args=('alex',))
    s2 = Thread(target=student, args=('wupeiqi',))
    s3 = Thread(target=student, args=('egon',))
    t.start()
    s1.start()
    s2.start()
    s3.start()

定時器

from threading import Timer

def task(name):
    print('hello %s'%name)

t = Timer(3,task,args=('egon',))
t.start()

線程隊列

import queue

q = queue.Queue() # 先進先出--隊列

q1 = queue.LifoQueue()  # 後進先出--堆棧

q2 = queue.PriorityQueue()  # 優先級隊列

q2.put('a')
q2.put('c')
q2.put('a')

print(q2.get())
print(q2.get())
print(q2.get())

線程池

基本方法

• t.submit(func,*args,**kwargs)：異步提交任務
• t.shutdown(wait=True)：pool.close() + pool.join()
• t.result()：拿結果
• t.add_done_callback(func)：添加回調函數

使用

import time, os, random

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


def task(name):
    print('name: %s pid:%s run' % (name, os.getpid()))
    time.sleep(random.randint(1, 3))
    return '我是回調函數'


def call_back(m):
    print(m.result())


if __name__ == '__main__':
    pool = ThreadPoolExecutor(5)
    t_list = []
    for i in range(10):
        t = pool.submit(task, 'alex%s' % i)
        t.add_done_callback(call_back)
        t_list.append(t)
    pool.shutdown(wait=True)  # close + join
    print('主')
    for t in t_list:
        print('----', t.result())