一文带您了解Python 网络编程：socket 示例

大家好，欢迎来到IT知识分享网。

套接字（Sockets ）和套接字 API 用于在网络上传递消息，它们提供了一种进程间通信 (inter-process communication，IPC) 的形式。网络可以是计算机上的一个逻辑本地网络，也可以是一个物理上连接到外部网络的网络，并通过该外部网络连接到其他网络。

套接字（Sockets ）有着悠久的历史。它们的使用起源于1971年的ARPAnet，后来在1983年发布的伯克利软件发布版（Berkeley Software Distribution，BSD）操作系统中成为了一种API，被称为伯克利套接字（Berkeley sockets）。

在90年代，随着万维网（World Wide Web）的兴起，网络编程也迅速发展。利用新连接的网络并使用套接字的不仅仅是Web服务器和浏览器。各种类型和规模的客户端-服务器应用程序也得到了广泛应用。

今天，尽管套接字API使用的底层协议多年来有所演变，并且出现了新协议，但底层API本身保持不变。

最常见的套接字应用程序类型是客户端-服务器应用程序，其中一方充当服务器并等待来自客户端的连接，这是主要的网络模式。

Python Socket API 概述

Python 的 socket 模块提供了一组API接口，用于访问套接字 API(the Berkeley sockets API)。该模块中的主要API 函数和方法包括：

• socket()
• .bind()
• .listen()
• .accept()
• .connect()
• .connect_ex()
• .send()
• .recv()
• .close()

Python 提供了一个方便且一致的 API，它直接映射到系统调用及其对应的 C 函数。作为其标准库的一部分，Python 还提供了一些类，使得使用这些底层套接字函数更加简单,比如 socketserver 模块，这是一个用于网络服务器的框架；此外，还有许多模块实现了更高级的互联网协议，如 HTTP 和 SMTP。

TCP 套接字

使用 socket.socket() 创建一个套接字对象，并将套接字类型指定为 socket.SOCK_STREAM。默认使用的协议是传输控制协议 ( Transmission Control Protocol ，TCP)。

传输控制协议 (TCP) 具有以下特点：

• 可靠性：网络中丢失的数据包会被发送方检测并重新传输。
• 按序数据传递：您的应用程序将按发送方写入数据的顺序读取数据。

相比之下，也可以使用 socket.SOCK_DGRAM 创建的用户数据报协议 (User Datagram Protocol，UDP) 套接字不具备可靠性，接收方读取的数据可能会与发送方写入的数据顺序不一致。TCP 让您无需担心数据包丢失、数据到达顺序混乱以及其他在网络通信中不可避免的陷阱。下图是 TCP 的套接字 API 调用顺序和数据流：

左侧列表示服务器。右侧列表示客户端。从左上角开始，注意服务器为设置“监听”套接字所进行的 API 调用：

• socket()
• .bind()
• .listen()
• .accept()

监听套接字的作用正如其名称所示：它监听来自客户端的连接。当客户端连接时，服务器调用 .accept() 来接受或完成连接。客户端调用 .connect() 来建立与服务器的连接，并启动三次握手。握手步骤很重要，因为它确保连接的每一端在网络中是可达的，换句话说，客户端可以到达服务器，反之亦然。有时，可能只有一个主机、客户端或服务器可以到达另一个。在中间部分是往返通信阶段，客户端和服务器通过调用 .send() 和 .recv() 来交换数据。最后，客户端和服务器关闭各自的套接字。

Echo Client and Server

上面介绍了套接字 API 以及客户端和服务器如何通信，下面是一个最为简单的第一个客户端和服务器。将从一个简单的实现开始。服务器将简单地将接收到的内容原样返回给客户端。

以下是服务器的代码：

import socket HOST = "127.0.0.1" # Standard loopback interface address (localhost) PORT = 65432 # Port to listen on (non-privileged ports are > 1023) with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind((HOST, PORT)) s.listen() while True: conn, addr = s.accept() with conn: print(f"Connected by {addr}") while True: data = conn.recv(1024) if not data: break conn.sendall(data) print(data)

这段代码实现了一个简单的回显服务器，功能如下：

1. 导入模块：使用 socket 模块来进行网络编程。
2. 定义地址和端口：服务器监听本地主机 (127.0.0.1) 和端口 65432。
3. 创建套接字：使用 IPv4 和 TCP 协议创建一个套接字。
4. 绑定和监听：将套接字绑定到指定的地址和端口，然后开始监听连接请求。
5. 处理连接：接受客户端连接并打印客户端地址。
6. 数据接收与回显：接收客户端发送的数据并将其回显给客户端，直到客户端断开连接。

