day11 消息隊列、多線程、多進程、協程

Python 消息隊列、多線程、多進程、協程（二）

本章內容簡介

1. queue 消息隊列詳解：經常使用參數、隊列的四種類型及實例；

2. Python 多線程： 線程池、線程鎖

3. Python 多進程:  進程鎖、進程池、數據共享

4. 協程

 

一. queue 消息隊列

1）經常使用參數：

a = queue.Queue() #括號裏是能夠放幾個數據，2就是2個；多的就放不進去了

a．put(11)             # put放數據時，能夠設置是否阻塞，超時時間

a．put(22)   

a.get()                   # 取數據，按模式取值，默認（阻塞）；能夠設置參數（block=False），超時時間（timeout=2）2秒；

a.qsize()   　　　　　# 計數，列隊裏有幾個值；

a.task_done()　　　 # 標識列隊裏的此任務已經完成了，和join配合使用；

a.join() 　　　　　　 # 若是列隊沒執行完，程序不終止，執行完要告訴隊列，執行完了，才能終止操做。

2）隊列的四種模式：

# queue.Queue，先進先出隊列

# queue.LifoQueue，後進先出隊列

# queue.PriorityQueue，優先級隊列

# queue.deque，雙向對隊

實例：

import queue
# q = queue.Queue(2) 先進先出隊列
# q = queue.LifoQueue() 後進先出隊列
# q = queue.PriorityQueue() 優先級隊列
# q = queue.deque()   雙向隊列

q = queue.LifoQueue()
q.put(123)
q.put(456)
# 打印；456
print(q.get())

# 優先級最小的拿出來
# 若是優先級同樣，則是誰先放，就先取出誰
q = queue.PriorityQueue()
q.put((1,'alex1'))
q.put((1,'alex2'))
q.put((1,'alex3'))
q.put((3,'alex3'))
# (1, 'alex1')
print(q.get())


q = queue.deque()
q.append(123)
q.append(333)
q.appendleft(456)
# deque([456, 123, 333])
print(q)
# 打印：456
print(q[0])
q.pop()     # 從右邊刪除
# deque([456, 123])
print(q)
q.popleft() # 從左邊刪除

隊列四種模式

 

3）隊列（生產者消費者模型） 

1）生產者消費者模型的做用：

一、解決阻塞

二、就是解耦，修改生產者，不會影響消費者，反之亦然。

2）在生產環境，用生產者消費者模型，就能夠解決：

一、處理瞬時併發的請求問題。瞬時的鏈接數就不會佔滿。因此服務器就不會掛了。

二、客戶端提交一個請求，不用等待處理完畢，能夠在頁面上作別的事情。

2.1）若是不用隊列存數據，服務端經過多線程來處理數據：

用戶往隊列存數據，服務器從隊列裏取數據。

沒有隊列的話，就跟最大鏈接數有關係，每一個服務器就有最大鏈接數。

客戶端要獲取服務器放回，服務器要查、修改數據庫或修改文件，要2分鐘，那客戶端就要掛起連接2分鐘，2萬個鏈接一半都要掛起，服務器就崩潰了。

 

若是沒有隊列，第一個用戶發來請求，連上服務器，佔用鏈接，等待2分鐘。

第二我的來也要佔用2分鐘。

web服務器

若是要處理併發，有10萬併發，若是：一臺機器接收一個鏈接，須要10萬個機器，等待2分鐘就處理完了。

2.2）把請求放在隊列的好處

用戶發來請求，把請求放到隊列裏，可讓鏈接立刻斷開，不會阻塞，就不佔用服務器的鏈接數了。若是看到訂單處理了沒，就要打開另一個頁面，查看請求是否處理。

服務器查詢處理任務的時候，每一個才花2分鐘，服務器耗時是沒有減小的。

可是這樣作，客戶端就不會持續的佔用鏈接了。那瞬時的鏈接數就不會佔滿。因此服務器就不會掛了。

可是後臺要處理10萬個請求，也須要50臺服務器。並不會減小服務器數量。

這樣就能處理瞬時併發的請求問題。

服務器只是處理請求，是修改數據庫的值，不是告訴客戶端。而是客戶端再發來請求，查詢數據庫已經修改的內容。

