Python Socket庫(kù)基礎(chǔ)方法與應(yīng)用詳解
一、Socket基礎(chǔ)方法詳解
Python的socket模塊提供了BSD Socket API接口,以下是核心方法:
1. 構(gòu)造函數(shù)
socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0)
- 參數(shù):
family: 地址族(常用AF_INET表示IPv4)type: 套接字類(lèi)型(SOCK_STREAM為T(mén)CP,SOCK_DGRAM為UDP)proto: 協(xié)議號(hào)(通常為0,自動(dòng)選擇)
2. 服務(wù)器端方法
- bind(address): 綁定IP地址和端口號(hào)(元組格式
(host, port)) - listen(backlog): 啟動(dòng)TCP監(jiān)聽(tīng),
backlog指定最大排隊(duì)連接數(shù) - accept(): 阻塞等待客戶(hù)端連接,返回
(conn, address)元組
3. 客戶(hù)端方法
- connect(address): 主動(dòng)連接服務(wù)器
4. 數(shù)據(jù)傳輸
- send(bytes): 發(fā)送TCP數(shù)據(jù)(可能未發(fā)送全部數(shù)據(jù))
- sendall(bytes): 發(fā)送全部TCP數(shù)據(jù)(推薦使用)
- recv(bufsize): 接收TCP數(shù)據(jù)(最大
bufsize字節(jié)) - sendto(bytes, address): 發(fā)送UDP數(shù)據(jù)
- recvfrom(bufsize): 接收UDP數(shù)據(jù),返回
(data, address)
5. 通用方法
- close(): 關(guān)閉套接字
- setsockopt(level, optname, value): 設(shè)置套接字選項(xiàng)(如
SO_REUSEADDR)
二、工作原理與實(shí)現(xiàn)機(jī)制
1. TCP通信流程


