同步程式設計
在同步編程中，任務是一個接一個地執行的。每項任務必須在下一項任務開始之前完成。這種線性方法很簡單，但效率可能較低，尤其是在處理文件讀取、網路請求或資料庫查詢等 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並發運行，將總執行時間減少到最長任務所花費的時間。
潛在應用
同步與非同步之間的選擇
即時訊息應用範例
讓我們創建一個基本的即時訊息應用程序，使用 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）
pip install streamlit websocket-client
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)
運行應用程式
uvicorn backend:app
streamlit run frontend.py
說明
後端（backend.py）：
前端（frontend.py）：
此範例示範了一個簡單的即時訊息應用程序，在後端使用 FastAPI 和 WebSockets，在前端使用 Streamlit。
以上是Python 中的同步和非同步程式設計：關鍵概念和應用的詳細內容。更多資訊請關注PHP中文網其他相關文章！