提交訂單以後，把這個訂單扔給隊列，程序返回「正在處理」，就不等待了，而後斷開這個鏈接，你能夠在頁面裏作別的事情，不用一直等待訂單處理完。這樣就不影響服務器的最大鏈接數。在頁面幫你發起一個alax請求，url，不停的請求（多是定時器），個人訂單成功沒有，個人訂單成功沒有，若是訂單成功了，就自動返回頁面：訂單成功

若是不用隊列的話，一個請求就佔用一個服務器，等待的人特別多，等待鏈接的個數太多了。服務器就掛掉了。

隊列就沒有最大個數限制，把請求發給隊列了，而後http連接就斷開了，就不用等待了。

12306買票的時候，下次再來請求的時候，就會告訴你，前面排了幾我的。

3）python queue的特色：

python的queue是內存級別的。rabbitmq能夠把隊列發到別的服務器上處理。

因此python裏的queue不能持久化，可是rabbitmq能夠持久化。

queue.Queue()這樣寫，隊列就沒有最大個數限制。queue.Queue(5)就是說隊列裏最多能放5個值

4）生產者消費者代碼示例：

import time,random
import queue,threading
q = queue.Queue()


def Producer(name):
    count =0
    while True:
        time.sleep(random.randrange(3))
        if q.qsize()<3:         # 只要盤子裏小於3個包子，廚師就開始作包子
            q.put(count)
            print("Producer %s has produced %s baozi.." %(name,count))
            count += 1

def Consumer(name):
    count =0
    while True:
        time.sleep(random.randrange(4))
        if not q.empty():       # 只要盤子裏有包子，顧客就要吃。
            data = q.get()
            print(data)
            print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' % (name,data))
        else:           # 盤子裏沒有包子
            print("---no baozi anymore----")
        count+=1

p1 = threading.Thread(target=Producer,args=('A',))
c1 = threading.Thread(target=Consumer,args=('B',))
c2 = threading.Thread(target=Consumer,args=('C',))
p1.start()
c1.start()
c2.start()

'''
當你設計複雜程序的時候，就能夠用生產者消費者模型，來鬆耦合你的代碼,也能夠減小阻塞。
'''

實例代碼

 

二. Python 多線程

1）線程鎖

• Lock，RLock

Lock只能鎖一次，RLock能夠遞歸多層，Lock不支持多層鎖嵌套，咱們通常用RLOCK

import threading
import time
NUM = 10

def func(lock):
    global NUM
    # 上鎖
    lock.acquire()
    lock.acquire()
    NUM -= 1
    time.sleep(2)
    print(NUM)
    # 開鎖
    lock.release()
    lock.release()

# Lock = threading.Lock()   # 不支持嵌套鎖，通常不用
RLock = threading.RLock()   # 通常用RLock，支持嵌套鎖。

for i in range(10):
    t = threading.Thread(target=func,args=(RLock,))
    t.start()

'''
死鎖：
就是你也搶資源，我也搶資源，誰也搶不走就是死鎖。
若是是python，就是Lock，弄成嵌套鎖，不支持，則變成死鎖。
解決辦法：
用RLock，支持嵌套鎖
'''

• 信號量 BoundedSemaphore

若是用線程鎖，一次只容許一個進入，若是用信號量能夠容許同時多少個一塊兒進入。

每次5個線程同時執行，可能就會同時修改一個值。

import threading
import time
NUM = 10

def func(i,lock):
    global NUM
    # 上鎖
    lock.acquire()  # 總共30個 一次執行5個 25個，依次類推：20,15。。。
    NUM -= 1
    time.sleep(2)
    print('NUM:',str(NUM),'i:',i)
    # 開鎖
    lock.release()

# Lock = threading.Lock()   # 不支持嵌套鎖，通常不用
# RLock = threading.RLock()   # 通常用RLock，支持嵌套鎖。
lock = threading.BoundedSemaphore(5)   # 參數是每次執行幾個線程

for i in range(30):
    t = threading.Thread(target=func,args=(i,lock,))
    t.start()

 

• event紅綠燈

要麼所有阻塞（紅燈），要麼所有放開（綠燈）

