阻塞、非阻塞、同步與異步

阻塞、非阻塞、同步與異步

阻塞與非阻塞

進程運行的三種狀態：運行、就緒、阻塞

阻塞和非阻塞：

​ 阻塞：程序運行時,遇到了IO,程序掛起,cpu被切走.

​ 非阻塞： 程序沒有遇到IO,程序遇到IO可是我經過某種手段,讓cpu強行運行個人程序.

提交任務的角度:

​ 同步： 提交一個任務,自任務開始運行直到此任務結束(可能有IO),返回一個返回值以後,我在提交下一個任務.

​ 異步：一次提交多個任務,而後我就直接執行下一行代碼.

同步調用與異步調用

同步調用：

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time
import random
import os

def task(i):
    print(f'{os.getpid()}開始任務')
    time.sleep(random.randint(1,3))
    print(f'{os.getpid()}任務結束')
    return i
if __name__ == '__main__':

    # 同步調用
    pool = ProcessPoolExecutor()
    for i in range(10):
        obj = pool.submit(task,i)
        # obj是一個動態對象,返回的當前的對象的狀態,有可能運行中,可能(就緒阻塞),還多是結束了.
        # obj.result() 必須等到這個任務完成後,返回告終果以後,在執行下一個任務.
        print(f'任務結果:{obj.result()}')  # 進程執行完成後返回結果

    pool.shutdown(wait=True)
    # shutdown: 讓個人主進程等待進程池中全部的子進程都結束任務以後,在執行. 有點相似與join.
    # shutdown: 在上一個進程池沒有完成全部的任務以前,不容許添加新的任務.
    # 一個任務是經過一個函數實現的,任務完成了他的返回值就是函數的返回值.
    print('===主')

異步調用：

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time
import random
import os

def task(i):
    print(f'{os.getpid()}開始任務')
    time.sleep(random.randint(1,3))
    print(f'{os.getpid()}任務結束')
    return i
if __name__ == '__main__':

    # 異步調用
    pool = ProcessPoolExecutor()
    for i in range(10):
        pool.submit(task,i)  # 未解決異步調用返回值問題

    pool.shutdown(wait=True)
    print('===主')

異步調用獲取結果的兩種方式：

​ 方式1：統一回收結果: 任務結束後對全部動態對象取結果，獲取函數返回值

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time
import random
import os

def task(i):
    print(f'{os.getpid()}開始任務')
    time.sleep(random.randint(1,3))
    print(f'{os.getpid()}任務結束')
    return i

if __name__ == '__main__':

    # 異步調用
    pool = ProcessPoolExecutor()
    l1 = []
    for i in range(10):
        obj = pool.submit(task,i)
        l1.append(obj)

    pool.shutdown(wait=True)
    print(l1)
    for i in l1:
        print(i.result())
    print('===主')
# 結果：
12708開始任務
8632開始任務
1848開始任務
14544開始任務
10704開始任務
18776開始任務
18480開始任務
18548開始任務
13916開始任務
17144開始任務
1848任務結束
14544任務結束
18548任務結束
8632任務結束
10704任務結束
18480任務結束
13916任務結束
17144任務結束
12708任務結束
18776任務結束
[<Future at 0x232b4a377b8 state=finished returned int>, <Future at 0x232b4a82c88 state=finished returned int>, <Future at 0x232b4a82d30 state=finished returned int>, <Future at 0x232b4a82dd8 state=finished returned int>, <Future at 0x232b4a82e80 state=finished returned int>, <Future at 0x232b4a82f28 state=finished returned int>, <Future at 0x232b4a8d048 state=finished returned int>, <Future at 0x232b4a8d128 state=finished returned int>, <Future at 0x232b4a8d208 state=finished returned int>, <Future at 0x232b4a8d2e8 state=finished returned int>]
0
1
2
3
4
5
6
7
8
9
===主

​ 方式2：異步調用+回調函數

異步調用+回調函數

requests模塊：

瀏覽器工做原理： 服務端發送一個請求,服務端驗證你的請求,若是正確,給你的瀏覽器返回一個文件，瀏覽器接收到文件,將文件裏面的代碼渲染成你看到的漂亮美麗的模樣.

