基于Python的socket庫(kù)實(shí)現(xiàn)通信功能的示例代碼
1.前言
這是一篇介紹使用python的socket庫(kù)實(shí)現(xiàn)通信功能的文章.系統(tǒng)主要分為客戶端和服務(wù)端。這里簡(jiǎn)單的給每個(gè)客戶端增加一個(gè)不重復(fù)的uid,客戶端之間可以根據(jù)這個(gè)uid選擇進(jìn)行廣播通信,即與所有在線的客戶端通信。也可以選擇與單一客戶端通信。
2.技術(shù)介紹
(1) socket:
Socket(套接字)是在應(yīng)用層和傳輸層之間的一個(gè)抽象層,它提供了對(duì)TCP/IP協(xié)議族進(jìn)行訪問(wèn)的接口。socket可以看成是一種特殊的文件描述符,與文件描述符類(lèi)似,開(kāi)發(fā)者可以對(duì)它進(jìn)行讀寫(xiě)操作,從而實(shí)現(xiàn)與遠(yuǎn)程主機(jī)的通信和數(shù)據(jù)傳輸。
在網(wǎng)絡(luò)編程中,Socket通常被用于實(shí)現(xiàn)客戶端和服務(wù)器之間的通信。當(dāng)一個(gè)程序需要向另一個(gè)程序發(fā)送數(shù)據(jù)時(shí),它會(huì)將數(shù)據(jù)包裝在一個(gè)Socket對(duì)象中,然后通過(guò)網(wǎng)絡(luò)傳輸給對(duì)方。接收端程序收到數(shù)據(jù)后,會(huì)再次將數(shù)據(jù)解包,存入自己的Socket緩存當(dāng)中。
在TCP/IP協(xié)議中,Socket是通過(guò)IP地址、端口號(hào)和協(xié)議類(lèi)型來(lái)標(biāo)識(shí)的。其中,IP地址用來(lái)唯一標(biāo)識(shí)一臺(tái)主機(jī),端口號(hào)則表示該主機(jī)上具體的網(wǎng)絡(luò)應(yīng)用程序,協(xié)議類(lèi)型指明該Socket使用的具體傳輸協(xié)議,如TCP或UDP等。
Socket編程比較靈活,可以根據(jù)不同的需求進(jìn)行不同的設(shè)置和配置,例如設(shè)置Socket的超時(shí)時(shí)間、緩沖區(qū)大小、數(shù)據(jù)包大小等。同時(shí),Socket也支持多種編程語(yǔ)言,如Python、Java和C/C++等。因此,在實(shí)現(xiàn)網(wǎng)絡(luò)應(yīng)用程序時(shí),Socket是非常常用和重要的組件之一。
(2) Python的socket庫(kù):
Python的socket庫(kù)是一個(gè)用于網(wǎng)絡(luò)編程的標(biāo)準(zhǔn)庫(kù),它提供了一套高級(jí)的網(wǎng)絡(luò)接口,使得開(kāi)發(fā)人員可以輕松地創(chuàng)建網(wǎng)絡(luò)應(yīng)用程序。
該庫(kù)允許Python程序通過(guò)創(chuàng)建Socket對(duì)象來(lái)完成網(wǎng)絡(luò)通信,每個(gè)Socket對(duì)象都與一個(gè)特定的協(xié)議族相關(guān)聯(lián),如AF_INET(使用IPv4協(xié)議)或AF_INET6(使用IPv6協(xié)議)。
socket庫(kù)提供了一系列函數(shù),可以用于創(chuàng)建、連接、監(jiān)聽(tīng)和發(fā)送數(shù)據(jù)等各種網(wǎng)絡(luò)操作。其中,一些常用的函數(shù)包括:
- socket() : 創(chuàng)建一個(gè)Socket對(duì)象,指定協(xié)議族、sock類(lèi)型、協(xié)議等參數(shù);
- bind() : 將Socket對(duì)象與本地地址進(jìn)行綁定;
- listen() : 開(kāi)始監(jiān)聽(tīng)指定端口上的連接請(qǐng)求;
- accept() : 接受客戶端連接請(qǐng)求,并返回新的Socket對(duì)象和客戶端地址;
- connect() : 連接到遠(yuǎn)端主機(jī)的指定端口;
- recv() 和 send() : 接收和發(fā)送數(shù)據(jù);
- close() : 關(guān)閉Socket連接。
使用Python的socket庫(kù),可以方便地實(shí)現(xiàn)基于TCP或UDP協(xié)議的網(wǎng)絡(luò)編程。同時(shí),socket庫(kù)也支持基于多線程和異步IO的網(wǎng)絡(luò)編程,可以用于處理高并發(fā)的網(wǎng)絡(luò)應(yīng)用程序。
3.系統(tǒng)實(shí)現(xiàn)
(1) 服務(wù)端(server.py)
import socket import threading import json HOST = '127.0.0.1' PORT = 8888 # 存儲(chǔ)客戶端的 uid 和 socket 對(duì)象的映射關(guān)系 clients = {} # 創(chuàng)建一個(gè) TCP 協(xié)議的套接字對(duì)象 server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 綁定并監(jiān)聽(tīng)一個(gè)指定的地址和端口號(hào) server_socket.bind((HOST, PORT)) # 開(kāi)始監(jiān)聽(tīng)客戶端對(duì)指定 IP 地址和端口號(hào)的連接請(qǐng)求,當(dāng)有客戶端發(fā)送連接請(qǐng)求時(shí), # 服務(wù)端可以接受客戶端連接,并創(chuàng)建一個(gè)新的套接字對(duì)象以處理客戶端與服務(wù)端之間的通信。 server_socket.listen() def handle_client(client_socket, client_address): uid = client_socket.recv(1024).decode('utf-8') print('uid:{}已上線'.format(uid)) clients[uid] = client_socket while True: try: data = client_socket.recv(1024) if data: # 解析消息 msg_dict = json.loads(data.decode('utf-8')) sender_id = msg_dict['sender_id'] receiver_id = msg_dict['receiver_id'] message = msg_dict['message'] print('用戶 {} 發(fā)送給用戶 {} 的消息: {}'.format(sender_id, receiver_id, message)) if receiver_id == 'all': # 廣播消息給所有客戶端 for uid, client in clients.items(): if uid != sender_id: client.sendall(data) elif receiver_id in clients and receiver_id != sender_id: # 發(fā)送消息給指定客戶端 client_socket = clients[receiver_id] client_socket.sendall(data) else: # 接收到的消息有誤,回復(fù)錯(cuò)誤信息 error = { '錯(cuò)誤': f'對(duì)方ID無(wú)效: {receiver_id}' } client_socket.sendall(json.dumps(error).encode('utf-8')) else: # 客戶端關(guān)閉了連接 print('客戶端關(guān)閉了連接') break except: # 發(fā)送和接收消息時(shí)出現(xiàn)異常,可能是客戶端關(guān)閉了連接 print('發(fā)送和接收消息時(shí)出現(xiàn)異常,可能是客戶端關(guān)閉了連接') break # 移除客戶端連接 print('用戶:{}已離線'.format(uid)) del clients[uid] # 關(guān)閉客戶端連接 client_socket.close() if __name__ == '__main__': print(f'Server listening on {HOST}:{PORT}') while True: client_socket, client_address = server_socket.accept() print(f'{client_address[0]}:{client_address[1]} connected') # 生成客戶端 UID,并存儲(chǔ)客戶端的 socket 對(duì)象和 UID 的映射關(guān)系 thread = threading.Thread(target=handle_client, args=(client_socket, client_address)) thread.daemon = True thread.start()
(2) 客戶端1(client1.py)
import socket import threading import uuid import json HOST = '127.0.0.1' PORT = 8888 # 隨機(jī)生成一個(gè)唯一 ID UID = str(uuid.uuid4()) # 創(chuàng)建一個(gè) TCP 協(xié)議的套接字對(duì)象 client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 連接指定主機(jī)和端口 client_socket.connect((HOST, PORT)) # 將UID發(fā)送給服務(wù)端 client_socket.sendall(UID.encode('utf-8')) def receive_message(): while True: data = client_socket.recv(1024) if data: # 解析消息 msg_dict = json.loads(data.decode('utf-8')) if 'sender_id' in msg_dict: sender_id = msg_dict['sender_id'] message = msg_dict['message'] print(f"{sender_id}: {message}") elif 'error' in msg_dict: error = msg_dict['error'] print(f"Error: {error}") else: # 服務(wù)端關(guān)閉了連接 print('服務(wù)器已下線...') break # 關(guān)閉客戶端連接 client_socket.close() def send_message(receiver_id): while True: message = input() # 構(gòu)造消息字典 msg_dict = { 'sender_id': UID, 'receiver_id': receiver_id, 'message': message } # 將消息字典序列化為JSON格式字符串 json_str = json.dumps(msg_dict) # 將消息發(fā)送給服務(wù)端 client_socket.sendall(json_str.encode('utf-8')) if __name__ == '__main__': print(f'你的ID: {UID}') receiver_id = input('輸入對(duì)方的ID接收對(duì)方消息或者輸入"all"接收廣播消息: ') # 啟動(dòng)接收消息的線程 t1 = threading.Thread(target=receive_message) t1.daemon = True t1.start() # 啟動(dòng)發(fā)送消息的線程 t2 = threading.Thread(target=send_message, args=(receiver_id,)) t2.daemon = True t2.start() # 等待發(fā)送線程結(jié)束,結(jié)束接收線程 t2.join() t1.join()
(3) 客戶端2(client2.py)
import socket import threading import uuid import json HOST = '127.0.0.1' PORT = 8888 # 隨機(jī)生成一個(gè)唯一 ID UID = str(uuid.uuid4()) # 創(chuàng)建一個(gè) TCP 協(xié)議的套接字對(duì)象 client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 連接指定主機(jī)和端口 client_socket.connect((HOST, PORT)) # 將UID發(fā)送給服務(wù)端 client_socket.sendall(UID.encode('utf-8')) def receive_message(): while True: data = client_socket.recv(1024) if data: # 解析消息 msg_dict = json.loads(data.decode('utf-8')) if 'sender_id' in msg_dict: sender_id = msg_dict['sender_id'] message = msg_dict['message'] print(f"{sender_id}: {message}") elif 'error' in msg_dict: error = msg_dict['error'] print(f"Error: {error}") else: # 服務(wù)端關(guān)閉了連接 print('服務(wù)器已下線...') break # 關(guān)閉客戶端連接 client_socket.close() def send_message(receiver_id): while True: message = input() # 構(gòu)造消息字典 msg_dict = { 'sender_id': UID, 'receiver_id': receiver_id, 'message': message } # 將消息字典序列化為JSON格式字符串 json_str = json.dumps(msg_dict) # 將消息發(fā)送給服務(wù)端 client_socket.sendall(json_str.encode('utf-8')) if __name__ == '__main__': print(f'你的ID: {UID}') receiver_id = input('輸入對(duì)方的ID接收對(duì)方消息或者輸入"all"接收廣播消息: ') # 啟動(dòng)接收消息的線程 t1 = threading.Thread(target=receive_message) t1.daemon = True t1.start() # 啟動(dòng)發(fā)送消息的線程 t2 = threading.Thread(target=send_message, args=(receiver_id,)) t2.daemon = True t2.start() # 等待發(fā)送線程結(jié)束,結(jié)束接收線程 t2.join() t1.join()
4.系統(tǒng)功能演示
(1)啟動(dòng)服務(wù)端和客戶端
注意事項(xiàng):需要需要先啟動(dòng)服務(wù)端
(2) 通信功能
client1輸入clent2的ID,在client2里輸入clent1的ID,然后就可以開(kāi)始通信了!
5.總結(jié)
這里提供了一個(gè)簡(jiǎn)單的實(shí)現(xiàn)使用python的socket庫(kù)實(shí)現(xiàn)網(wǎng)絡(luò)通信功能。
以上就是基于Python的socket庫(kù)實(shí)現(xiàn)通信功能的示例代碼的詳細(xì)內(nèi)容,更多關(guān)于Python socket庫(kù)通信功能的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
python web框架Flask實(shí)現(xiàn)圖形驗(yàn)證碼及驗(yàn)證碼的動(dòng)態(tài)刷新實(shí)例
在本篇文章里小編給大家整理的是關(guān)于python web框架Flask實(shí)現(xiàn)圖形驗(yàn)證碼的相關(guān)知識(shí)點(diǎn),有需要的朋友們參考下。2019-10-10Python如何聲明以管理員方式運(yùn)行(附實(shí)戰(zhàn)案例)
由于Windows的安全機(jī)制,Python寫(xiě)的腳本缺少了管理員權(quán)限,運(yùn)行就會(huì)受到一些限制,這篇文章主要介紹了Python如何聲明以管理員方式運(yùn)行的相關(guān)資料,文中通過(guò)代碼介紹的非常詳細(xì),需要的朋友可以參考下2025-04-04使用Atom支持基于Jupyter的Python開(kāi)教程詳解
這篇文章主要介紹了使用Atom支持基于Jupyter的Python開(kāi)發(fā),Vscode雖然說(shuō)也有連接Jupyter的工具,但是交互式的開(kāi)發(fā)Hydrogen體驗(yàn)更好,需要的朋友可以參考下2021-08-08一個(gè)基于flask的web應(yīng)用誕生 用戶注冊(cè)功能開(kāi)發(fā)(5)
一個(gè)基于flask的web應(yīng)用誕生第五篇,這篇文章主要介紹了用戶注冊(cè)功能開(kāi)發(fā),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2017-04-04python消費(fèi)kafka數(shù)據(jù)批量插入到es的方法
今天小編就為大家分享一篇python消費(fèi)kafka數(shù)據(jù)批量插入到es的方法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-12-12