import threading

def func(i,e):
    print(i)    # 10個線程併發打印：0-9 ，而後到wait的時候，就開始檢測是什麼燈
    e.wait()    # 檢測是什麼燈，若是是紅燈，停；綠燈，行
    print(i+100)

event = threading.Event()

for i in range(10):
    t = threading.Thread(target=func,args=(i,event))
    t.start()

event.clear()   # 設置成紅燈
inp = input('>>>')
if inp == "1":
    event.set() #設置成綠燈

• 線程鎖條件-condition1

Lock,RLock：線程鎖使用場景：

Lock,RLock是多個用戶同時修改一份數據，可能會出現髒數據，數據就會亂，就加互斥鎖，一次只能讓一我的修改數據，就能解決。

condition，event，BoundedSemaphore 使用場景：

若是寫了個爬蟲，在創建數據庫鏈接，線程就等着，什麼能數據庫能用了，就開通線程，再爬蟲。

event是kua一下，全走了。

notify維護一個隊列，傳幾個，就只能出去幾回。

import threading


def func(i, con):
    print(i)
    con.acquire()
    con.wait()
    print(i + 100)
    con.release()


c = threading.Condition()
for i in range(10):
    t = threading.Thread(target=func, args=(i, c,))
    t.start()

while True:
    inp = input(">>>")
    if inp == 'q':
        break
    c.acquire()
    c.notify(int(inp))
    c.release()

• 線程鎖條件-condition2

con.wait_for裏傳一個函數名當參數，返回布爾值，是True，就執行下面的代碼。反之，就不執行。

不管是否返回True，都是用了一個線程。

import threading

def condition():
    ret = False
    r = input('>>>')
    if r == 'true':
        ret = True
    else:
        ret = False
    return ret


def func(i,con):
    print(i)
    con.acquire()
    con.wait_for(condition)     # 只能一個一個過
    print(i+100)
    con.release()

c = threading.Condition()
for i in range(10):
    t = threading.Thread(target=func, args=(i,c,))
    t.start()

• 線程鎖定時器

from threading import Timer

def hello():
    print("hello, world")

t = Timer(1, hello)  # 等1秒，執行hello
t.start()  # after 1 seconds, "hello, world" will be printed

2)自定義線程池

• 自定義線程池基礎版

import queue
import threading
import time
class TheadPool:
    def __init__(self,maxsize = 5):
        self.maxsize = maxsize
        self._q = queue.Queue(maxsize)
        for i in range(maxsize):    # 一、初始化的時候，先往隊列裏放5個線程
            self._q.put(threading.Thread)
        # 【threading.Thread, threading.Thread, threading.Thread, threading.Thread】
    def get_thread(self):
        return self._q.get()

    def add_thread(self):
        self._q.put(threading.Thread)

pool = TheadPool(5)

def task(arg,p):
    # 二、線程併發執行，5個線程在瞬間（1秒鐘以內）從隊列裏取出5個（執行get_thread()方法）
    #    5個線程在瞬間打印0-4，就是i
    print(arg)
    time.sleep(1)   # 三、停了5秒
    p.add_thread()  # 四、5個線程執行：隊列添加線程（由於是5個線程執行，一個線程添加一個，隊列總共是5個線程）

for i in range(20): # 五、而後這樣先取走5個，再put5個，而後打印i，就會出現：第一次打印：0-4，而後是：5-9,10-14,15-19
    # threading.Thread類
    t = pool.get_thread()
    obj = t(target=task,args=(i,pool,))
    obj.start()

這個程序的問題：
線程沒有被重用，線程一下開到最大（浪費）

• 自定義線程池思路

不要把隊列裏放線程，而是聽任務，開三個線程來從隊列裏取任務，若是都取完了，就會阻塞

方法1：

設置超時時間

方法2：

往隊列尾部，加三個空值，若是取得是空值，則終止線程。

沒有空閒線程，而且已經建立的線程小於最大的線程數，這樣纔會建立線程。

•  出現的問題

注意；以前的自定義線程池，若是定義queue的最多能放值的個數，pool = ThreadPool(5,5)

terminate就很差使了。

