Python并發編程06 /同步/異步調用/異步調用+回調函數

1.如何看源碼

2.昨日回顧

          
            #1.遞歸鎖：RLock，同一把鎖，引用一次計數+1，釋放一次計數-1，只要計數不為零，其他線程就搶不到，能解決死鎖問題。
#2.信號量：同一時刻可以設置搶鎖的線程或者進程的數量
#3.GIL鎖：全局解釋器鎖，同一時刻只能一個線程進入解釋器，Cpython解釋器具有的。
#   優點：保證了解釋器的資源數據的安全
#   缺點：單進程的多線程不能利用多核（并行）
#4.GIL鎖與自己的鎖的區別聯系：
#   相同點：都是互斥鎖
#   GIL鎖保證了解釋器數據資源的安全
#   自己的鎖保證的是進程線程中的數據安全
#5.線程池、進程池：控制開啟線程或者進程的數量
#6.IO密集型：單進程的多線程并發
#  計算密集型：多進程并發或者并行
          
        

3.阻塞、同步調用、異步調用

3.1概念

          
            進程運行的三個狀態：運行，就緒，阻塞
執行的角度：
    阻塞：程序運行時，遇到了IO，程序掛起，CPU被切走
    非阻塞：程序沒有遇到IO，程序遇到IO但是通過某種手段，讓CPU強行運行我的程序
提交任務的角度：
    同步：提交一個任務，自任務開始運行直到此任務結束（可能有IO），返回一個返回值之后，我再提交下一個任務
    異步：一次提交多個任務，然后就執行下一行代碼
    返回結果如何回收？
    
#給三個人發布任務:
#同步: 先告知第一個人完成寫書的任務,我從原地等待,等他兩天之后完成了,告訴完事了,我在發布下一個任務......
#異步: 直接將三個任務告知三個人,我就忙我的我,直到三個人完成之后,告知我.    
          
        

3.2異步調用

          
            異步調用返回值如何接收？
# 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)
#     # shutdown: 讓我的主進程等待進程池中所有的子進程都結束任務之后,在執行. 有點類似與join.
#     # shutdown: 在上一個進程池沒有完成所有的任務之前,不允許添加新的任務.
#     # 一個任務是通過一個函數實現的,任務完成了他的返回值就是函數的返回值.
#     print('===主')
          
        

3.3同步調用

          
            # 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('===主')
          
        

3.4異步如何取結果

          
            方式一: 異步調用,統一回收結果.
# 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('===主')
# 統一回收結果: 我不能馬上收到任何一個已經完成的任務的返回值,我只能等到所有的任務全部結束統一回收.

          
        

4.異步調用+回調函數

4.1瀏覽器工作原理

          
            #向服務器發送一個請求，服務端驗證你的請求，如果正確，給你的瀏覽器返回一個文件
#瀏覽器接收到文件，將文件里面的代碼渲染成你看到的好看的模樣。
          
        

4.2什么叫爬蟲

          
            #1.利用代碼模擬一個瀏覽器，進行瀏覽器的工作流程得到一堆源代碼
#2.對源代碼進行數據清洗得到想要的數據
          
        

4.3異步調用+回調函數

          
            # 版本一:
# from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
# import time
# import random
# import os
# import requests
# def task(url):
#     '''模擬的就是爬取多個源代碼 一定有IO操作'''
#     ret = requests.get(url)
#     if ret.status_code == 200:
#         return ret.text
# def parse(content):
#     '''模擬對數據進行分析 一般沒有IO'''
#     return len(content)
# if __name__ == '__main__':
#     # 開啟線程池,并發并行的執行
#     url_list = [
#         'http://www.baidu.com',
#         'http://www.JD.com',
#         'http://www.JD.com',
#         'http://www.JD.com',
#         'http://www.taobao.com',
#         'https://www.cnblogs.com/jin-xin/articles/7459977.html',
#         'https://www.luffycity.com/',
#         'https://www.cnblogs.com/jin-xin/articles/9811379.html',
#         'https://www.cnblogs.com/jin-xin/articles/11245654.html',
#         'https://www.sina.com.cn/',
#     ]
#     pool = ThreadPoolExecutor(4)
#     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()))
# # '''
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# print('===主')
# 版本一:
#     1. 異步發出10個任務,并發的執行,但是統一的接收所有的任務的返回值.(效率低,不能實時的獲取結果)
#     2. 分析結果流程是串行,影響效率.

