Python網(wǎng)絡(luò)編程的 11 個(gè)重要知識(shí)點(diǎn)
作者：小白PythonAI編程
網(wǎng)絡(luò)編程就是讓程序通過(guò)網(wǎng)絡(luò)發(fā)送數(shù)據(jù)給其他程序或接收其他程序的數(shù)據(jù)。Python中的網(wǎng)絡(luò)編程主要使用 socket 模塊。
1. 網(wǎng)絡(luò)編程基礎(chǔ)
網(wǎng)絡(luò)編程就是讓程序通過(guò)網(wǎng)絡(luò)發(fā)送數(shù)據(jù)給其他程序或接收其他程序的數(shù)據(jù)。Python中的網(wǎng)絡(luò)編程主要使用 socket 模塊。
2. TCP服務(wù)器示例
import socket
# 創(chuàng)建 socket 對(duì)象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 綁定端口
host = 'localhost'
port = 12345
server_socket.bind((host, port))
# 監(jiān)聽(tīng)連接
server_socket.listen(5)
print('Server listening on port:', port)
while True:
# 建立客戶(hù)端連接
client_socket, addr = server_socket.accept()
print('Got connection from', addr)
# 接收客戶(hù)端消息
msg = client_socket.recv(1024).decode()
print('Message from client:', msg)
# 發(fā)送響應(yīng)
response = 'Thank you for connecting'
client_socket.send(response.encode())
# 關(guān)閉連接
client_socket.close()這個(gè)簡(jiǎn)單的服務(wù)器監(jiān)聽(tīng)12345端口，當(dāng)有客戶(hù)端連接時(shí)，會(huì)打印客戶(hù)端地址，并接收客戶(hù)端的消息，然后發(fā)送響應(yīng)并關(guān)閉連接。
3. TCP客戶(hù)端示例
import socket
# 創(chuàng)建 socket 對(duì)象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 連接服務(wù)器
host = 'localhost'
port = 12345
client_socket.connect((host, port))
# 發(fā)送消息
msg = 'Hello, server!'
client_socket.send(msg.encode())
# 接收響應(yīng)
response = client_socket.recv(1024).decode()
print('Response from server:', response)
# 關(guān)閉連接
client_socket.close()這個(gè)簡(jiǎn)單的客戶(hù)端連接服務(wù)器，發(fā)送一條消息，接收服務(wù)器的響應(yīng)，并關(guān)閉連接。
4. UDP服務(wù)器示例
import socket
# 創(chuàng)建 socket 對(duì)象
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:
# 接收客戶(hù)端消息
msg, addr = server_socket.recvfrom(1024)
print('Message from client:', msg.decode(), 'at', addr)
# 發(fā)送響應(yīng)
response = 'Thank you for your message'
server_socket.sendto(response.encode(), addr)這個(gè)簡(jiǎn)單的UDP服務(wù)器監(jiān)聽(tīng)12345端口，接收客戶(hù)端的消息，然后發(fā)送響應(yīng)。
5. UDP客戶(hù)端示例
import socket
# 創(chuàng)建 socket 對(duì)象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 發(fā)送消息
host = 'localhost'
port = 12345
msg = 'Hello, server!'
client_socket.sendto(msg.encode(), (host, port))
# 接收響應(yīng)
response, addr = client_socket.recvfrom(1024)
print('Response from server:', response.decode())
# 關(guān)閉連接
client_socket.close()這個(gè)簡(jiǎn)單的UDP客戶(hù)端發(fā)送一條消息，接收服務(wù)器的響應(yīng)，并關(guān)閉連接。
6. 多線(xiàn)程TCP服務(wù)器示例
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()
# 創(chuàng)建 socket 對(duì)象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 綁定端口
host = 'localhost'
port = 12345
server_socket.bind((host, port))
# 監(jiān)聽(tīng)連接
server_socket.listen(5)
print('Server listening on port:', port)
while True:
# 建立客戶(hù)端連接
client_socket, addr = server_socket.accept()
# 在新線(xiàn)程中處理客戶(hù)端連接
thread = threading.Thread(target=handle_client, args=(client_socket, addr))
thread.start()這個(gè)服務(wù)器使用多線(xiàn)程處理多個(gè)客戶(hù)端連接，每個(gè)客戶(hù)端連接都在一個(gè)新線(xiàn)程中處理。
7. 非阻塞I/O TCP服務(wù)器示例
import socket
import select
# 創(chuàng)建 socket 對(duì)象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 設(shè)置為非阻塞模式
server_socket.setblocking(False)
# 綁定端口
host = 'localhost'
port = 12345
server_socket.bind((host, port))
# 監(jiān)聽(tīng)連接
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:
# 建立客戶(hù)端連接
client_socket, addr = server_socket.accept()
client_socket.setblocking(False)
inputs.append(client_socket)
print('Got connection from', addr)
else:
# 接收客戶(hù)端消息
data = sock.recv(1024)
if data:
print('Message from client:', data.decode())
sock.send(data.upper())
else:
# 客戶(hù)端斷開(kāi)連接
print('Client disconnected')
inputs.remove(sock)
sock.close()這個(gè)非阻塞TCP服務(wù)器使用select模塊同時(shí)處理多個(gè)客戶(hù)端連接，提高了程序的響應(yīng)速度。
8. 使用HTTP協(xié)議示例
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())
# 創(chuàng)建 HTTP 服務(wù)器
server_address = ('localhost', 8000)
httpd = HTTPServer(server_address, SimpleHTTPRequestHandler)
print('Starting simple HTTP server...')
httpd.serve_forever()這個(gè)簡(jiǎn)單的HTTP服務(wù)器監(jiān)聽(tīng)8000端口，當(dāng)收到GET請(qǐng)求時(shí)，返回“Hello, World!”的響應(yīng)。
9. 發(fā)送HTTP請(qǐng)求示例
首先安裝requests庫(kù)：
pip install requests然后編寫(xiě)代碼：
import requests
url = 'http://localhost:8000'
response = requests.get(url)
print('Response status code:', response.status_code)
print('Response content:', response.text)這段代碼向本地HTTP服務(wù)器發(fā)送GET請(qǐng)求，并打印響應(yīng)的狀態(tài)碼和內(nèi)容。
10. WebSocket編程示例
首先安裝websockets庫(kù)：
pip install websockets然后編寫(xiě)代碼：
import asyncio
import websockets
async def echo(websocket, path):
async for message in websocket:
print(f'Received message: {message}')
await websocket.send(message)
# 創(chuàng)建 WebSocket 服務(wù)器
start_server = websockets.serve(echo, 'localhost', 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()這個(gè)簡(jiǎn)單的WebSocket服務(wù)器監(jiān)聽(tīng)8765端口，當(dāng)收到消息時(shí)，將其原樣返回。
11. 使用WebSocket客戶(hù)端示例
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())這個(gè)簡(jiǎn)單的WebSocket客戶(hù)端連接服務(wù)器，發(fā)送一條消息，并接收服務(wù)器的響應(yīng)。
12. 實(shí)戰(zhàn)案例：實(shí)時(shí)聊天應(yīng)用
接下來(lái)，我們將創(chuàng)建一個(gè)簡(jiǎn)單的實(shí)時(shí)聊天應(yīng)用，包括一個(gè)WebSocket服務(wù)器和多個(gè)客戶(hù)端。
(1) 創(chuàng)建WebSocket服務(wù)器
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)
# 創(chuàng)建 WebSocket 服務(wù)器
start_server = websockets.serve(chat, 'localhost', 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()這個(gè)WebSocket服務(wù)器監(jiān)聽(tīng)8765端口，并將所有客戶(hù)端的消息廣播給其他客戶(hù)端。
(2) 創(chuàng)建WebSocket客戶(hù)端
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())這個(gè)WebSocket客戶(hù)端連接服務(wù)器，發(fā)送消息，并接收服務(wù)器的廣播消息。
責(zé)任編輯：趙寧寧
來(lái)源：
小白PythonAI編程