有的時候會一直阻塞，由於隊列裏已經有5個了，再往裏面put一個，就超出queue裏最大的個數。

• 自定義線程池代碼

import queue
import threading
import contextlib
import time

StopEvent = object()
RUN = 0     # 定義線程池的三種狀態
CLOSE = 1
TERMINATE = 2
iNum=0
'''
開啓最大個數爲5個的隊列，
'''

class ThreadPool(object):

    def __init__(self, max_num, max_task_num = None):
        if max_task_num:        # 若是傳了最大隊列數，就設置，不然就是無限大。
            self.q = queue.Queue(max_task_num)
        else:
            self.q = queue.Queue()
        self.max_num = max_num      # 設置最大線程數
        self.cancel = False         # 假如已經執行close了，就再也不執行任務，生成線程處理了
        self.generate_list = []     # 已經生成的線程數列表
        self.free_list = []         # 空閒的線程數列表
        self._state = RUN

    def run(self, func, args, callback=None):
        """
        線程池執行一個任務
        :param func: 任務函數
        :param args: 任務函數所需參數
        :param callback: 任務執行失敗或成功後執行的回調函數，回調函數有兩個參數一、任務函數執行狀態；二、任務函數返回值（默認爲None，即：不執行回調函數）
        :return: 若是線程池已經終止，則返回True不然None
        """
        if self.cancel:         # 假如已經執行close了，就再也不執行任務，生成線程處理了
            return
        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num: # 假如空閒的線程列表爲空，而且已經生成的線程數小於最大線程數
            self.generate_thread()      # 建立線程

        w = (func, args, callback,)     # 把當前任務放入隊列，也就是run循環了300次，就有300個任務放入隊列
        self.q.put(w)       # 注意：隊列數是多少個，就要開啓幾個線程，由於當要關閉的線程池時，
                            # 要把空對象加到隊列。線程判斷獲取到是空對象（此時已經把queue裏的任務都取完了）就關閉線程。

        global iNum
        iNum+=1
        # print('qsize:',str(self.q.qsize()))
    def generate_thread(self):
        """
        建立一個線程
        """
        t = threading.Thread(target=self.call)  # 執行call函數
        t.start()

    def call(self):
        """
        循環去獲取任務函數並執行任務函數
        """
        current_thread = threading.currentThread    # 獲取當前線程
        self.generate_list.append(current_thread)   # 把當前線程加入到已經生成線程列表

        event = self.q.get()        # 從隊列裏取一個任務
        while event != StopEvent:   # 假如 這個任務不是空對象

            func, arguments, callback = event      # 傳進去的任務是個元組，由函數，參數，回調函數組成。
            try:
                result = func(*arguments)       # 執行任務，返回result
                success = True                 # 執行成功，返回狀態爲True
            except Exception as e:
                success = False
                result = None
            else:
                if callback is not None:       # 假若有回調函數
                    try:
                        callback(success, result)   # 把狀態和返回值傳給回調函數執行
                    except Exception as e:
                        pass
            # 執行worker_state函數，空閒線程列表裏是否加入個線程。在yield處執行with下的代碼
            with self.worker_state(self.free_list, current_thread):
                if self._state == TERMINATE:        # 假如線程池狀態是TERMINATE
                    print(11111111111111111111111)
                    event = StopEvent               # 就把當前任務賦值爲空對象，while循環不知足，這樣就走else的內容

                else:
                    event = self.q.get()            # 若是不是TERMINATE狀態，則把當前任務賦給event對象
        else:       # 若是while循環不知足，或者while循環完了，沒有break，就執行else內容。

            self.generate_list.remove(current_thread)   # 隊列獲取到了空對象，就關閉線程（從列表中移除當前的線程）
            print(len(self.generate_list))


    def close(self):        # 先執行close(),再執行join()
        """
        執行完全部的任務後，全部線程中止
        """
        if self._state == RUN:
            self._state = CLOSE
            self.cancel = True
        full_size = len(self.generate_list)     # 查看已經生成的線程數個數
        while full_size:
            self.q.put(StopEvent)       # 往隊列尾部加上一個空對象，因爲隊列是先進先出的，因此空對象是最後獲取的，經過空對象就能關閉線程。
            full_size -= 1      # 循環的次數爲生成的線程的總個數

    def terminate(self):        # 直接執行terminate()
        """
        不管是否還有任務，終止線程
        """
        self._state = TERMINATE
        print("len:",str(len(self.generate_list)))
        while self.generate_list:   # 假如線程列表不爲空，就往隊列里加上空對象
            print('q.qsize():',str(self.q.qsize()))
            self.q.get()
            self.q.put(StopEvent)

        # self.q = queue.Queue()
        print(self.q.empty())       # 查看隊列是否爲空，至關於q.size==0
        print('------------'+str(self.q.qsize()))

    def join(self):     # CLOSE和join結合用
        """Waits until all outstanding tasks have been completed."""
        assert self._state in (CLOSE,)
        delay = 0.0005
        if self._state==CLOSE:
            while self.q.qsize() > 0:
                delay = min(delay * 2, .05)



    @contextlib.contextmanager      # 上下文管理器
    def worker_state(self, state_list, worker_thread):  # 傳入的是空閒線程列表和當前線程
        """
        用於記錄線程中正在等待的線程數
        """
        state_list.append(worker_thread)        # 把當前線程加到空閒線程裏，yield前面的代碼至關於執行__enter__,
        try:
            yield           # yield是執行with worker_state下的代碼，
        finally:            # yield後面的代碼至關於執行__exit__
            state_list.remove(worker_thread)    # 執行完一個queue的全部任務了，就移除這個線程了。由於一個隊列對應着一個線程。