版本二: 針對版本一的缺點2,改進,讓串行編程并發或者并行.
# from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
# import time
# import random
# import os
# import requests
#
# def task(url):
#     '''模擬的就是爬取多個源代碼 一定有IO操作'''
#     ret = requests.get(url)
#     if ret.status_code == 200:
#         return parse(ret.text)
# def parse(content):
#     '''模擬對數據進行分析 一般沒有IO'''
#     return len(content)
# if __name__ == '__main__':
#     # 開啟線程池,并發并行的執行
#     url_list = [
#         'http://www.baidu.com',
#         'http://www.JD.com',
#         'http://www.JD.com',
#         'http://www.JD.com',
#         'http://www.taobao.com',
#         'https://www.cnblogs.com/jin-xin/articles/7459977.html',
#         'https://www.luffycity.com/',
#         'https://www.cnblogs.com/jin-xin/articles/9811379.html',
#         'https://www.cnblogs.com/jin-xin/articles/11245654.html',
#         'https://www.sina.com.cn/',
#     ]
#     pool = ThreadPoolExecutor(4)
#     obj_list = []
#     for url in url_list:
#         obj = pool.submit(task, url)
#         obj_list.append(obj)
#     '''
#     # 1 在開一個線程進程池,并發并行的處理. 再開一個線程進程池,開銷大.
#     # 2 將原來的任務擴大,
#     版本一:
#         線程池設置4個線程, 異步發起10個任務,每個任務是通過網頁獲取源碼, 并發執行,
#         最后統一用列表回收10個任務, 串行著分析源碼.
#     版本二:
#         線程池設置4個線程, 異步發起10個任務,每個任務是通過網頁獲取源碼+數據分析, 并發執行,
#         最后將所有的結果展示出來.
#         耦合性增強了.
#         并發執行任務,此任務最好是IO阻塞,才能發揮最大的效果
#     '''
#     pool.shutdown(wait=True)
#     for res in obj_list:  # [obj1, obj2,obj3....obj10]
#         print(res.result())

版本三:
# 基于 異步調用回收所有任務的結果我要做到實時回收結果,
# 并發執行任務每個任務只是處理IO阻塞的,不能增加新得功能.
# 異步調用 + 回調函數
# from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
# import time
# import random
# import os
# import requests
# 
# def task(url):
#     '''模擬的就是爬取多個源代碼 一定有IO操作'''
#     ret = requests.get(url)
#     if ret.status_code == 200:
#         return ret.text
# def parse(obj):
#     '''模擬對數據進行分析 一般沒有IO'''
#     print(len(obj.result()))
# 
# if __name__ == '__main__':
#     # 開啟線程池,并發并行的執行
#     url_list = [
#         'http://www.baidu.com',
#         'http://www.JD.com',
#         'http://www.JD.com',
#         'http://www.JD.com',
#         'http://www.taobao.com',
#         'https://www.cnblogs.com/jin-xin/articles/7459977.html',
#         'https://www.luffycity.com/',
#         'https://www.cnblogs.com/jin-xin/articles/9811379.html',
#         'https://www.cnblogs.com/jin-xin/articles/11245654.html',
#         'https://www.sina.com.cn/',
#     ]
#     pool = ThreadPoolExecutor(4)
#     for url in url_list:
#         obj = pool.submit(task, url)
#         obj.add_done_callback(parse)
# 
#     '''
#     線程池設置4個線程, 異步發起10個任務,每個任務是通過網頁獲取源碼, 并發執行,
#     當一個任務完成之后,將parse這個分析代碼的任務交由剩余的空閑的線程去執行,你這個線程繼續去處理其他任務.
#     如果進程池+回調: 回調函數由主進程去執行.
#     如果線程池+回調: 回到函數由空閑的線程去執行.

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

# 異步 + 回調:
# 異步處理的IO類型.
# 回調處理非IO