協程,事件,隊列,同步,非同步,回調函數

来源:https://www.cnblogs.com/tangjian219/archive/2019/08/28/11426743.html
-Advertisement-
Play Games

協程 Event事件 python 添加全局變數,修改全局變數,實現一個線程在某一個節點讓下一個線程繼續工作 import time from threading import Thread from threading import current_thread flag = False def ...


協程

什麼是協成?單個線程併發的處理多個任務,程式控制協成的切換+保持狀態,協成的切換速度非常快,矇蔽了操作系統的眼睛,讓操作系統認為CPU一直在運行
進程或線程都是由操作系統控制CPU來回切換,遇到阻塞就切換執行其他任務,協成是程式控制的,霸占CPU執行任務,會在操作系統控制CPU之前來回切換,操作系統就認為CPU一直在運作

協程的優點:
    1.開銷小
    2.運行速度快
    3.協程會長期霸占CPU只執行我程式里的所有任務
協程的缺點:
    1.協程屬於微併發,處理任務不易過多
    2.協程的本質是單線程,無法利用多核,可以一個程式開啟多個進程,每個進程開啟多個線程.每個線程開啟協程
協程處理io密集型比較好

協程的特點:
    1.必須在只有一個單線程里實現併發
    2.修改共用數據不需要加鎖
    3.保持狀態
    4.一個協程遇到io操作就自動切換到其他協程 
    
工作中:
​   一般在工作中我們都是進程+線程+協程的方式來實現併發，以達到最好的併發效果，如果是4核的cpu，一般起5個進程，每個進程中20個線程（5倍cpu數量），每個線程可以起500個協程，大規模爬取頁面的時候，等待網路延遲的時間的時候，我們就可以用協程去實現併發。 併發數量 = 5 * 20 * 500 = 50000個併發，這是一般一個4cpu的機器最大的併發數。nginx在負載均衡的時候最大承載量就是5w個
#當一個任務沒有阻塞的時候
import time
def task():
    res = 1
    for i in range(1,1000000):
        res += 1
def task1():
    res = 1
    for i in range(1,1000000):
        res -= 1
strt = time.time()
task()
task1()
end = time.time()
print(f"串列執行效率:{end - strt}") #串列執行效率:0.07783079147338867


#第一次接觸協程是yield
#沒有io阻塞
import time
def task():
    res = 1
    for i in range(1,1000000):
        res += 1
        yield res
def task1():
    g = task()
    res = 1
    for i in range(1,1000000):
        res -= 1
        next(g)
strt = time.time()
task()
task1()
end = time.time()
print(f"協成執行效率:{end - strt}")#協成執行效率:0.21143341064453125
純計算密集型,沒有io情況下,串列的執行效率高於協程


# 並不是真正的阻塞
import gevent
def eat(name):
    print(f"{name} eat 1")   #1
    gevent.sleep(2)    #模擬的是gevent可以識別的阻塞
    print(f"{name} eat 2")   #4
def play(name):
    print(f"{name} play 1")   #2
    gevent.sleep(1)
    print(f"{name} play 2")   #3
g1 = gevent.spawn(eat,'八戒')
g2 = gevent.spawn(play,name = '悟空')
g1.join()
g2.join()
print("主")                    #5



import threading
from gevent import monkey
import gevent
import time
monkey.patch_all()        # 打補丁:將下麵的所有的任務的阻塞打上標記,遇到這個標記就切換

def eat():
    print(f"線程1:{threading.current_thread().getName()}")    #1
    print('eat food 1')                                       #2
    time.sleep(2)                                             #3阻塞,去執行別的任務
    print('eat food 2')                                       #8

def play():
    print(f"線程2:{threading.current_thread().getName()}")    #4
    print('play 1')                                           #5
    time.sleep(1)                                             #6阻塞,再去執行別的任務,上一個任務還在阻塞,繼續向下執行
    print('play 2')                                           #7
g1 = gevent.spawn(eat)
g2 = gevent.spawn(play)
gevent.joinall([g1,g2])
print(f"{threading.current_thread().getName()}")              #9
註意在各個任務中的阻塞時間

Event事件

添加全局變數,修改全局變數,實現一個線程在某一個節點讓下一個線程繼續工作
import time
from threading import Thread
from threading import current_thread
flag = False
def task():
    print(f"{current_thread().name}檢測伺服器是否正常開啟.....")
    time.sleep(2)
    global flag
    flag = True
def task1():
    while 1:
        time.sleep(1)
        print(f"{current_thread().name}正在連接伺服器....")
        if flag:
            print(f"{current_thread().name}連接成功")
            return

if __name__ == '__main__':
    t1 = Thread(target=task1)
    t2 = Thread(target=task1)
    t3 = Thread(target=task1)
    t = Thread(target=task)

    t.start()
    t1.start()
    t2.start()
    t3.start()
    
    
import time
from threading import Thread
from threading import current_thread
from threading import Event
import random
event = Event()   #預設為False
def task1():
    print(f"{current_thread().getName()} 檢測伺服器知否開啟...") #獲取線程名字
    time.sleep(random.randint(1,3))
