简体中文（ZH-CN）English（EN）繁体中文（ZH-TW）日本語（JA）한국어（KO）Melayu（MS）Français（FR）Deutsch（DE）
首頁
文章
問答
課程
專題
下載
遊戲
詞典
最近更新

首頁  >  文章  >  後端開發  >  Python 中的同步和非同步程式設計：關鍵概念和應用

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

WBOY
WBOY原創
2024-08-31 06:03:37543瀏覽

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("""
    <!DOCTYPE html>
    <html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages');
                var message = document.createElement('li');
                message.appendChild(document.createTextNode(event.data));
                messages.appendChild(message);
            };

            function sendMessage(event) {
                var input = document.getElementById("messageText");
                ws.send(input.value);
                input.value = '';
                event.preventDefault();
            }
        </script>
    </body>
    </html>
    """)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

前端（Streamlit）

  1. 安裝依賴項： 
pip install streamlit websocket-client
  1. 前端程式碼（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)

運行應用程式

  1. 啟動FastAPI後端： 
uvicorn backend:app
  1. 啟動 Streamlit 前端： 
streamlit run frontend.py

說明
後端（backend.py）：

  • FastAPI 應用程式在 /ws 處有一個 WebSocket 端點。
  • ConnectionManager 處理 WebSocket 連接，向所有連接的客戶端廣播訊息。
  • 根端點 (/) 提供一個簡單的 HTML 頁面來測試 WebSocket 連線。

前端（frontend.py）：

  • Streamlit 應用程式連接到 WebSocket 伺服器並偵聽傳入訊息。
  • 一個單獨的線程處理 WebSocket 連接，以防止阻塞 Streamlit 應用程式。
  • 使用者可以透過輸入框傳送訊息，訊息會被傳送到WebSocket伺服器並顯示在聊天中。

此範例示範了一個簡單的即時訊息應用程序，在後端使用 FastAPI 和 WebSockets，在前端使用 Streamlit。

以上是Python 中的同步和非同步程式設計：關鍵概念和應用的詳細內容。更多資訊請關注PHP中文網其他相關文章！

Python flask html fastapi pip if for using Thread number this display input database http websocket
陳述：
本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn
上一篇：推出 Intelliscore：預測足球比賽結果的擴展下一篇：推出 Intelliscore：預測足球比賽結果的擴展

相關文章

看更多