pool = ThreadPool(5,5)

def callback(status, result):
    # status, execute action status
    # result, execute action return value
    pass

def action(i):
    print(i)

for i in range(200):
    ret = pool.run(action, (i,), callback)
pool.terminate()
# pool.close()
# pool.join()
print(1234234523452345234523452345234523452345234523455)
# time.sleep(1)
print(pool.q.qsize())
print(len(pool.generate_list), len(pool.free_list))
print('iNum:',iNum)
# print(len(pool.generate_list), len(pool.free_list))

自定義線程池

 

三. Python 多進程

• 多進程

在windows里加main才能執行，若是在linux不加main能夠執行。

在windows下，若是在程序裏，不方便加main，只能放棄了。

from multiprocessing import Process

def foo(i):
    print('say hi',i)

if __name__ == '__main__':
    for i in range(10):
        p = Process(target=foo,args=(i,))
        p.start()

1）進程之間的數據共享

這個例子是單線程，數據是共享的，不管單線程仍是多線程都是共享的。

from multiprocessing import Process
import multiprocessing

def foo(i,li):
    li.append(i)
    print('say hi',i,li)


if __name__ == '__main__':
    li = []
    for i in range(10):
        # p = Process(target=foo,args=(i,li))
        foo(i,li)
        # p.daemon = True
        # p.start()
        # p.join()

• 進程默認數據不共享

#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'WangQiaomei'

from multiprocessing import Process
import multiprocessing

def foo(i,li):
    li.append(i)
    print('say hi',i,li)


if __name__ == '__main__':
    li = []
    for i in range(10):
        p = Process(target=foo,args=(i,li))
        # foo(i,li)
        # p.daemon = True
        p.start()
        # p.join()

• queues實現：進程之間數據共享

from multiprocessing import Process
from multiprocessing import queues
import multiprocessing

def foo(i,arg):
    arg.put(i)
    print('say hi',i,arg.qsize())


if __name__ == '__main__':
    # li =[]
    li = queues.Queue(20,ctx=multiprocessing)
    for i in range(10):
        p = Process(target=foo,args=(i,li,))
        # p.daemon = True
        p.start()
        # p.join()

• 數組和列表的區別：

數組和列表的特色比較：

1、數組類型必定：

數組只要定義好了，類型必須是一致的

python裏列表裏，能夠放字符串也能夠放數字。

2、數組個數必定：

建立數組的時候，就要指定數組多大，好比數組是10，再添加11個，就會報錯

列表是動態的，個數不必定。

數組和列表的相鄰元素的內存位置比較： 

python的列表是基於c來實現。

python的列表相鄰的兩個元素在內存裏，不必定挨着。是用鏈表實現的。