以下是client 代码：

import socket HOST = "127.0.0.1" # The server's hostname or IP address PORT = 65432 # The port used by the server with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((HOST, PORT)) s.sendall(b"Hello, world") data = s.recv(1024) print(f"Received {data!r}")

这段代码实现了一个简单的客户端，功能如下：

1. 导入模块：使用 socket 模块进行网络编程。
2. 定义服务器地址和端口：客户端连接到本地主机 (127.0.0.1) 的端口 65432。
3. 创建并连接套接字：使用 IPv4 和 TCP 协议创建套接字，并连接到指定的服务器地址和端口。
4. 发送数据：向服务器发送字节数据 b”Hello, world”。
5. 接收数据：从服务器接收最多 1024 字节的数据。
6. 打印接收到的数据：将接收到的数据以原始格式输出。

通信过程解析

现在，您将更详细地了解客户端和服务器之间是如何进行通信的：

使用回环接口（ loopback interface ）（IPv4 地址 127.0.0.1 或 IPv6 地址 ::1）时，数据不会离开主机或接触到外部网络。在上面的示意图中，回环接口（ loopback interface ）位于主机内部。这代表了回环接口的内部特性，显示了穿越它的连接和数据仅在主机内部。这也是为什么回环接口和 IP 地址 127.0.0.1 或 ::1 被称为“localhost”。

应用程序使用回环接口（ loopback interface ）与在主机上运行的其他进程进行通信，同时确保安全性和与外部网络的隔离。因为它是内部的，仅从主机内部可以访问，所以不会暴露在外部。

当您在应用程序中使用 127.0.0.1 或 ::1 以外的 IP 地址时，它通常绑定到连接到外部网络的以太网接口。这是通向“localhost”之外的其他主机的网关。

多连接客户端和服务器

在接下来的两个部分中，您将创建一个服务器和客户端，使用来自 selectors 模块的选择器对象来处理多个连接。服务器示例代码如下：

# multiconn-server.py import sys import socket import selectors import types sel = selectors.DefaultSelector() host = "127.0.0.1" port = 65432 lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) lsock.bind((host, port)) lsock.listen() print(f"Listening on {(host, port)}") lsock.setblocking(False) sel.register(lsock, selectors.EVENT_READ, data=None) def accept_wrapper(sock): conn, addr = sock.accept() # Should be ready to read print(f"Accepted connection from {addr}") conn.setblocking(False) data = types.SimpleNamespace(addr=addr, inb=b"", outb=b"") events = selectors.EVENT_READ | selectors.EVENT_WRITE sel.register(conn, events, data=data) def service_connection(key, mask): sock = key.fileobj data = key.data if mask & selectors.EVENT_READ: recv_data = sock.recv(1024) # Should be ready to read if recv_data: data.outb += recv_data else: print(f"Closing connection to {data.addr}") sel.unregister(sock) sock.close() if mask & selectors.EVENT_WRITE: if data.outb: print(f"Echoing {data.outb!r} to {data.addr}") sent = sock.send(data.outb) # Should be ready to write data.outb = data.outb[sent:] try: while True: events = sel.select(timeout=None) for key, mask in events: if key.data is None: accept_wrapper(key.fileobj) else: service_connection(key, mask) except KeyboardInterrupt: print("Caught keyboard interrupt, exiting") finally: sel.close()

这段代码实现了一个多连接的服务器，能够同时处理多个客户端的连接。

导入和初始化

• socket 和 selectors 模块用于创建和管理网络连接。
• sel 是选择器对象，用于管理多个套接字的 I/O 事件。
• 服务器设置
• 创建一个监听套接字 lsock，绑定到 127.0.0.1 地址和端口 65432。
• 设置套接字为非阻塞模式，并使用选择器 sel 注册监听套接字，监控其 EVENT_READ 事件（即有新的连接请求）。

accept_wrapper 函数

• 当有新连接到来时，接受连接，并为每个连接创建一个新的非阻塞套接字。
• 为新连接创建一个数据对象 data，包含连接的地址以及输入和输出缓冲区。
• 使用选择器注册新连接，监控其 EVENT_READ 和 EVENT_WRITE 事件。