def task2():
    print(f"{current_thread().getName()}正在嘗試連接伺服器...")
    count = 1
    while count < 4:
        time.sleep(1)
        print(f"{current_thread().getName()}連接第{count}次")
        if count < 4:
            time.sleep(0.5)
            event.set() #將event狀態修改為True
            print('連接成功')
        count += 1
t1 = Thread(target=task1)
t2 = Thread(target=task2)
t1.start()
t2.start()



import time
from threading import Thread
from threading import current_thread
from threading import Event
import random
event = Event()
def check():
    print(f"{current_thread().name}檢測伺服器是否開啟...")
    time.sleep(random.randint(1,3))
    event.set()
    print("連接成功")
def connect():
    count = 1
    while not event.is_set(): #event.is_set()判斷狀態是True或False
        if count == 4:
            print('連接次數過多,已斷開')
            break
        event.wait(1) #輪詢檢測event狀態,如果為真,就向下執行,是個阻塞, # 只阻塞1秒,1秒之後如果還沒有進行set 直接進行下一步操作.
        print(f"{current_thread().name}嘗試連接{count}次")
        count += 1
    else:
        print(f"{current_thread().name}連接成功")
t1 = Thread(target=check)
t2 = Thread(target=connect)
t1.start()
t2.start()

線程隊列

import queue
# 先進先出原則
q = queue.Queue(3)
q.put(1)
q.put(2)
q.put(3)
print(q.get())
print(q.get())
print(q.get())
# print(q.get(block=False)) #設置多取會報錯
q.get(timeout=2)#阻塞兩秒,後報錯
# print(q.get())多去一個會阻塞



# 後進先出
q = queue.LifoQueue(4)
q.put(1)
q.put(2)
q.put('alex')
q.put('太白')

print(q.get())
print(q.get())
print(q.get())
print(q.get())
# 太白
# alex
# 2
# 1

優先順序隊列
最小的先取出
q = queue.PriorityQueue(4)
q.put((1,'qq'))
q.put((-2,'qq1'))
q.put((0,'qq2'))

print(q.get())
print(q.get())
print(q.get())

#(-2, 'qq1')
#(0, 'qq2')
#(1, 'qq')

同步

同步調用(提交完任務後，就在原地等待任務執行完後，拿到結果，再執行下一行代碼)
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):                #同時開啟10個進程
        obj = pool.submit(task,i)      #類似於發佈任務
        print(f"任務結果:{obj.result()}")  # 對象加result()就變成同步,獲取的是運行狀態.obj就為動態對象, (running,pending,finish)  動態對象.result()獲取結果
                                       #obj.result() 必須等到這個任務完成後,返回了結果,在執行下一個任務
    pool.shutdown(wait=True)           #shutdown 讓我的主進程等待進程池中所有子進程結束後,在執行,類似於join
                                       #在上一個進程池沒有完成任務之前,不允許添加新的任務,一個任務通過一個函數實現,任務完成了他的返回值就是函數的返回值
    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) #等待進程池中所有子進程結束後在向下執行,類似於join
    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()
    l = []
    for i in range(10):
        obj = pool.submit(task,i)
        l.append(obj)
    pool.shutdown(wait=True)
    for i in l:
        print(i.result())
    print("主")
雖然是同時發佈了任務,但是在回收結果的時候,不能馬上收到一個結束任務的返回值,只能等所有任務結束後統一收到結果 


非同步調用的如何取值?
1.
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time
import random
import os
import requests
# 模擬的就是爬取多個源代碼 一定有IO操作
def task(url):
    ret = requests.get(url)
    if ret.status_code == 200:
        return ret.text
def parse(content):
    return len(content)

if __name__ == '__main__':
    ret = task('http://www.baidu.com')
    print(parse(ret))

    ret = task('http://www.JD.com')
    print(parse(ret))

    ret = task('http://www.taobao.com')
    print(parse(ret))

    ret = task('https://www.cnblogs.com/jin-xin/articles/7459977.html')
    print(parse(ret))
#這樣寫為執行一個task函數,在執行一個parse函數,執行效率低,一個任務執行2秒,20個任務就是40秒,耗時
#這兩個函數為一個任務,當一個任務執行完成後再執行下一個任務,串列

2.

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time
import random
import os
import requests
# 模擬的就是爬取多個源代碼 一定有IO操作
def task(url):
    ret = requests.get(url)
    if ret.status_code == 200:
        return ret.text
def parse(content):
    return len(content)

if __name__ == '__main__':
#     開啟一個線程池,併發執行
    url_list = ['http://www.baidu.com',
                'http://www.JD.com',
                'http://www.taobao.com',
                'https://www.cnblogs.com/jin-xin/articles/7459977.html'
    ]
    pool = ThreadPoolExecutor(4)
    obj_list = []

    for url in url_list:
        obj = pool.submit(task,url)
        obj_list.append(obj)
    pool.shutdown(wait=True)