爬蟲原理：

1. 利用代碼模擬一個瀏覽器,進行瀏覽器的工做流程獲得一堆源代碼.

2. 對源代碼進行數據清洗獲得我想要數據.

import requests
ret = requests.get('http://www.baidu.com')
if ret.status_code == 200:
    print(ret.text)

引入回調函數的三個過程：

版本1：

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import requests

def task(url):
    content = requests.get(url)
    return content.text

def parse(obj):
    return len(obj.result())

if __name__ == '__main__':
    pool = ThreadPoolExecutor(4)
    url_list = ['http://www.JD.com','http://www.JD.com', 'https://home.cnblogs.com/u/lifangzheng/',
         'https://wizardforcel.gitbooks.io/gopl-zh/content/ch0/ch0-01.html', 'https://www.pypypy.cn/#/',
         'https://www.liaoxuefeng.com/', 'https://home.cnblogs.com/u/lifangzheng/',
         'https://home.cnblogs.com/u/lifangzheng/', 'https://gitee.com/clover16', 'https://gitee.com/clover16']
    obj_list = []
    for url in url_list:
        obj = pool.submit(task,url)
        obj_list.append(obj)

    pool.shutdown(wait=True)
    for res in obj_list:
        print(parse(res.result()))
# 版本一的兩個缺陷:
#    1. 異步發出10個任務,併發的執行,可是統一的接收全部的任務的返回值.(效率低,不能實時的獲取結果)
#    2. 分析結果流程是串行,影響效率.

版本2：（基於版本一的第二個缺點改進而來）

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import requests

def task(url):
    content = requests.get(url)
    return parse(content.text)

def parse(obj):
    return len(obj.result())  # 嵌套函數，併發執行中分析結果，增長了函數的耦合性
# 併發執行任務,每一個任務是經過網頁獲取源碼+數據分析,此任務最好是IO阻塞,才能發揮最大的效果
if __name__ == '__main__':
    pool = ThreadPoolExecutor(4)
    url_list = ['http://www.JD.com','http://www.JD.com', 'https://home.cnblogs.com/u/lifangzheng/',
         'https://wizardforcel.gitbooks.io/gopl-zh/content/ch0/ch0-01.html', 'https://www.pypypy.cn/#/',
         'https://www.liaoxuefeng.com/', 'https://home.cnblogs.com/u/lifangzheng/',
         'https://home.cnblogs.com/u/lifangzheng/', 'https://gitee.com/clover16', 'https://gitee.com/clover16']
    obj_list = []
    for url in url_list:
        obj = pool.submit(task,url)
        obj_list.append(obj)

    pool.shutdown(wait=True)
    for res in obj_list:
        print(parse(res.result()))

版本3：

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import requests

def task(url):
    content = requests.get(url)
    return content.text

def parse(obj):
    print(len(obj.result()))

if __name__ == '__main__':
    pool = ThreadPoolExecutor(4)
    url_list = ['http://www.JD.com','http://www.JD.com', 'https://home.cnblogs.com/u/lifangzheng/',
         'https://wizardforcel.gitbooks.io/gopl-zh/content/ch0/ch0-01.html', 'https://www.pypypy.cn/#/',
         'https://www.liaoxuefeng.com/', 'https://home.cnblogs.com/u/lifangzheng/',
         'https://home.cnblogs.com/u/lifangzheng/', 'https://gitee.com/clover16', 'https://gitee.com/clover16']
    for url in url_list:
        obj = pool.submit(task,url)
        obj.add_done_callback(parse)  # add_done_callback函數無返回值
        # 線程發佈後，由空閒線程執行回調函數

PS：異步處理IO類型，回調處理非IO類型

異步調用和回調函數的關係？

異步站在發佈任務的角度，回調站在接收結果的角度: 回調函數 按順序接收每一個任務的結果,進行下一步處理.

線程池+回調函數和進程池+回調函數的小區別：

進程池+回調: 回調函數由主進程去執行，線程池+回調: 回到函數由空閒的線程去執行.