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

来源:https://www.cnblogs.com/mylu/archive/2019/07/25/11247125.html
-Advertisement-
Play Games

9.11 進程池與線程池 池子使用來限制併發的任務數目，限制我們的電腦在一個自己可承受的範圍內去併發地執行任務 池子內什麼時候裝進程：併發的任務屬於計算密集型 池子內什麼時候裝線程：併發的任務屬於IO密集型 進程池： 線程池： 9.112 基於多線程實現併發的套接字通信(使用線程池) 服務端： f ...


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()
View Code

客戶端：

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()
View Code

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
View Code

基於線程池：

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
View Code

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')

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

-Advertisement-
Play Games
更多相關文章
  • 整合SSM框架必備基礎—SpringMVC（下）
    在上一篇文章 "《整合SSM框架必備基礎—SpringMVC（上）》" 中，胖達介紹了關於SpringMVC的誕生、優勢以及執行流程等理論知識點，這篇文章打算在實操中加深一下對SpringMVC的認識，畢竟實踐才是學習技術最有效的方法嘛，Let's Go! 一、 首先來創建一個Web小項目吧 JDK ...
  • 2019年7月訓練（壹）
    2019-07-25 luogu P3627 [APIO2009]搶掠計劃 卡了三個小時，看了題解才作出來的（菜） 前驅知識： 壹~鄰接表存儲/遍歷 貳~SPFA跑最長路（<改>就行了） 叄~Tarjan縮點 壹.鄰接表儲存 兩個，add存無邊權，未縮點；build有邊權，已縮點。 貳.輸入 懶得開 ...
  • 淺析java設計模式(一)----異構容器,可以存儲任何對象類型為其他類提供該對象
    java類型異構容器,可以存儲任何對象類型為其他類提供該對象 ...
  • 從零探索Java網路編程01之 TCP/IP 與 Socket
    最近完成了幾項比較簡單的項目, 日子有些鬆散, 終於是在996里偷了點閑暇時光, 想著來研究研究些啥吧? 一個普通的控制台日誌映入了我的眼帘(孽緣呀): (圖中使用 SpringBoot 的 log4j 來輸出日誌, logginglevel: debug, jdk版本為1.8) 造成這種現象的原因 ...
  • 切片（Slice）
    Python提供了切片（Slice）操作符：可以一次取出多個列表元素 L[0:3]表示，從索引0開始取，直到索引3為止，但不包括索引3。0可以省略：L[:3] L[:]:就是整個列表 補充： 前10個數，每兩個取一個： >>> L[:10:2] [0, 2, 4, 6, 8] s[:2:-1]表示從 ...
  • mysql（一）隨筆
    DDL、DML、DCL、DQL的簡單操作，從溫習中學到之前沒有體會到的一些知識，在忙碌中偷得一日閑。 ...
  • Java編程基礎階段筆記 day01 Java語言概述
    ​目錄內容 DOS命令 電腦配置 Java語言的特性 Java兩種核心機制 Java語言環境搭建 第一個Java程式 註釋 Java語句說明 編程風格 作業 DOS命令 電腦配置 Java語言的特性 Java兩種核心機制 Java語言環境搭建 第一個Java程式 註釋 Java語句說明 編程風格 作 ...
  • Fortran文件讀寫--查找內容
    program ex implicit none character(len=40) A(3000),B(3000),C(3000) !A異常、B已開挖、C需標記 integer i,j,N1,N2,count !N1是10號文件行數,N2是11號文件行數，count是計數器 open(unit=1... ...
一周排行
    -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.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類