Python 中的同步和非同步程式設計：關鍵概念和應用-Python教學-PHP中文網

首頁

後端開發

Python教學

Python 中的同步和非同步程式設計：關鍵概念和應用

WBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWB

Aug 31, 2024 am 06:03 AM

Synchronous and Asynchronous Programming in Python: Key Concepts and Applications

同步程式設計
在同步編程中，任務是一個接一個地執行的。每項任務必須在下一項任務開始之前完成。這種線性方法很簡單，但效率可能較低，尤其是在處理文件讀取、網路請求或資料庫查詢等 I/O 密集型操作時。

import time

def task1():
    print("Starting task 1...")
    time.sleep(2)
    print("Task 1 completed")

def task2():
    print("Starting task 2...")
    time.sleep(2)
    print("Task 2 completed")

def main():
    task1()
    task2()

if __name__ == "__main__":
    main()

在此範例中，任務 1 必須在任務 2 開始之前完成。總執行時間是每個任務所花費時間的總和。

非同步程式設計
非同步編程允許多個任務同時運行，從而提高效率，特別是對於 I/O 密集型任務。 Python 的 asyncio 函式庫提供了非同步程式設計所需的工具。

import asyncio

async def task1():
    print("Starting task 1...")
    await asyncio.sleep(2)
    print("Task 1 completed")

async def task2():
    print("Starting task 2...")
    await asyncio.sleep(2)
    print("Task 2 completed")

async def main():
    await asyncio.gather(task1(), task2())

if __name__ == "__main__":
    asyncio.run(main())

在這個例子中，task1和task2並發運行，將總執行時間減少到最長任務所花費的時間。

潛在應用

Web 伺服器和 API：

同步：像 Flask 這樣的傳統 Web 框架會依序處理請求。在處理大量請求時，這可能會成為瓶頸。
非同步：FastAPI 和 aiohttp 等框架使用非同步程式設計來同時處理多個請求，從而提高吞吐量和效能。

即時訊息應用程式：

同步：如果按順序處理每個訊息，處理即時訊息可能會導致延遲。
非同步：使用具有非同步處理功能的 WebSocket（例如 websockets 庫）可以實現即時雙向通信，從而實現高效能聊天應用程式、即時通知等。

資料處理管道：

同步：依序處理大型資料集可能非常耗時。
非同步：非同步任務可以同時取得、處理和儲存數據，顯著減少處理時間。像 aiohttp 和 aiomysql 這樣的函式庫可用於非同步 HTTP 請求和資料庫操作。

網頁抓取：

同步：順序取得網頁可能會很慢且效率低。
非同步：使用aiohttp進行非同步HTTP請求可以同時取得多個網頁，並加快網頁抓取流程。

文件I/O操作：

同步：順序讀取/寫入大檔案可能會阻塞其他操作。
非同步：使用 aiofile 的非同步檔案 I/O 操作可以透過允許其他任務同時運行來提高效能。

同步與非同步之間的選擇

對 CPU 密集型任務使用同步編程，這些任務是計算密集型的，並且可以從順序運行中受益。
對 I/O 密集型任務使用非同步編程，其中操作涉及等待外部資源，例如網路請求、檔案 I/O 或資料庫查詢。

即時訊息應用範例
讓我們創建一個基本的即時訊息應用程序，使用 FastAPI 作為後端，使用 WebSockets 進行即時通訊。我們將使用 Streamlit 作為前端來顯示訊息。

後端（FastAPI + WebSockets）

1.安裝相依性：
pip install fastapi uvicorn websockets

2.後端程式碼(backend.py):

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_message(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_message(data)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

@app.get("/")
async def get():
    return HTMLResponse("""
    
    
    
        <title>Chat</title>
    
    
        <h1 id="WebSocket-Chat">WebSocket Chat</h1>

前端（Streamlit）

安裝依賴項：

pip install streamlit websocket-client

前端程式碼（frontend.py）：

import streamlit as st
import asyncio
import threading
from websocket import create_connection, WebSocket

st.title("Real-time Messaging Application")

if 'messages' not in st.session_state:
    st.session_state.messages = []

def websocket_thread():
    ws = create_connection("ws://localhost:8000/ws")
    st.session_state.ws = ws
    while True:
        message = ws.recv()
        st.session_state.messages.append(message)
        st.experimental_rerun()

if 'ws' not in st.session_state:
    threading.Thread(target=websocket_thread, daemon=True).start()

input_message = st.text_input("Enter your message:")

if st.button("Send"):
    if input_message:
        st.session_state.ws.send(input_message)
        st.session_state.messages.append(f"You: {input_message}")

st.subheader("Chat Messages:")
for message in st.session_state.messages:
    st.write(message)

運行應用程式