Python网络编程：socket模块的入门与实践
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 socket
def tcp_server():
host = '127.0.0.1'
port = 12345
s = 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 socket
def tcp_client():
host = '127.0.0.1'
port = 12345
s = 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 socket
def udp_server():
host = '127.0.0.1'
port = 12345
s = 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 socket
def udp_client():
host = '127.0.0.1'
port = 12345
s = 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 socket
def tcp_client_with_error_handling():
host = '127.0.0.1'
port = 12345
try:
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 threading
def handle_client(conn, addr):
print(f"连接地址: {addr}")
conn.send(b"欢迎访问服务器！")
conn.close()
def tcp_server_multithreaded():
host = '127.0.0.1'
port = 12345
s = 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 socket
def non_blocking_client():
host = '127.0.0.1'
port = 12345
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setblocking(0)
try:
s.connect((host, port))
except BlockingIOError:
pass
while True:
try:
message = s.recv(1024)
if message:
print(f"从服务器接收: {message.decode()}")
break
except BlockingIOError:
continue
s.close()
if __name__ == "__main__":
non_blocking_client()实际应用示例
简单的聊天室
可以通过TCP编写一个简单的聊天室，服务器接收消息并广播给所有连接的客户端。
聊天服务器
import socket
import threading
clients = []
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()
break
def chat_server():
host = '127.0.0.1'
port = 12345
server_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 threading
def receive_messages(client_socket):
while True:
try:
message = client_socket.recv(1024)
if message:
print(message.decode())
except:
break
def chat_client():
host = '127.0.0.1'
port = 12345
client_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()
原文地址：https://blog.csdn.net/Richardlygo/article/details/140574808
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！