由於個數不限制，開始是10個長度，因此可能第11個被佔用了。

每一個元素，記錄上一個和下一個的位置在哪裏。能夠找到位置。

字符串和int類型的佔用內存的位置大小確定不同。因此數組，不僅是長度同樣，類型也要同樣。

對於數組的話，相鄰的元素是挨着的。

數組是int類型，而且長度是肯定的。因此是相鄰的。

數組的內存地址是連續的，列表不是，就是鏈表，鏈表是每一個元素記錄上一個位置和下一個位置在哪裏。

• 數組實現：進程之間數據共享

from multiprocessing import Process
from multiprocessing import queues
import multiprocessing
from multiprocessing import Array

def foo(i,arg):
    # arg.put(i)
    # print('say hi',i,arg.qszie())
    arg[i] = i + 100
    for item in arg:
        print(item)
    print("====================")

if __name__ == '__main__':
    # li =[]
    # li = queues.Queue(20,ctx=multiprocessing)
    li = Array('i',5)
    for i in range(5):
        p = Process(target=foo,args=(i,li,))
        # p.daemon = True
        p.start()
        # p.join()

注意：Array的參數，寫了i，就只能放數字：

• dict實現：進程之間數據共享

用Manager()對象建立一個特殊的字典。
For循環建立了多個進程，每一個進程均可以利用dict。
dict.values()就是獲取它全部的值，
若是字典獲取的值是遞增的，說明數據是共享的。

 AttributeError:'ForkAwareLocal' object has no attribute 'connection'

若是把join註釋就會報錯：

報錯：

    conn = self._tls.connection

AttributeError: 'ForkAwareLocal' object has no attribute 'connection'

from multiprocessing import Process
import multiprocessing
from multiprocessing import Manager

def foo(i,arg):
    arg[i] = i + 100
    print(arg.values())

if __name__ == '__main__':
    obj = Manager()
    li = obj.dict()
    for i in range(10):
        p = Process(target=foo,args=(i,li,))
        p.start()
        # p.join()        # 方式1
    # 方式2
    # import time
    # time.sleep(10)

緣由是：
li = obj.dict()是在主進程建立的
for循環裏建立的是子進程，子進程是修改主進程：arg[i] = i + 100（arg就是li）
主進程和子進程都在執行，主進程裏有個字典，子進程要修改這個字典。
進程和進程之間要通訊的話，須要建立鏈接的。至關於兩邊都寫上一個socket，進程之間經過鏈接進行操做。
主進程執行到底部，說明執行完了，會把它裏面的鏈接斷開了。
主進程把鏈接斷開了，子進程就鏈接不上主進程。
若是在底部寫停10秒，主進程就中止下來，並無執行完。主進程沒有執行完，鏈接尚未斷開，那子進程就能夠鏈接它了。

解決方法1：停10秒（不建議）

from multiprocessing import Process
import multiprocessing
from multiprocessing import Manager

def foo(i,arg):
    arg[i] = i + 100
    print(arg.values())

if __name__ == '__main__':
    obj = Manager()
    li = obj.dict()
    for i in range(10):
        p = Process(target=foo,args=(i,li,))
        p.start()
        # p.join()        # 方式1
    # 方式2
    import time
    time.sleep(10)

代碼

測試：

若是把sleep(10)改爲sleep(0.1),那麼只會打印前面幾行，而後又報以前的錯誤

打印：

[102]

[102, 105]

[102, 103, 105]

[102, 103, 105, 107]

接着報錯：

    conn = self._tls.connection

AttributeError: 'ForkAwareLocal' object has no attribute 'connection'

解決辦法2：用join

p.join()
因此使用多進程的常規方法是，先調用start啓動進程，再調用join要求主進程等待當前子進程的結束。
join是用來阻塞當前線程的，每次循環：p.start()以後，p就提示主線程，須要等待p結束才向下執行，那主線程就乖乖的等着啦。

from multiprocessing import Process
import multiprocessing
from multiprocessing import Manager

def foo(i,arg):
    arg[i] = i + 100
    print(arg.values())