#這個for迴圈把爬取的源碼添加到一個類表中,對象,執行task函數
    for i in obj_list:
        print(parse(i.result()))
迴圈取出類表中的元素,當做參數傳入parse函數中,進行數據分析
以上版本缺點:非同步的發出10個任務,併發的執行任務,但是分析結果的流程是串列,沒有做到結束一個任務就返回一個返回值,效率低

3.

def task(url):
    ret = requests.get(url)
    if ret.status_code == 200:
        return parse(ret.text)    #更改函數,直接返回並調用parse函數
def parse(content):
    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 i in obj_list:
        print(i.result())#直接返回結果,不用再調用函數,直接全部列印出來結果

4.非同步調用+回調函數**************************
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)
        
        #add_done_callback回調函數,執行一個任務就返回一個函數的返回值,增加回調函數，發佈完任務後直接執行本行，當同一進程執行完第一次任務後，主進程立即對其結果進行分析，不必等所有進程全部執行完再分析，提高了效率
線程池設置4個線程,非同步發起10個任務,每次執行4個任務,執行完任務的時間肯定有先後順序,先執行完的,將parse分析代碼的任務
交給空閑線程去執行,然後這個線程再去執行其他任務,比爬取源碼的任務就和分析數據的任務共同併發執行


非同步回收收取結果的兩種方式?
1.將所有任務的結果同意回收
2.完成一個任務,返回一個結果

非同步+回調函數
回調函數:按順序接收每個任務的結果,進行下一步處理
前提:
    非同步處理的多個任務(io任務),回調函數處理的非io
區別:
    多進程,由主進程執行回調函數的代碼
    多線程,由空閑線程執行回調函數的代碼

您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 彙編筆記_基礎篇
    1 匯流排 內部匯流排：實現CPU內部各器件（運算器、控制器、寄存器等）之間的聯繫。 匯流排（外部匯流排）：實現CPU和主板上其他器件的聯繫。 地址匯流排 CPU通過地址匯流排指定存儲器單元。 一個CPU有N根地址線，則這個CPU的地址匯流排的寬度為N，這個CPU最多可以尋找2的N次方個記憶體單元（位元組）。 控制總 ...
  • 阿裡JAVA實習生入職測試題（2019最新）
    根據我的理解和搜集的資料，儘可能清晰完整的回答（逐步完善，持續更新） 1、String類為什麼是final的 首先分析String的源碼： 類被final關鍵字限定，說明它不可以被繼承，沒有子類。即持有一個String對象的引用，它必然是String類，而不會是其他的類。 value是用來存儲值的， ...
  • 人臉識別Demo
    ★、本實例使用百度智能雲-人工智慧-人臉識別API實現。 ★、樓下安裝了刷臉進門。閑暇時無聊寫了個Demo 主界面顯示如下圖： 本實例，包括了所有人臉識別API的調用。 1、 創建樓號，對應API中創建用戶組，詳見： https://ai.baidu.com/docs#/Face-Set-V3/58 ...
  • python捕捉詳細異常堆棧的方法
    python中有 try——except 的方法捕獲異常，可以獲取到異常的種類以及自定義異常， 但是有時候對於debug測試來說，信息不全，比如說 觸發異常的具體位置在哪： import traceback try: num= int('abc')except Exception: tracebac ...
  • Spring Security 解析(三) —— 個性化認證 以及 RememberMe 實現
    Spring Security 解析(三) —— 個性化認證 以及 RememberMe 實現   在學習Spring Cloud 時，遇到了授權服務oauth 相關內容時，總是一知半解，因此決定先把Spring Security 、Spring Security Oauth2 ...
  • Java開發筆記（一百四十三）FXML佈局的基本格式
    前面介紹了JavaFX的常見控制項用法，雖然JavaFX控制項比起AWT與Swing要好用些，但是一樣通過代碼編寫控制項界面，並沒有提高什麼開發效率。要想瀏覽界面的展示效果，都必須運行測試程式才能觀看，即使只是微調控制項的大小，也得重新運行程式查看效果，顯然既費時又費力。為此JavaFX提供了另一種給界面排 ...
  • Tesseract：簡單的Java光學字元識別
    1.1 介紹 開發具有一定價值的符號是人類特有的特征。對於人們來說識別這些符號和理解圖片上的文字是非常正常的事情。與電腦那樣去抓取文字不同，我們完全是基於視覺的本能去閱讀它們。 另一方面，電腦的工作需要具體的和有組織的內容。它們需要數字化的表示，而不是圖形化的。 有時候，這是不可能的。有時，我們 ...
  • JAVA虛擬機記憶體模型
    概念 （這是我學習過程中的一些總結） JAVA虛擬機記憶體模型 從屬於線程的記憶體區域 JVM的記憶體劃分中，有部分區域是線程私有的，有部分是屬於整個JVM進程；我們將這部分歸為一類。 1.程式計數器（Program Counter Register) 在JVM規範中，每個線程都有自己的程式計數器。這是一 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類