Python網絡編程的 11 個重要知識點
作者：小白PythonAI編程
網絡編程就是讓程序通過網絡發送數據給其他程序或接收其他程序的數據。Python中的網絡編程主要使用 socket 模塊。
1. 網絡編程基礎
網絡編程就是讓程序通過網絡發送數據給其他程序或接收其他程序的數據。Python中的網絡編程主要使用 socket 模塊。
2. TCP服務器示例
import socket
# 創建 socket 對象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 綁定端口
host = 'localhost'
port = 12345
server_socket.bind((host, port))
# 監聽連接
server_socket.listen(5)
print('Server listening on port:', port)
while True:
# 建立客戶端連接
client_socket, addr = server_socket.accept()
print('Got connection from', addr)
# 接收客戶端消息
msg = client_socket.recv(1024).decode()
print('Message from client:', msg)
# 發送響應
response = 'Thank you for connecting'
client_socket.send(response.encode())
# 關閉連接
client_socket.close()這個簡單的服務器監聽12345端口，當有客戶端連接時，會打印客戶端地址，并接收客戶端的消息，然后發送響應并關閉連接。
3. TCP客戶端示例
import socket
# 創建 socket 對象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 連接服務器
host = 'localhost'
port = 12345
client_socket.connect((host, port))
# 發送消息
msg = 'Hello, server!'
client_socket.send(msg.encode())
# 接收響應
response = client_socket.recv(1024).decode()
print('Response from server:', response)
# 關閉連接
client_socket.close()這個簡單的客戶端連接服務器，發送一條消息，接收服務器的響應，并關閉連接。
4. UDP服務器示例
import socket
# 創建 socket 對象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 綁定端口
host = 'localhost'
port = 12345
server_socket.bind((host, port))
print('Server listening on port:', port)
while True:
# 接收客戶端消息
msg, addr = server_socket.recvfrom(1024)
print('Message from client:', msg.decode(), 'at', addr)
# 發送響應
response = 'Thank you for your message'
server_socket.sendto(response.encode(), addr)這個簡單的UDP服務器監聽12345端口，接收客戶端的消息，然后發送響應。
5. UDP客戶端示例
import socket
# 創建 socket 對象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 發送消息
host = 'localhost'
port = 12345
msg = 'Hello, server!'
client_socket.sendto(msg.encode(), (host, port))
# 接收響應
response, addr = client_socket.recvfrom(1024)
print('Response from server:', response.decode())
# 關閉連接
client_socket.close()這個簡單的UDP客戶端發送一條消息，接收服務器的響應，并關閉連接。
6. 多線程TCP服務器示例
import socket
import threading
def handle_client(client_socket, addr):
print('Got connection from', addr)
msg = client_socket.recv(1024).decode()
print('Message from client:', msg)
response = 'Thank you for connecting'
client_socket.send(response.encode())
client_socket.close()
# 創建 socket 對象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 綁定端口
host = 'localhost'
port = 12345
server_socket.bind((host, port))
# 監聽連接
server_socket.listen(5)
print('Server listening on port:', port)
while True:
# 建立客戶端連接
client_socket, addr = server_socket.accept()
# 在新線程中處理客戶端連接
thread = threading.Thread(target=handle_client, args=(client_socket, addr))
thread.start()這個服務器使用多線程處理多個客戶端連接，每個客戶端連接都在一個新線程中處理。
7. 非阻塞I/O TCP服務器示例
import socket
import select
# 創建 socket 對象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 設置為非阻塞模式
server_socket.setblocking(False)
# 綁定端口
host = 'localhost'
port = 12345
server_socket.bind((host, port))
# 監聽連接
server_socket.listen(5)
print('Server listening on port:', port)
inputs = [server_socket]
outputs = []
while True:
readable, writable, exceptional = select.select(inputs, outputs, inputs)
for sock in readable:
if sock == server_socket:
# 建立客戶端連接
client_socket, addr = server_socket.accept()
client_socket.setblocking(False)
inputs.append(client_socket)
print('Got connection from', addr)
else:
# 接收客戶端消息
data = sock.recv(1024)
if data:
print('Message from client:', data.decode())
sock.send(data.upper())
else:
# 客戶端斷開連接
print('Client disconnected')
inputs.remove(sock)
sock.close()這個非阻塞TCP服務器使用select模塊同時處理多個客戶端連接，提高了程序的響應速度。
8. 使用HTTP協議示例
from http.server import HTTPServer, BaseHTTPRequestHandler
class SimpleHTTPRequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.end_headers()
message = 'Hello, World!'
self.wfile.write(message.encode())
# 創建 HTTP 服務器
server_address = ('localhost', 8000)
httpd = HTTPServer(server_address, SimpleHTTPRequestHandler)
print('Starting simple HTTP server...')
httpd.serve_forever()這個簡單的HTTP服務器監聽8000端口，當收到GET請求時，返回“Hello, World!”的響應。
9. 發送HTTP請求示例
首先安裝requests庫：
pip install requests然后編寫代碼：
import requests
url = 'http://localhost:8000'
response = requests.get(url)
print('Response status code:', response.status_code)
print('Response content:', response.text)這段代碼向本地HTTP服務器發送GET請求，并打印響應的狀態碼和內容。
10. WebSocket編程示例
首先安裝websockets庫：
pip install websockets然后編寫代碼：
import asyncio
import websockets
async def echo(websocket, path):
async for message in websocket:
print(f'Received message: {message}')
await websocket.send(message)
# 創建 WebSocket 服務器
start_server = websockets.serve(echo, 'localhost', 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()這個簡單的WebSocket服務器監聽8765端口，當收到消息時，將其原樣返回。
11. 使用WebSocket客戶端示例
import asyncio
import websockets
async def send_message():
uri = 'ws://localhost:8765'
async with websockets.connect(uri) as websocket:
message = 'Hello, WebSocket!'
await websocket.send(message)
print(f'Sent message: {message}')
response = await websocket.recv()
print(f'Received response: {response}')
asyncio.get_event_loop().run_until_complete(send_message())這個簡單的WebSocket客戶端連接服務器，發送一條消息，并接收服務器的響應。
12. 實戰案例：實時聊天應用
接下來，我們將創建一個簡單的實時聊天應用，包括一個WebSocket服務器和多個客戶端。
(1) 創建WebSocket服務器
import asyncio
import websockets
connected_clients = set()
async def broadcast(message):
if connected_clients:
await asyncio.wait([client.send(message) for client in connected_clients])
async def chat(websocket, path):
connected_clients.add(websocket)
try:
async for message in websocket:
print(f'Received message: {message}')
await broadcast(message)
finally:
connected_clients.remove(websocket)
# 創建 WebSocket 服務器
start_server = websockets.serve(chat, 'localhost', 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()這個WebSocket服務器監聽8765端口，并將所有客戶端的消息廣播給其他客戶端。
(2) 創建WebSocket客戶端
import asyncio
import websockets
async def chat_client():
uri = 'ws://localhost:8765'
async with websockets.connect(uri) as websocket:
while True:
message = input('Enter your message: ')
await websocket.send(message)
print('Sent message:', message)
response = await websocket.recv()
print('Received response:', response)
asyncio.get_event_loop().run_until_complete(chat_client())這個WebSocket客戶端連接服務器，發送消息，并接收服務器的廣播消息。
責任編輯：趙寧寧
來源：
小白PythonAI編程