if __name__ == '__main__':
    obj = Manager()
    li = obj.dict()
    for i in range(10):
        p = Process(target=foo,args=(i,li,))
        p.start()
        p.join()        # 方式1:
    # 方式2
    # import time
    # time.sleep(10)

 總結：進程之間共享的方式：

queues,數組和字典的方式

dict對類型沒有限制，跟使用字典是如出一轍的。用數組則限制了數據類型。

進程和進程之間要通訊，是要鏈接的。

主進程執行到底部了，就執行完了，就把鏈接斷開了。子進程就連不上主進程了。

2）進程鎖

沒有鎖，多個進程就會一塊兒修改數據：

加把鎖就把進程鎖住了，同一時間只有一個進程能夠運行，其餘都等着。

RLock, Lock, Event, Condition, Semaphore    # 這些方法跟線程的使用方法是同樣的

from multiprocessing import Process
from multiprocessing import queues
from multiprocessing import Array
from multiprocessing import RLock, Lock, Event, Condition, Semaphore    # 這些方法跟線程的使用方法是同樣的
import multiprocessing
import time

def foo(i,lis,lc):
    lc.acquire()        # 加鎖
    lis[0] = lis[0] - 1
    time.sleep(1)
    print('say hi',lis[0])
    lc.release()        # 釋放鎖

if __name__ == "__main__":
    # li = []
    li = Array('i', 1)
    li[0] = 10
    lock = RLock()
    for i in range(10):
        p = Process(target=foo,args=(i,li,lock))
        p.start()

3）進程池

進程池串行-apply

apply從進程池裏取進程，而後一個一個執行，第一個進程執行完，第二個進程才執行，進程之間是串行的操做。這樣就不是併發操做，沒有太大意義。

from multiprocessing import Pool

def f1(arg):
    print(arg)
if __name__ == "__main__":
    pool = Pool(5)

    for i in range(10):
        pool.apply(func=f1,args=(i,))   # apply執行函數，傳入參數
    print('end')

進程池異步-apply_rsync

from multiprocessing import Pool

def f1(arg):
    print(arg)
if __name__ == "__main__":
    pool = Pool(5)

    for i in range(10):
        # pool.apply(func=f1,args=(i,))   # apply執行函數，傳入參數
        pool.apply_async(func=f1,args=(i,))
    print('end')

這10個任務kua一下全執行了，主進程執行到end了。
主進程執行完了，子進程就被終止掉了。
主進程執行完了，就再也不等子線程了，若是要等就要設置參數。
多線程線程默認也是，主進程不等子進程，多線程是：daemon=True加join來讓他等。

主進程等子進程

1. close等子進程所有執行完

join是終止進程，必需要前面執行close或者terminate方法。

執行close，等全部任務（10個）所有執行完，再終止

執行terminate，表示當即終止，無論你當前的任務是否執行完，都當即終止。

from multiprocessing import Pool
import time


def f1(arg):
    time.sleep(1)   # 加這句是爲了看出5個5個執行的效果。
    print(arg)
if __name__ == "__main__":
    pool = Pool(5)

    for i in range(10):
        # pool.apply(func=f1,args=(i,))   # apply執行函數，傳入參數
        pool.apply_async(func=f1,args=(i,))
    pool.close()
    pool.join()     # join表示：主進程執行到這裏的時候，夯住了，等子進程結束的時候，再往下執行。
    print('end')

光執行join，會觸發下面的斷言錯誤：
assert self._state in (CLOSE, TERMINATE)
join源代碼有這句，只有符合這個條件的，纔不會報錯。
這個條件就是：執行join以前，必須執行close或者terminate方法。
close+join：是等子線程所有執行完了，才繼續往下執行。

• terminate當即終止

from multiprocessing import Pool
import time

def f1(arg):
    time.sleep(1)
    print(arg)
if __name__ == "__main__":
    pool = Pool(5)

    for i in range(10):
        # pool.apply(func=f1,args=(i,))   # apply執行函數，傳入參數
        pool.apply_async(func=f1,args=(i,))
    time.sleep(1.5)
    pool.terminate()   # 當即終止
    pool.join()
    print('end')

