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
: 套接字類型(SOCK_STREAM
為TCP,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)聽,
backlog
指定最大排隊(duì)連接數(shù) - accept(): 阻塞等待客戶端連接,返回
(conn, address)
元組
3. 客戶端方法
- 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:多線程聊天室(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() # 客戶端實(shí)現(xiàn)需配合使用連接和發(fā)送/接收線程
應(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)避免端口占用問題 - 粘包處理:TCP是流式協(xié)議,需自定義消息邊界(如長(zhǎng)度前綴)
- 并發(fā)模型:
- 多線程:適合簡(jiǎn)單并發(fā)場(chǎng)景
- select:適合I/O多路復(fù)用
- asyncio:適合高并發(fā)異步處理
- 異常處理:必須處理
ConnectionResetError
等網(wǎng)絡(luò)異常 - 數(shù)據(jù)編碼:網(wǎng)絡(luò)傳輸使用bytes類型,需注意編解碼
以上代碼示例展示了socket編程在不同場(chǎng)景下的典型應(yīng)用,實(shí)際開發(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)文章
淺談opencv自動(dòng)光學(xué)檢測(cè)、目標(biāo)分割和檢測(cè)(連通區(qū)域和findContours)
這篇文章主要介紹了淺談opencv自動(dòng)光學(xué)檢測(cè)、目標(biāo)分割和檢測(cè)(連通區(qū)域和findContours),具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2020-06-06Python format補(bǔ)0的實(shí)現(xiàn)方法
對(duì)于一些數(shù)字的處理,我們可能需要讓它們滿足一定格式的要求,本文主要介紹了Python format補(bǔ)0的實(shí)現(xiàn)方法,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-07-07Python 調(diào)用 Outlook 發(fā)送郵件過程解析
這篇文章主要介紹了Python 調(diào)用 Outlook 發(fā)送郵件過程解析,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-08-08基于Python實(shí)現(xiàn)PDF區(qū)域文本提取工具
這篇文章主要為大家介紹了如何通過Python實(shí)現(xiàn)一個(gè)非常精簡(jiǎn)的圖像化的PDF區(qū)域選擇提取工具,文中示例代碼講解詳細(xì),感興趣的小伙伴可以學(xué)習(xí)一下2021-12-12使用Python編寫Prometheus監(jiān)控的方法
今天小編就為大家分享一篇關(guān)于使用Python編寫Prometheus監(jiān)控的方法,小編覺得內(nèi)容挺不錯(cuò)的,現(xiàn)在分享給大家,具有很好的參考價(jià)值,需要的朋友一起跟隨小編來看看吧2018-10-10