python 之 並發編程（進程池與線程池、同步異步阻塞非阻塞、線程queue）

9.11 進程池與線程池

池子使用來限制並發的任務數目，限制我們的計算機在一個自己可承受的范圍內去並發地執行任務

池子內什么時候裝進程：並發的任務屬於計算密集型 池子內什么時候裝線程：並發的任務屬於IO密集型

進程池：

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
​
def task(x):
    print('%s 接客' %os.getpid())
    time.sleep(random.randint(2,5))
    return x**2
​
if __name__ == '__main__':  # ProcessPoolExecutor創建並開啟指定數目的進程
    p=ProcessPoolExecutor() # 默認開啟的進程數是cpu的核數
​
    for i in range(20):
        p.submit(task,i)    # 一下並行執行四個任務，等其中一個任務執行完后再執行下一個

線程池：

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
​
def task(x):
    print('%s 接客' %x)
    time.sleep(random.randint(2,5))
    return x**2
​
if __name__ == '__main__':  # ThreadPoolExecutor創建並開啟指定數目的線程
    p=ThreadPoolExecutor(4) # 默認開啟的線程數是cpu的核數*5
​
    for i in range(20):
        p.submit(task,i)    # 一下並發執行四個任務，等其中一個任務執行完后再並發執行下一個

9.112 基於多線程實現並發的套接字通信(使用線程池)

服務端：

from socket import *
from threading import Thread
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
​
tpool=ThreadPoolExecutor(3)         #ThreadPoolExecutor創建並開啟指定數目的線程
def communicate(conn,client_addr):
    while True:  # 通訊循環
        try:
            data = conn.recv(1024)
            if not data: break
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()
​
def server():
    server=socket(AF_INET,SOCK_STREAM)
    server.bind(('127.0.0.1',8080))
    server.listen(5)
​
    while True: # 鏈接循環
        conn,client_addr=server.accept()
        print(client_addr)
        # t=Thread(target=communicate,args=(conn,client_addr))
        # t.start()
        tpool.submit(communicate,conn,client_addr)#一下並發執行3個任務，等其中一個任務執行完后再並發執行下一個
    server.close()
​
if __name__ == '__main__':
    server()

客戶端：

from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))
​
while True:
    msg=input('>>>: ').strip()
    if not msg:continue
    client.send(msg.encode('utf-8'))
    data=client.recv(1024)
    print(data.decode('utf-8'))
​
client.close()

9.12 同步異步阻塞非阻塞

阻塞與非阻塞指的是程序的兩種運行狀態:

阻塞：遇到 I/O 就發生阻塞，程序一旦遇到阻塞操作就會停在原地，並且立刻釋放CPU資源

非阻塞（就緒態或運行態）：沒有遇到 I/O 操作，或者通過某種手段讓程序即便是遇到 I/O 操作也不會停在原地，執行其他操作，力求盡可能多的占有CPU

同步與異步指的是提交任務的兩種方式：

同步調用：提交完任務后，就在原地等待，直到任務運行完畢后，拿到任務的返回值，才繼續執行下一行代碼

異步調用：提交完任務后，不在原地等待，直接執行下一行代碼

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
#from multiprocessing import Pool
def task(x):
    print('%s 接客' %x)
    time.sleep(random.randint(1,3))
    return x**2
​
if __name__ == '__main__':
    # 異步調用
    p=ThreadPoolExecutor(4) # 默認開啟的線程數是cpu的核數*5
    obj_l=[]
    for i in range(10):
        obj=p.submit(task,i)
        obj_l.append(obj)
​
    # p.close()
    # p.join()
    p.shutdown(wait=True)# shutdown指的是不能再往進程池內提交任務,wait=True指等待進程池或線程池內所有的任務都運行完畢
    print(obj_l[3].result()) # 9    #最后拿結果
    print('主')
​
    # 同步調用
    p=ThreadPoolExecutor(4) # 默認開啟的線程數是cpu的核數*5
    for i in range(10):
        print(p.submit(task,i).result())
    print('主')

9.121 異步調用+回調機制

問題：

1、任務的返回值不能得到及時的處理，必須等到所有任務都運行完畢才能統一進行處理

2、解析的過程是串行執行的,如果解析一次需要花費2s,解析9次則需要花費18s

基於進程池：

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import requests
import os
import time
import random
​
def get(url):
    print('%s GET %s' %(os.getpid(),url))
    response=requests.get(url)
    time.sleep(random.randint(1,3))
    if response.status_code == 200:
        return response.text
​
def pasrse(obj):            # 干解析的活
    res=obj.result()        # 回調拿結果
    print('%s 解析結果為：%s' %(os.getpid(),len(res))) # 4108 解析結果為：2443
​
if __name__ == '__main__':
    urls=[
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.python.org',
    ]
​
    pool=ProcessPoolExecutor(4)
    for url in urls:
        obj=pool.submit(get,url)    #parse函數會在obj對應的任務執行完畢后自動執行，會把obj自動傳給parse
        obj.add_done_callback(pasrse)   #四個進程並發爬取信息，主進程在執行解析操作
​
    print('主進程',os.getpid())         # 主進程 4108

基於線程池：

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread
import requests
import os
import time
import random
​
def get(url):
    print('%s GET %s' %(current_thread().name,url))
    response=requests.get(url)
    time.sleep(random.randint(1,3))
    if response.status_code == 200:
        return response.text
​
def pasrse(obj):     # 干解析的活
    res=obj.result()
    print('%s 解析結果為：%s' %(current_thread().name,len(res)))#ThreadPoolExecutor-0_1 解析結果為：
                                                         #2443
if __name__ == '__main__':                              #ThreadPoolExecutor-0_3 解析結果為：2443
    urls=[
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.python.org',
    ]
​
    pool=ThreadPoolExecutor(4)
    for url in urls:
        obj=pool.submit(get,url)    #parse函數會在obj對應的任務執行完畢后自動執行，會把obj自動傳給parse
        obj.add_done_callback(pasrse)     #四個線程並發爬取信息，空閑者執行解析操作
    print('主線程',current_thread().name)  #主線程 MainThread

9.13 線程queue

隊列：先進先出 queue.Queue()

import queue
q=queue.Queue(3) 
​
q.put(1)
q.put(2)
q.put(3)
# q.put(4)   阻塞
​
print(q.get())  #1
print(q.get())  #2
print(q.get())  #3

堆棧：后進先出 queue.LifoQueue()

import queue
q=queue.LifoQueue(3) 
​
q.put('a')
q.put('b')
q.put('c')
​
print(q.get())  #c
print(q.get())  #b
print(q.get())  #a

優先級隊列:可以以小元組的形式往隊列里存值，第一個元素代表優先級，數字越小優先級越高

PriorityQueue()

import queue
q=queue.PriorityQueue(3) 
q.put((10,'user1'))
q.put((-3,'user2'))
q.put((-2,'user3'))
​
print(q.get())  #(-3, 'user2')
print(q.get())  #(-2, 'user3')
print(q.get())  #(10, 'user1')