光執行join，會觸發下面的斷言錯誤：
assert self._state in (CLOSE, TERMINATE)
join源代碼有這句，只有符合這個條件的，纔不會報錯。
這個條件就是：執行join以前，必須執行close或者terminate方法。
terminate+join：是表示當即終止，無論你當前的任務是否執行完，都當即終止。

 

四. 協程

1 協程及gevent原理 

IO密集型：用多線程+gevent（更好），多線程

計算密集型：用多進程

 

協程原理：利用一個線程，分解一個線程成爲多個「微線程」==>程序級別

 

若是寫爬蟲，就訪問別的網站，拿別人源碼。http請求叫IO請求，用多線程。

假設要訪問3個url，建立3個線程，都在等待着，第一個有數據返回就繼續執行，以此類推。

在等待過程當中，就什麼事也沒幹。

 

協程的方式。

計算機幫你建立進程、線程。線程是人爲建立出來的。用一個線程，一下子執行這個操做，一下子執行那個操做。

協程是隻用一個線程。程序員利用io多路複用的方式，讓協程：

先訪問一個url，不等待返回，就再訪問第二個url，訪問第三個url，而後也在等待。

greenlet本質是實現協程的。

注意：協程自己不高效，協程的本質只是程序員調用的，那爲啥gevent這麼高效率呢，是由於用了協程（greenlet）+IO多路複用的方式。

是IO多路複用的用法才能高效。因此用的時候就用gevent就行了。

用多線程：假設每爬一個網址須要2秒，3個url，就是3個請求，等待2秒，就能夠繼續往下走。

若是用gevent，用單線程，單線程應該從上到下執行，用for循環讀取3個url，往地址發送url請求，就是IO請求，線程是不等待的。

for循環再拿第二個url，再發第三個url。在這過程當中，誰先回來，就處理誰。

 

資源佔用上，多線程佔用了3個線程，2秒鐘，多線程啥也沒幹，在等待。gevent在2秒鐘，只要發送請求了，接着就想幹什麼幹什麼。

2. greenlet協程

from greenlet import greenlet

def test1():
    print(12)
    gr2.switch()
    print(34)
    gr2.switch()

def test2():
    print(56)
    gr1.switch()
    print(78)

gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

3. gevent

gevent切換協程：

import gevent

def foo():
    print('Running in foo')         # 第1步
    gevent.sleep(0)
    print('Explicit context switch to foo again')   # 第3步

def bar():
    print('Explicit context to bar')    # 第2步
    gevent.sleep(0)
    print('Implicit context switch back to bar')    # 第4步

gevent.joinall([
    gevent.spawn(foo),
    gevent.spawn(bar),
])

 4.gevent使用方法

from gevent import monkey; monkey.patch_all()
import gevent
import requests

# 這個函數是發http請求的
def f(url):
    print('GET: %s' % url)
    resp = requests.get(url)
    data = resp.text    # 獲取內容
    print('%d bytes received from %s.' % (len(data), url))

gevent.joinall([
        gevent.spawn(f, 'https://www.python.org/'),     # 建立了一個協程
        gevent.spawn(f, 'https://www.yahoo.com/'),      # 建立了一個協程
        gevent.spawn(f, 'https://github.com/'),         # 建立了一個協程
])

建立了三個協程。總共就一個線程，經過for循環發送三個url請求。而後等待結果，誰先回來，就處理誰。
經過requests.get(url)發送url請求，誰先回來，就拿到數據（data），拿到數據就能夠處理數據了。

gevent的使用場景舉例：

一、scrapy框架內部用的gevent。發請求性能比線程高不少。

二、作api（url）監控，把代碼發佈到哪一個url，得自動檢測下返回值是否是200，或是指定的狀態碼。

發佈完成以後，就要發送http請求過去檢測一下返回的狀態碼。若是有20個url請求，就用gevent一下全給發了，就不必建立多個線程，一個線程就足以了，而後配合多進程+gevent，又能夠利用多顆cpu的優點了。

monkey.patch_all()是什麼？

發送http請求，是request本質上調用socket來發。原來執行http請求，就會通知我一下，執行完了，默認socket是沒有這個功能的。這至關於把原來的socket修改了，修改爲特殊功能的socket，發送請求若是完事了，會告訴你完事了。

其實內部就是把io請求作了個封裝而已。