service_connection 函数：处理现有连接的 I/O 操作：

• 如果有数据可读 (EVENT_READ)，从客户端读取数据并将其存储在输出缓冲区中。
• 如果可以写入 (EVENT_WRITE)，将输出缓冲区中的数据发送回客户端（回显）。

事件循环

• 在 try 块中，持续监听和处理所有注册的套接字的 I/O 事件。
• 如果有新的连接请求，调用 accept_wrapper 处理。
• 如果有现有连接的读写事件，调用 service_connection 处理。
• 使用 except KeyboardInterrupt 捕获键盘中断，安全地关闭选择器并退出。

该服务器可以同时处理多个客户端连接，接收并回显客户端发送的数据。通过使用 selectors 模块，服务器能够高效地管理多个非阻塞套接字，从而实现多连接处理。

client示例代码如下：

import sys import socket import selectors import types sel = selectors.DefaultSelector() messages = [b"Message 1 from client.", b"Message 2 from client."] def start_connections(host, port, num_conns): server_addr = (host, port) for i in range(0, num_conns): connid = i + 1 print(f"Starting connection {connid} to {server_addr}") sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setblocking(False) sock.connect_ex(server_addr) events = selectors.EVENT_READ | selectors.EVENT_WRITE data = types.SimpleNamespace( connid=connid, msg_total=sum(len(m) for m in messages), recv_total=0, messages=messages.copy(), outb=b"", ) sel.register(sock, events, data=data) def service_connection(key, mask): sock = key.fileobj data = key.data if mask & selectors.EVENT_READ: recv_data = sock.recv(1024) # Should be ready to read if recv_data: print(f"Received {recv_data!r} from connection {data.connid}") data.recv_total += len(recv_data) if not recv_data or data.recv_total == data.msg_total: print(f"Closing connection {data.connid}") sel.unregister(sock) sock.close() if mask & selectors.EVENT_WRITE: if data.messages: data.outb = data.messages.pop(0) if data.outb: print(f"Sending {data.outb!r} to connection {data.connid}") sent = sock.send(data.outb) # Should be ready to write data.outb = data.outb[sent:] # Start the connections host, port = "127.0.0.1", 65432 num_conns = 2 start_connections(host, port, num_conns) try: while True: events = sel.select(timeout=None) for key, mask in events: service_connection(key, mask) except KeyboardInterrupt: print("Caught keyboard interrupt, exiting") finally: sel.close()

导入和初始化

• socket 和 selectors 模块用于创建和管理网络连接，处理多个非阻塞套接字的 I/O 事件。
• messages 是客户端将发送的消息列表。

start_connections 函数

• 用于启动多个客户端连接。
• 每个连接都会创建一个套接字并设置为非阻塞模式，然后尝试连接到服务器。
• 使用 selectors 注册每个套接字，以监控其读写事件。

service_connection 函数

• 处理每个连接的 I/O 操作。
• 如果有数据可读 (EVENT_READ)，从服务器读取数据并检查是否接收完毕。
• 如果可以写入 (EVENT_WRITE)，发送消息到服务器。
• 当所有消息发送完毕且数据接收完成后，关闭连接。

事件循环

• 持续监听和处理连接的 I/O 事件，直到所有连接关闭或用户中断程序。

这个代码实现了一个多连接的客户端，可以同时处理多个与服务器的连接，发送和接收数据。

实现一个ping

ping 通过发送 ICMP 回显请求来检查主机是否在线并连接到网络。下面代码实现一个ping 功能：

