Python网络编程：socket模块的入门与实践

时间:2025/7/14 18:25:33来源：https://blog.csdn.net/Richardlygo/article/details/140574808 浏览次数:0次

Socket模块的基本概念

创建Socket

在Python中，可以使用socket模块创建Socket对象：

import socket# 创建一个TCP/IP socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

地址族与Socket类型

socket.AF_INET：IPv4地址族
socket.SOCK_STREAM：TCP流套接字
socket.SOCK_DGRAM：UDP数据报套接字

TCP编程

TCP服务器

以下是一个简单的TCP服务器示例，它接收客户端连接并返回一个简单的消息：

import socketdef tcp_server():host = '127.0.0.1'port = 12345s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)s.bind((host, port))s.listen(5)print(f"服务器启动，监听端口 {port}")while True:conn, addr = s.accept()print(f"连接地址: {addr}")conn.send(b"欢迎访问服务器！")conn.close()if __name__ == "__main__":tcp_server()

TCP客户端

以下是一个简单的TCP客户端示例，它连接到服务器并接收消息：

import socketdef tcp_client():host = '127.0.0.1'port = 12345s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)s.connect((host, port))message = s.recv(1024)print(f"从服务器接收: {message.decode()}")s.close()if __name__ == "__main__":tcp_client()

UDP编程

UDP服务器

以下是一个简单的UDP服务器示例，它接收客户端消息并返回响应：


import socketdef udp_server():host = '127.0.0.1'port = 12345s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)s.bind((host, port))print(f"UDP服务器启动，监听端口 {port}")while True:data, addr = s.recvfrom(1024)print(f"收到来自 {addr} 的消息: {data.decode()}")s.sendto(b"已收到消息", addr)if __name__ == "__main__":udp_server()

UDP客户端

以下是一个简单的UDP客户端示例，它发送消息到服务器并接收响应：

import socketdef udp_client():host = '127.0.0.1'port = 12345s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)message = "你好，服务器"s.sendto(message.encode(), (host, port))data, addr = s.recvfrom(1024)print(f"从服务器接收: {data.decode()}")s.close()if __name__ == "__main__":udp_client()

错误处理

在网络编程中，处理可能出现的错误是非常重要的。可以使用try-except语句来捕获和处理异常。

import socketdef tcp_client_with_error_handling():host = '127.0.0.1'port = 12345try:s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)s.connect((host, port))message = s.recv(1024)print(f"从服务器接收: {message.decode()}")except socket.error as e:print(f"Socket 错误: {e}")finally:s.close()if __name__ == "__main__":tcp_client_with_error_handling()

高级Socket编程

多线程TCP服务器

使用多线程可以处理多个客户端连接。

import socket
import threadingdef handle_client(conn, addr):print(f"连接地址: {addr}")conn.send(b"欢迎访问服务器！")conn.close()def tcp_server_multithreaded():host = '127.0.0.1'port = 12345s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)s.bind((host, port))s.listen(5)print(f"服务器启动，监听端口 {port}")while True:conn, addr = s.accept()client_thread = threading.Thread(target=handle_client, args=(conn, addr))client_thread.start()if __name__ == "__main__":tcp_server_multithreaded()

非阻塞Socket

非阻塞Socket允许在没有数据时立即返回，而不是等待数据。


import socketdef non_blocking_client():host = '127.0.0.1'port = 12345s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)s.setblocking(0)try:s.connect((host, port))except BlockingIOError:passwhile True:try:message = s.recv(1024)if message:print(f"从服务器接收: {message.decode()}")breakexcept BlockingIOError:continues.close()if __name__ == "__main__":non_blocking_client()

实际应用示例

简单的聊天室

可以通过TCP编写一个简单的聊天室，服务器接收消息并广播给所有连接的客户端。

聊天服务器


import socket
import threadingclients = []def broadcast(message, client_socket):for client in clients:if client != client_socket:try:client.send(message)except:clients.remove(client)def handle_client(client_socket):while True:try:message = client_socket.recv(1024)if message:print(f"收到消息: {message.decode()}")broadcast(message, client_socket)except:clients.remove(client_socket)client_socket.close()breakdef chat_server():host = '127.0.0.1'port = 12345server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)server_socket.bind((host, port))server_socket.listen(5)print(f"聊天服务器启动，监听端口 {port}")while True:client_socket, addr = server_socket.accept()print(f"连接地址: {addr}")clients.append(client_socket)client_thread = threading.Thread(target=handle_client, args=(client_socket,))client_thread.start()if __name__ == "__main__":chat_server()

聊天客户端

import socket
import threadingdef receive_messages(client_socket):while True:try:message = client_socket.recv(1024)if message:print(message.decode())except:breakdef chat_client():host = '127.0.0.1'port = 12345client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)client_socket.connect((host, port))receive_thread = threading.Thread(target=receive_messages, args=(client_socket,))receive_thread.start()while True:message = input()client_socket.send(message.encode())if __name__ == "__main__":chat_client()