2. UDP通信特點(diǎn)
- 無(wú)連接協(xié)議
- 數(shù)據(jù)包可能丟失或亂序
- 適合實(shí)時(shí)性要求高的場(chǎng)景
3. 三次握手(TCP)
- SYN →
- ← SYN-ACK
- ACK →
三、應(yīng)用領(lǐng)域與實(shí)現(xiàn)代碼
應(yīng)用1:基礎(chǔ)HTTP服務(wù)器(TCP)
import socket
def start_web_server(host='', port=8080):
with socket.socket() as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((host, port))
s.listen(5)
print(f"Server running on {port}")
while True:
conn, addr = s.accept()
with conn:
request = conn.recv(1024).decode()
# 解析HTTP請(qǐng)求頭
headers = request.split('\n')
filename = headers[0].split()[1][1:] or 'index.html'
try:
with open(filename, 'rb') as f:
content = f.read()
response = b'HTTP/1.1 200 OK\n\n' + content
except FileNotFoundError:
response = b'HTTP/1.1 404 Not Found\n\n<h1>404 Error</h1>'
conn.sendall(response)
# 使用示例
start_web_server()
應(yīng)用2:多線(xiàn)程聊天室(TCP)
import socket
import threading
clients = []
def handle_client(client):
while True:
try:
msg = client.recv(1024)
if not msg:
break
for c in clients:
if c != client:
c.sendall(msg)
except:
break
clients.remove(client)
client.close()
def start_chat_server(port=9000):
with socket.socket() as s:
s.bind(('', port))
s.listen()
print(f"Chat server started on {port}")
while True:
client, addr = s.accept()
clients.append(client)
thread = threading.Thread(target=handle_client, args=(client,))
thread.start()
# 客戶(hù)端實(shí)現(xiàn)需配合使用連接和發(fā)送/接收線(xiàn)程
應(yīng)用3:文件傳輸(TCP)
import socket
import hashlib
def send_file(filename, host='127.0.0.1', port=6000):
with socket.socket() as s:
s.connect((host, port))
with open(filename, 'rb') as f:
file_data = f.read()
# 計(jì)算文件哈希值
file_hash = hashlib.md5(file_data).hexdigest().encode()
# 先發(fā)送哈希值
s.sendall(file_hash)
# 發(fā)送文件數(shù)據(jù)
s.sendall(file_data)
print("File sent successfully")
def receive_file(port=6000):
with socket.socket() as s:
s.bind(('', port))
s.listen()
conn, addr = s.accept()
with conn:
# 接收哈希值
file_hash = conn.recv(32)
file_data = b''
while True:
data = conn.recv(4096)
if not data:
break
file_data += data
# 驗(yàn)證完整性
if hashlib.md5(file_data).hexdigest().encode() == file_hash:
with open('received_file', 'wb') as f:
f.write(file_data)
print("File received successfully")
else:
print("File corrupted")
應(yīng)用4:端口掃描工具(TCP SYN)
import socket
from concurrent.futures import ThreadPoolExecutor
def scan_port(host, port):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(1)
try:
s.connect((host, port))
print(f"Port {port} is open")
return True
except:
return False
def port_scanner(host='127.0.0.1', start=1, end=1024):
print(f"Scanning {host}...")
with ThreadPoolExecutor(max_workers=100) as executor:
results = executor.map(lambda p: scan_port(host, p), range(start, end+1))
return sum(results)
# 使用示例
port_scanner(end=100)
四、關(guān)鍵技術(shù)要點(diǎn)
- 地址重用:使用
SO_REUSEADDR選項(xiàng)避免端口占用問(wèn)題 - 粘包處理:TCP是流式協(xié)議,需自定義消息邊界(如長(zhǎng)度前綴)
- 并發(fā)模型:
- 多線(xiàn)程:適合簡(jiǎn)單并發(fā)場(chǎng)景
- select:適合I/O多路復(fù)用
- asyncio:適合高并發(fā)異步處理
- 異常處理:必須處理
ConnectionResetError等網(wǎng)絡(luò)異常 - 數(shù)據(jù)編碼:網(wǎng)絡(luò)傳輸使用bytes類(lèi)型,需注意編解碼
以上代碼示例展示了socket編程在不同場(chǎng)景下的典型應(yīng)用,實(shí)際開(kāi)發(fā)中需要根據(jù)具體需求添加錯(cuò)誤處理、日志記錄和安全機(jī)制。
到此這篇關(guān)于Python Socket庫(kù)基礎(chǔ)方法與應(yīng)用詳解的文章就介紹到這了,更多相關(guān)Python Socket庫(kù)方法與應(yīng)用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
如何基于Python獲取SonarQube的檢查報(bào)告信息詳解
SonarQube 是一種流行的開(kāi)源平臺(tái),用于持續(xù)檢查代碼質(zhì)量問(wèn)題,它支持多種編程語(yǔ)言,包括 Python,這篇文章主要介紹了如何基于Python獲取SonarQube的檢查報(bào)告信息的相關(guān)資料,需要的朋友可以參考下2026-06-06
Python+Selenium使用Page Object實(shí)現(xiàn)頁(yè)面自動(dòng)化測(cè)試
這篇文章主要介紹了Python+Selenium使用Page Object實(shí)現(xiàn)頁(yè)面自動(dòng)化測(cè)試,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-07-07
Python基礎(chǔ)之變量的相關(guān)知識(shí)總結(jié)
今天給大家?guī)?lái)的是關(guān)于Python的相關(guān)知識(shí),文章圍繞著Python變量展開(kāi),文中有非常詳細(xì)的介紹及代碼示例,需要的朋友可以參考下2021-06-06
解決python使用list()時(shí)總是報(bào)錯(cuò)的問(wèn)題
這篇文章主要介紹了解決python使用list()時(shí)總是報(bào)錯(cuò)的問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-05-05
Python學(xué)習(xí)筆記_數(shù)據(jù)排序方法
Python對(duì)數(shù)據(jù)排序有兩種方法:下面我們來(lái)簡(jiǎn)單分析下2014-05-05
Python獲取當(dāng)前函數(shù)名稱(chēng)方法實(shí)例分享
這篇文章主要介紹了Python獲取當(dāng)前函數(shù)名稱(chēng)方法實(shí)例分享,具有一定借鑒價(jià)值2018-01-01