import os import socket import struct import time import select ICMP_ECHO_REQUEST = 8 # ICMP Echo Request type (ping) ICMP_CODE = socket.getprotobyname("icmp") def checksum(source_string): """ 计算ICMP校验和 """ sum = 0 count_to = (len(source_string) // 2) * 2 for count in range(0, count_to, 2): this_val = source_string[count + 1] * 256 + source_string[count] sum = sum + this_val sum = sum & 0xffffffff if count_to < len(source_string): sum = sum + source_string[len(source_string) - 1] sum = sum & 0xffffffff sum = (sum >> 16) + (sum & 0xffff) sum = sum + (sum >> 16) answer = ~sum answer = answer & 0xffff answer = answer >> 8 | (answer << 8 & 0xff00) return answer def create_packet(id): """ 创建ICMP数据包 """ header = struct.pack('bbHHh', ICMP_ECHO_REQUEST, 0, 0, id, 1) data = struct.pack('d', time.time()) my_checksum = checksum(header + data) header = struct.pack('bbHHh', ICMP_ECHO_REQUEST, 0, socket.htons(my_checksum), id, 1) return header + data def ping(host, count=4): """ 发送ICMP Echo Request，并接收Echo Reply """ try: dest = socket.gethostbyname(host) except socket.gaierror: print(f"Ping request could not find host {host}. Please check the name and try again.") return print(f"正在 Ping {host} [{dest}] 具有 32 字节的数据:") sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, ICMP_CODE) sock.settimeout(1) packet_id = os.getpid() & 0xFFFF times = [] for i in range(count): packet = create_packet(packet_id) start_time = time.time() sock.sendto(packet, (dest, 1)) while True: start_select = time.time() what_ready = select.select([sock], [], [], 1) how_long_in_select = (time.time() - start_select) if what_ready[0] == []: # 超时 print("请求超时。") break time_received = time.time() rec_packet, addr = sock.recvfrom(1024) icmp_header = rec_packet[20:28] type, code, checksum, packet_id, sequence = struct.unpack("bbHHh", icmp_header) if type == 0 and packet_id == os.getpid() & 0xFFFF: bytes_in_double = struct.calcsize("d") time_sent = struct.unpack("d", rec_packet[28:28 + bytes_in_double])[0] rtt = (time_received - time_sent) * 1000 times.append(rtt) ttl = struct.unpack("B", rec_packet[8:9])[0] print(f"来自 {addr[0]} 的回复: 字节=32 时间={int(rtt)}ms TTL={ttl}") break if time.time() - start_time > 1: print("请求超时。") break time.sleep(1) # 等待1秒钟再发送下一个请求 sock.close() # 输出统计信息 if times: print(f"\n{dest} 的 Ping 统计信息:") print(f" 数据包: 已发送 = {count}, 已接收 = {len(times)}, 丢失 = {count - len(times)} ({((count - len(times)) / count) * 100}% 丢失)，") print(f"往返行程的估计时间(以毫秒为单位):") print(f" 最短 = {int(min(times))}ms，最长 = {int(max(times))}ms，平均 = {int(sum(times) / len(times))}ms") else: print("请求超时。") # 使用示例 ping("www.baidu.com") print("*"*50) ping('172.30.81.86')

这段代码实现了一个简单的 ping 命令，使用 Python 发送 ICMP Echo 请求，并接收 Echo 回复，以检查目标主机是否在线并测量响应时间。以下是代码的简要解读：

1. ICMP_ECHO_REQUEST 和 ICMP_CODE:
2. ICMP_ECHO_REQUEST 是 ICMP Echo 请求的类型码（值为 8），表示这是一个 ping 请求。
3. ICMP_CODE 通过 socket.getprotobyname(“icmp”) 获取 ICMP 协议的编号，用于创建原始套接字。
4. checksum 函数:
5. 计算 ICMP 数据包的校验和，用于确保数据在传输过程中没有损坏。
6. create_packet 函数:
7. 创建 ICMP 数据包，包括报头和数据部分。报头包含类型、代码、校验和、ID 和序列号。数据部分包含当前的时间戳，用于计算往返时间（RTT）。
8. ping 函数:
9. ping 函数接受目标主机名和发送请求的次数（默认为 4）。
10. 首先解析目标主机名为 IP 地址，然后创建一个原始套接字。
11. 通过循环发送 ICMP 请求并接收响应。每次发送请求后，使用 select 函数等待响应，处理超时情况。
12. 接收到响应后，解析 ICMP 报头，提取 RTT（往返时间）和 TTL（生存时间），并输出格式化的结果。
13. 完成所有请求后，输出统计信息，包括数据包的发送和接收数量、丢失的百分比以及 RTT 的最短、最长和平均值。
14. 输出结果:
15. 代码执行时，将以格式化的方式输出 ping 的结果，类似于系统的 ping 命令。

这段代码实现了一个基本的 ping 工具，可以通过 ICMP 协议检测网络连接情况并测量延迟。

Python socket 编程是一个基本但非常重要的主题。网络和套接字涉及的内容非常广泛，需要相当多的时间来熟悉和掌握这些概念。就像学习 Python 一样，随着你对这些独立部分的逐步熟悉，并投入更多时间进行理解和实践，这些内容最终会变得更加清晰易懂。