2025-06-10

Linux网络编程实战:自定义协议与序列化/反序列化技术详解

本篇文章将从自定义网络协议设计的基本原则出发,逐步讲解如何在 Linux 环境下以 C 语言实现自定义协议的序列化(serialization)与反序列化(deserialization)技术。通过代码示例、图解与详细说明,帮助你迅速掌握构建高效、可靠网络通信的核心技能。

目录

  1. 引言
  2. 自定义协议设计要点

  3. 序列化与反序列化基础原理

  4. 示例协议定义与数据包结构

  5. 序列化实现详解(发送端)

  6. 反序列化实现详解(接收端)

  7. 实战:完整客户端与服务器示例

  8. 常见注意事项与优化建议

  9. 总结

1. 引言

在现代分布式系统、网络服务中,往往需要在不同组件之间实现高效、可靠的数据交换。虽然诸如 HTTP、WebSocket、gRPC、Protocol Buffers 等通用协议和框架已广泛应用,但在某些性能敏感或定制化需求场景下(如游戏服务器、物联网设备、嵌入式系统等),我们仍需针对业务特点自定义轻量级协议。

自定义协议的核心在于:

  1. 尽可能少的头部开销,减少单条消息的网络流量;
  2. 明确的字段定义与固定/变长设计,方便快速解析;
  3. 可拓展性,当新功能增加时,可以向后兼容。

本文以 Linux C 网络编程为切入点,深入剖析从协议设计到序列化与反序列化实现的全过程,帮助你在 0-1 之间掌握一套定制化高效协议的开发思路与实践细节。


2. 自定义协议设计要点

2.1 为什么需要自定义协议

  • 性能需求:在高并发、低延迟场景下,尽量减少额外字符与冗余字段,比如在游戏服务器,网络带宽和处理时延都很敏感;
  • 资源受限:在嵌入式、物联网设备上,CPU 和内存资源有限,不能使用过于臃肿的高级库;
  • 协议可控:最大限度贴合业务需求,高度灵活,可随时调整;
  • 跨语言/跨平台定制:在没有统一框架的前提下,不同设备需手动实现解析逻辑,自定义协议能使双方达成一致。

2.2 协议结构的核心组成

一个自定义二进制协议,通常包含以下几部分:

  1. 固定长度的包头(Header)

    • 一般包含:版本号、消息类型、数据总长度、消息 ID、校验码/签名等;
    • 通过包头能够快速判断整条报文长度,从而做粘包/拆包处理;
  2. 可选的扩展字段(Options/Flags)

    • 如果协议需进一步扩展,可以预留若干字节用于标识后续字段含义;
    • 比如支持压缩、加密等标志;
  3. 可变长度的消息体(Payload)

    • 具体业务数据,如聊天内容、指令参数、二进制文件片段等;
    • 通常根据包头中的 length 指定其长度;
  4. 可选的尾部校验(Checksum/MAC)

    • 对整个包(或包头+消息体)做 CRC 校验,确保数据在传输过程中未被篡改。

图示:协议整体三段式结构

+----------+----------------------+---------------+
| Packet   | Payload              | Checksum      |
| Header   | (Data Body)          | (可选)       |
+----------+----------------------+---------------+
| fixed    | variable             | fixed (e.g., 4B) |
+----------+----------------------+---------------+

其中,Header 中最关键的是:

  • Magic Number(魔数)或协议版本:用于快速校验是否为本协议;
  • Payload Length:指明消息体长度,接收端据此分配缓存并防止粘包;
  • Message Type / Command:指明消息的业务含义,接收端根据类型派发给不同的处理函数;
  • Request ID / Sequence Number(可选):用于客户端-服务器双向交互模式下的请求/响应映射。

2.3 常见协议字段与对齐问题

在 C 语言中直接定义结构体时,编译器会对字段进行对齐(alignment)——默认 32 位系统会按 4 字节对齐、64 位按 8 字节对齐。若我们直接将结构体 sizeof 的内存块当作网络报文头部,可能会多出“填充字节”(Padding),导致发送的数据与预期格式不一致。

示例:结构体默认对齐产生的额外字节

// 假设在 64 位 Linux 下编译
struct MyHeader {
    uint32_t magic;       // 4 字节
    uint16_t version;     // 2 字节
    uint16_t msg_type;    // 2 字节
    uint32_t payload_len; // 4 字节
};
// 编译器会按 4 字节对齐,sizeof(MyHeader) 可能为 12 字节(无填充)
// 但如果字段顺序不当,比如 uint8_t 在前面,就会出现填充字节。

如果想强制“紧凑打包”,可使用:

#pragma pack(push, 1)
struct MyHeader {
    uint32_t magic;       // 4 B
    uint16_t version;     // 2 B
    uint16_t msg_type;    // 2 B
    uint32_t payload_len; // 4 B
};
#pragma pack(pop)
// 通过 #pragma pack(1) 可确保 sizeof(MyHeader) == 12,无填充

设计要点总结

  • 明确字段顺序与大小:可从大到小、或将同类型字段放在一起,减少隐式对齐带来的填充;
  • 使用 #pragma pack(1)__attribute__((packed)):编译器指令,保证结构体按“字节对齐”最小化;
  • 避免直接把结构体整体 memcpy 到网络缓冲区,除非你清楚对齐与端序问题

3. 序列化与反序列化基础原理

3.1 什么是序列化

序列化(Serialization)指的是将程序中使用的内存数据结构(如结构体、对象)转换为可在网络中传输存储到磁盘连续字节流,常见场景:

  • 在网络传输场景下,将多个字段、数组、字符串等进行“打包”后通过 socket send() 发送;
  • 在持久化场景下,将内存中的对象写入文件、数据库;

序列化的要求

  1. 可还原(可逆):接收端必须能够根据字节流还原到与发送端完全一致的结构;
  2. 跨平台一致性:如果发送端是大端(Big-endian),接收端是小端(Little-endian),需要统一约定;
  3. 高效:控制序列化后的字节长度,避免冗余;

3.2 什么是反序列化

反序列化(Deserialization)指的是将接收到的字节流还原为程序可用的数据结构(如结构体、数组、字符串)。具体步骤:

  1. 解析固定长度头部:根据协议定义,从字节流中取出前 N 个字节,将其填充到对应的字段中;
  2. 根据头部字段值动态分配或读取:如头部给定 payload_len = 100,此时就需要从 socket 中再 recv(100) 字节;
  3. 将读取的字节赋值或 memcpy 到结构体字段或指针缓冲区

    • 对于数值(整数、浮点数)需要做“字节序转换”(htonl/ntohl 等);
    • 对于字符串/二进制数据可直接 memcpy

如果协议中还包含校验和或签名,需要在“还原完整结构”后进行一次校验,确保数据未损坏。

3.3 端序(Endian)与字节对齐

  • 端序:大端(Big‐Endian)与小端(Little‐Endian)。x86/x64 架构一般使用小端存储,即数值最低有效字节放在内存低地址;而网络规范(TCP/IP)更常使用大端(网络字节序)。

    • 小端示例(0x12345678 存储在连续 4 字节内存):

      内存地址 ↑
      +--------+--------+--------+--------+
      | 0x78   | 0x56   | 0x34   | 0x12   |
      +--------+--------+--------+--------+
    • 大端示例

      内存地址 ↑
      +--------+--------+--------+--------+
      | 0x12   | 0x34   | 0x56   | 0x78   |
      +--------+--------+--------+--------+

在网络通信中,必须统一使用网络字节序(大端)传输整数,常用函数:

  • htonl(uint32_t hostlong):将主机字节序(host)转换为网络字节序(network),针对 32 位;
  • htons(uint16_t hostshort):针对 16 位;
  • ntohl(uint32_t netlong)ntohs(uint16_t netshort):分别将网络字节序转换为主机字节序。

注意:浮点数没有标准的 “htonf/ntohf”,如果协议中需要传输浮点数,一般做法是:

  1. 将浮点数 floatdouble 通过 memcpy 拷贝到 uint32_t / uint64_t
  2. 再用 htonl / htonll(若平台支持)转换,接收端再逆向操作。
  • 字节对齐:如前文所述,C 语言中的结构体会为了快速访问而在字段之间填充“对齐字节”。若直接 memcpy(&mystruct, buf, sizeof(mystruct)) 会导致与协议设计不一致,需手动“紧凑打包”或显式地一个字段一个字段地写入/读取。

4. 示例协议定义与数据包结构

为了让读者更直观地理解,下文将以“简易聊天协议”为例,设计一套完整的二进制协议,包含文本消息心跳包两种类型。

4.1 示例场景:简易聊天协议

  • 客户端与服务器之间需进行双向文本通信,每条消息需携带:

    1. 消息类型(1=文本消息,2=心跳包)
    2. 消息序号(uint32):用于确认;
    3. 用户名长度(uint8) + 用户名内容
    4. 消息正文长度(uint16) + 消息正文内容
  • 当客户端无数据发送超时(例如 30 秒未发任何消息)时,需发送“心跳包”以维持连接;服务器端收到心跳包后,只需回复一个“心跳响应”(类型=2)即可。

4.2 数据包整体结构图解

+==========================  Header (固定长度) ==========================+
| Magic (2B) | Version (1B) | MsgType (1B) | MsgSeq (4B) | UsernameLen (1B) | 
+==========================================================================+
|   Username (variable, UsernameLen B)                                     
+==========================================================================+
|   BodyLen (2B)   |   Body (variable, BodyLen B)                           
+==========================================================================+
|   Checksum (4B, 可选)                                                     
+==========================================================================+
  • Magic (2B):协议标识,如 0xABCD
  • Version (1B):协议版本,如 0x01
  • MsgType (1B):消息类型,1=文本消息;2=心跳包;
  • MsgSeq (4B):消息序号,自增的 uint32_t
  • UsernameLen (1B):用户名长度,最多 255 字节;
  • Username (variable):根据 UsernameLen,存储用户名(UTF-8);
  • BodyLen (2B):正文长度,uint16_t,最多 65535 字节;
  • Body (variable):正文内容,例如聊天文字(UTF-8);
  • Checksum (4B,可选):可以使用 CRC32,也可以不加;如果加,则在整个包(从 Magic 到 Body)计算 CRC。

示意图(ASCII 版)

┌────────────────────────────────────────────────────────────────────┐
│  Off  |  Size  | Field                                           │
├────────────────────────────────────────────────────────────────────┤
│   0   |   2B   | Magic: 0xABCD                                  │
│   2   |   1B   | Version: 0x01                                  │
│   3   |   1B   | MsgType: 1 or 2                                │
│   4   |   4B   | MsgSeq (uint32_t, 网络字节序)                   │
│   8   |   1B   | UsernameLen (uint8_t)                           │
│   9   | UsernameLen │ Username (UTF-8, 变长)                   │
│  9+ULen   │   2B   │ BodyLen (uint16_t, 网络字节序)            │
│ 11+ULen   │ BodyLen  │ Body (UTF-8, 变长)                          │
│11+ULen+BLen│  4B   │ Checksum (uint32_t, 可选,网络字节序)         │
└────────────────────────────────────────────────────────────────────┘

4.3 字段说明

  1. Magic (2B)

    • 固定值 0xABCD,用于快速判定“这是不是我们设计的协议包”;
    • 接收端先 recv(2),判断是否为 0xABCD,否则可直接断开或丢弃。
  2. Version (1B)

    • 允许未来对协议进行“升级”时进行版本兼容检查;
    • 例如当前版本为 0x01,若收到版本不一致,可告知客户端进行升级。
  3. MsgType (1B)

    • 1 表示文本消息2 表示心跳包
    • 接收端 switch(msg_type) 分发到不同的处理函数,文本消息需要继续解析用户名与正文,而心跳包只需立刻回复一个空心跳响应包。
  4. MsgSeq (4B)

    • 用于客户端/服务器做双向消息确认时可以对号入座,或用于重传策略;
    • 必须使用 htonl() 将本机字节序转换为网络字节序;
  5. UsernameLen (1B) + Username (variable)

    • 用户名长度最多 255 字节,UTF-8 编码支持多语言;
    • 存储后无需以 \0 结尾,因为长度已经在前面给出。
  6. BodyLen (2B) + Body (variable)

    • 正文长度采用 uint16_t(最大 65535),已能满足绝大多数聊天消息需求;
    • 同样无需追加结尾符,接收端根据长度精确 recv
  7. Checksum (4B,可选)

    • 协议包从 Magic(字节 0)到 Body 的最后一个字节,全部计算一次 CRC32(或其他校验方式),将结果插入最后 4 字节;
    • 接收端在收到完整包后再次计算 CRC32,与此字段对比,一致则数据正常,否则丢弃或重传。

为什么要有 Checksum?

  • 在高可靠性要求下(例如关键指令、金融交易),网络传输可能会引入数据位翻转,CRC32 校验可以快速过滤坏包;
  • 如果对延迟更敏感,可取消 Checksum 节省 4 字节与计算开销。

5. 序列化实现详解(发送端)

下面从“发送端”角度,详细讲解如何将上述协议设计“打包”为字节流,通过 socket send() 发出。

5.1 C 语言结构体定义

#include <stdint.h>

#pragma pack(push, 1) // 1 字节对齐,避免编译器插入填充字节
typedef struct {
    uint16_t magic;      // 2B:固定 0xABCD
    uint8_t  version;    // 1B:协议版本,0x01
    uint8_t  msg_type;   // 1B:1=文本消息, 2=心跳
    uint32_t msg_seq;    // 4B:消息序号(网络字节序)
    uint8_t  user_len;   // 1B:用户名长度
    // Username 紧随其后,大小 user_len
    // uint16_t body_len // 2B:正文长度(网络字节序)
    // Body 紧随其后,大小 body_len
    // uint32_t checksum // 4B:CRC32 (可选)
} PacketHeader;
#pragma pack(pop)

#define MAGIC_NUMBER 0xABCD
#define PROTOCOL_VERSION 0x01

// 校验是否真正按照 1 字节对齐
// sizeof(PacketHeader) == 9
  • #pragma pack(push, 1) / #pragma pack(pop) 强制结构体按 1 字节对齐,确保 sizeof(PacketHeader) == 9(2 + 1 + 1 + 4 + 1 = 9)。
  • Username 与 Body 均为“变长跟随”,不能写入到这一固定大小的结构体里。

5.2 手动填充与字节转换

要打包一条“文本消息”,需要依次执行以下步骤:

  1. 分配一个足够大的缓冲区,至少要能容纳 PacketHeader + username + body + (可选checksum)
  2. 填充 PacketHeader

    • magic = htons(MAGIC_NUMBER);
    • version = PROTOCOL_VERSION;
    • msg_type = 1;
    • msg_seq = htonl(next_seq);
    • user_len = username_len;
  3. memcpy 复制 Username 紧跟在 Header 之后;
  4. 填充 BodyLen:在 Username 之后的位置写入 uint16_t body_len = htons(actual_body_len);
  5. memcpy 复制 Body(正文文字)
  6. 计算并填充 Checksum(可选)

    • 假设要加 CRC32,则在 buf 从字节 0 到 body_end 计算 CRC32,得到 uint32_t crc = crc32(buf, header_len + user_len + 2 + body_len);
    • crc = htonl(crc); memcpy(buf + offset_of_checksum, &crc, 4);
#include <arpa/inet.h>
#include <stdlib.h>
#include <string.h>
#include <zlib.h> // 假设使用 zlib 提供的 CRC32 函数

/**
 * 构造并发送一条文本消息
 * @param sockfd      已建立连接的 socket 描述符
 * @param username    用户名字符串(C-字符串,\0 结尾,但不传输 \0)
 * @param message     正文字符串
 * @param seq         本次消息序号,自增
 * @return int       成功返回 0,失败返回 -1
 */
int send_text_message(int sockfd, const char *username, const char *message, uint32_t seq) {
    size_t username_len = strlen(username);
    size_t body_len     = strlen(message);

    if (username_len > 255 || body_len > 65535) {
        return -1; // 超过协议限制
    }

    // ① 计算总长度:Header (9B) + Username + BodyLen (2B) + Body + Checksum (4B)
    size_t total_len = sizeof(PacketHeader) + username_len + 2 + body_len + 4;
    uint8_t *buf = (uint8_t *)malloc(total_len);
    if (!buf) return -1;

    // ② 填充 PacketHeader
    PacketHeader header;
    header.magic    = htons(MAGIC_NUMBER);    // 网络字节序
    header.version  = PROTOCOL_VERSION;
    header.msg_type = 1;                      // 文本消息
    header.msg_seq  = htonl(seq);             // 网络字节序
    header.user_len = (uint8_t)username_len;

    // ③ 复制 Header 到 buf
    memcpy(buf, &header, sizeof(PacketHeader));

    // ④ 复制 Username
    memcpy(buf + sizeof(PacketHeader), username, username_len);

    // ⑤ 填充 BodyLen(2B)& 复制 Body
    uint16_t net_body_len = htons((uint16_t)body_len);
    size_t offset_bodylen = sizeof(PacketHeader) + username_len;
    memcpy(buf + offset_bodylen, &net_body_len, sizeof(uint16_t));
    // 复制消息正文
    memcpy(buf + offset_bodylen + sizeof(uint16_t), message, body_len);

    // ⑥ 计算 CRC32 并填充(覆盖最后 4B)
    uint32_t crc = crc32(0L, Z_NULL, 0);
    crc = crc32(crc, buf, (uInt)(total_len - 4));            // 不包含最后 4B
    uint32_t net_crc = htonl(crc);
    memcpy(buf + total_len - 4, &net_crc, sizeof(uint32_t));

    // ⑦ 通过 socket 发送
    ssize_t sent = send(sockfd, buf, total_len, 0);
    free(buf);
    if (sent != (ssize_t)total_len) {
        return -1;
    }
    return 0;
}
  • zlib 中的 crc32() 可以快速计算 CRC32 校验码;
  • 注意所有整数字段都要使用 htons / htonl 转换为网络字节序;
  • 发送端没有拆包问题,因为我们只 send() 一次 buf,在网络层会尽量保证原子性(如果 total\_len < TCP 最大报文长度,一般不会被拆分)。

5.3 示例代码:打包与发送(整合版)

#include <arpa/inet.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <zlib.h>
#include <stdint.h>

#pragma pack(push, 1)
typedef struct {
    uint16_t magic;      // 2B
    uint8_t  version;    // 1B
    uint8_t  msg_type;   // 1B
    uint32_t msg_seq;    // 4B
    uint8_t  user_len;   // 1B
} PacketHeader;
#pragma pack(pop)

#define MAGIC_NUMBER 0xABCD
#define PROTOCOL_VERSION 0x01

// 返回 0 成功,-1 失败
int send_text_message(int sockfd, const char *username, const char *message, uint32_t seq) {
    size_t username_len = strlen(username);
    size_t body_len     = strlen(message);

    if (username_len > 255 || body_len > 65535) {
        return -1;
    }

    size_t total_len = sizeof(PacketHeader) + username_len + 2 + body_len + 4;
    uint8_t *buf = (uint8_t *)malloc(total_len);
    if (!buf) return -1;

    PacketHeader header;
    header.magic    = htons(MAGIC_NUMBER);
    header.version  = PROTOCOL_VERSION;
    header.msg_type = 1; // 文本消息
    header.msg_seq  = htonl(seq);
    header.user_len = (uint8_t)username_len;

    memcpy(buf, &header, sizeof(PacketHeader));
    memcpy(buf + sizeof(PacketHeader), username, username_len);

    uint16_t net_body_len = htons((uint16_t)body_len);
    size_t offset_bodylen = sizeof(PacketHeader) + username_len;
    memcpy(buf + offset_bodylen, &net_body_len, sizeof(uint16_t));
    memcpy(buf + offset_bodylen + sizeof(uint16_t), message, body_len);

    // 计算 CRC32(不包含最后 4B),并写入末尾
    uint32_t crc = crc32(0L, Z_NULL, 0);
    crc = crc32(crc, buf, (uInt)(total_len - 4));
    uint32_t net_crc = htonl(crc);
    memcpy(buf + total_len - 4, &net_crc, sizeof(uint32_t));

    ssize_t sent = send(sockfd, buf, total_len, 0);
    free(buf);
    return (sent == (ssize_t)total_len) ? 0 : -1;
}

完整打包过程:

  1. 准备 Header
  2. 复制 Username
  3. 填充 BodyLen & 复制 Body
  4. 计算并填充 Checksum
  5. 调用 send() 发送整条消息

6. 反序列化实现详解(接收端)

在网络接收端,由于 TCP 是面向字节流的协议,不保证一次 recv() 就能读到完整的一条消息,因此必须按照“包头定长 + 拆包”原则:

  1. 先读定长包头(这里是 2B + 1B + 1B + 4B + 1B = 9B);
  2. 解析包头字段,计算用户名长度与正文长度
  3. 按需 recv 余下的 “用户名 + BodyLen(2B) + Body”
  4. 最后再 recv Checksum(4B)
  5. 校验 CRC,若一致则处理业务,否则丢弃

6.1 读到原始字节流后的分包逻辑

+=======================+
| TCP Stream (字节流)   |
+=======================+
| <- recv(9) ->         | // 先读取固定 9 字节 Header
|                       |
| <- recv(username_len) ->  // 再读取 用户名
|                       |
| <- recv(2) ->         | // 读取 body_len
|                       |
| <- recv(body_len) ->  // 读取正文
|                       |
| <- recv(4) ->         | // 读取 Checksum
|                       |
|  ...                  | // 下一个消息的头部或下一个粘包
+=======================+
  • 注意:

    • 如果一次 recv() 未读满 9 字节,需要循环 recv 直到凑够;
    • 同理,对于 username_lenbody_lenchecksum 的读取都需要循环直到拿够指定字节数。
    • 若中途 recv() 返回 0,说明对端正常关闭;若返回 <0errno != EAGAIN && errno != EWOULDBLOCK,是错误,需要关闭连接。

6.2 解析头部与有效载荷

处理思路如下:

  1. 读取 Header(9B)

    • 使用一个大小为 9 字节的临时缓冲区 uint8_t head_buf[9]
    • 不断调用 n = recv(sockfd, head_buf + already_read, 9 - already_read, 0),直到已读 9 字节;
  2. head_buf 解析字段

    uint16_t magic  = ntohs(*(uint16_t *)(head_buf + 0));
    uint8_t  version= *(uint8_t  *)(head_buf + 2);
    uint8_t  msg_type= *(uint8_t *)(head_buf + 3);
    uint32_t msg_seq = ntohl(*(uint32_t *)(head_buf + 4));
    uint8_t  user_len = *(uint8_t *)(head_buf + 8);
    • 如果 magic != 0xABCDversion != 0x01,应拒绝或丢弃;
  3. 读取 Username(user\_len 字节)

    • 分配 char *username = malloc(user_len + 1)
    • 循环 recv 直到 user_len 字节读完;最后补 username[user_len] = '\0'
  4. 读取正文长度(2B)

    • 分配 uint8_t bodylen_buf[2];循环 recv 直到读满 2 字节;
    • uint16_t body_len = ntohs(*(uint16_t *)bodylen_buf);
  5. 读取正文(body\_len 字节)

    • 分配 char *body = malloc(body_len + 1)
    • 循环 recv 直到 body_len 字节读完;最后补 body[body_len] = '\0'
  6. 读取并校验 Checksum(4B)

    • 分配 uint8_t checksum_buf[4];循环 recv 直到读满 4 字节;
    • uint32_t recv_crc = ntohl(*(uint32_t *)checksum_buf);
    • 重新计算:crc32(0L, Z_NULL, 0)

      crc = crc32(crc, head_buf, 9);
      crc = crc32(crc, (const Bytef *)username, user_len);
      crc = crc32(crc, bodylen_buf, 2);
      crc = crc32(crc, (const Bytef *)body, body_len);
    • 如果 crc != recv_crc,则数据损坏,丢弃并断开连接或回复“协议错误”;

6.3 示例代码:接收与解析

#include <arpa/inet.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <zlib.h>
#include <stdint.h>
#include <stdio.h>
#include <errno.h>

#pragma pack(push, 1)
typedef struct {
    uint16_t magic;
    uint8_t  version;
    uint8_t  msg_type;
    uint32_t msg_seq;
    uint8_t  user_len;
} PacketHeader;
#pragma pack(pop)

#define MAGIC_NUMBER 0xABCD
#define PROTOCOL_VERSION 0x01

/**
 * 从 socket 中读取指定字节数到 buf(循环 recv)
 * @param sockfd 已连接 socket
 * @param buf    目标缓冲区
 * @param len    需要读取的字节数
 * @return int   读取成功返回 0;对端关闭或出错返回 -1
 */
int recv_nbytes(int sockfd, void *buf, size_t len) {
    size_t  left = len;
    ssize_t n;
    uint8_t *ptr = (uint8_t *)buf;

    while (left > 0) {
        n = recv(sockfd, ptr, left, 0);
        if (n == 0) {
            // 对端关闭
            return -1;
        } else if (n < 0) {
            if (errno == EINTR) continue; // 被信号中断,重试
            return -1;                   // 其他错误
        }
        ptr  += n;
        left -= n;
    }
    return 0;
}

/**
 * 处理一条消息:读取并解析
 * @param sockfd  已连接 socket
 * @return int    0=成功处理,-1=出错或对端关闭
 */
int handle_one_message(int sockfd) {
    PacketHeader header;
    // 1. 读取 Header (9B)
    if (recv_nbytes(sockfd, &header, sizeof(PacketHeader)) < 0) {
        return -1;
    }

    uint16_t magic = ntohs(header.magic);
    if (magic != MAGIC_NUMBER) {
        fprintf(stderr, "协议魔数错误: 0x%04x\n", magic);
        return -1;
    }
    if (header.version != PROTOCOL_VERSION) {
        fprintf(stderr, "协议版本不匹配: %d\n", header.version);
        return -1;
    }
    uint8_t msg_type = header.msg_type;
    uint32_t msg_seq = ntohl(header.msg_seq);
    uint8_t user_len = header.user_len;

    // 2. 读取 Username
    char *username = (char *)malloc(user_len + 1);
    if (!username) return -1;
    if (recv_nbytes(sockfd, username, user_len) < 0) {
        free(username);
        return -1;
    }
    username[user_len] = '\0';

    // 3. 读取 BodyLen (2B)
    uint16_t net_body_len;
    if (recv_nbytes(sockfd, &net_body_len, sizeof(uint16_t)) < 0) {
        free(username);
        return -1;
    }
    uint16_t body_len = ntohs(net_body_len);

    // 4. 读取 Body
    char *body = (char *)malloc(body_len + 1);
    if (!body) {
        free(username);
        return -1;
    }
    if (recv_nbytes(sockfd, body, body_len) < 0) {
        free(username);
        free(body);
        return -1;
    }
    body[body_len] = '\0';

    // 5. 读取 Checksum (4B)
    uint32_t net_recv_crc;
    if (recv_nbytes(sockfd, &net_recv_crc, sizeof(uint32_t)) < 0) {
        free(username);
        free(body);
        return -1;
    }
    uint32_t recv_crc = ntohl(net_recv_crc);

    // 6. 校验 CRC32
    uLong crc = crc32(0L, Z_NULL, 0);
    crc = crc32(crc, (const Bytef *)&header, sizeof(PacketHeader));
    crc = crc32(crc, (const Bytef *)username, user_len);
    crc = crc32(crc, (const Bytef *)&net_body_len, sizeof(uint16_t));
    crc = crc32(crc, (const Bytef *)body, body_len);

    if ((uint32_t)crc != recv_crc) {
        fprintf(stderr, "CRC 校验失败: 0x%08x vs 0x%08x\n", (uint32_t)crc, recv_crc);
        free(username);
        free(body);
        return -1;
    }

    // 7. 处理业务逻辑
    if (msg_type == 1) {
        // 文本消息
        printf("收到消息 seq=%u, user=%s, body=%s\n", msg_seq, username, body);
        // …(后续可以回送 ACK、广播给其他客户端等)
    } else if (msg_type == 2) {
        // 心跳包
        printf("收到心跳,seq=%u, user=%s\n", msg_seq, username);
        // 可以直接发送一个心跳响应:msg_type=2, body_len=0
    } else {
        fprintf(stderr, "未知消息类型: %d\n", msg_type);
    }

    free(username);
    free(body);
    return 0;
}
  • 函数 recv_nbytes() 循环调用 recv(),确保“指定字节数”能被完全读取;
  • 按顺序读取:头部 → 用户名 → 正文长度 → 正文 → 校验码;
  • 校验 CRC32、版本、魔数,若不通过即舍弃该条消息;
  • 根据 msg_type 做业务分发。

7. 实战:完整客户端与服务器示例

为了进一步巩固上述原理,本节给出一个简易客户端与服务器的完整示例。

  • 服务器:监听某端口,循环 accept() 新连接,每个连接启动一个子线程/子进程(或使用 IO 多路复用),负责调用 handle_one_message() 读取并解析客户端发来的每一条消息;
  • 客户端:读取终端输入(用户名 + 消息),调用 send_text_message() 将消息打包并发到服务器;每隔 30 秒如果没有输入,主动发送心跳包。
注意:为了简化代码示例,本处采用“单线程 + 阻塞 I/O + select”来监听客户端连接,实际生产可用 epoll/kqueue/IOCP 等。

7.1 服务器端实现要点

  1. 创建监听 socketbind() + listen()
  2. 进入主循环

    • 使用 select()poll() 监听 listen_fd 与所有客户端 conn_fd[]
    • 如果 listen_fd 可读,则 accept() 新连接,并加入 conn_fd 集合;
    • 如果 conn_fd 可读,则调用 handle_one_message(conn_fd);若返回 -1,关闭该 conn_fd
  3. 心跳响应:若遇到 msg_type == 2,可在 handle_one_message 里直接构造一个空心跳响应包(msg_type=2, username="", body_len=0),通过 send() 返还给客户端。
// 省略常见头文件与辅助函数(如 send_text_message, handle_one_message, recv_nbytes 等)
// 下面给出核心的服务器主循环(使用 select)

#define SERVER_PORT 8888
#define MAX_CLIENTS  FD_SETSIZE  // select 限制

int main() {
    int listen_fd, max_fd, i;
    int client_fds[MAX_CLIENTS];
    struct sockaddr_in serv_addr, cli_addr;
    fd_set all_set, read_set;

    // 1. 创建监听套接字
    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_fd < 0) { perror("socket"); exit(1); }
    int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family      = AF_INET;
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_addr.sin_port        = htons(SERVER_PORT);
    bind(listen_fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
    listen(listen_fd, 10);

    // 2. 初始化客户端数组
    for (i = 0; i < MAX_CLIENTS; i++) client_fds[i] = -1;

    max_fd = listen_fd;
    FD_ZERO(&all_set);
    FD_SET(listen_fd, &all_set);

    printf("服务器启动,监听端口 %d\n", SERVER_PORT);

    while (1) {
        read_set = all_set;
        int nready = select(max_fd + 1, &read_set, NULL, NULL, NULL);
        if (nready < 0) { perror("select"); break; }

        // 3. 监听套接字可读:新连接
        if (FD_ISSET(listen_fd, &read_set)) {
            socklen_t cli_len = sizeof(cli_addr);
            int conn_fd = accept(listen_fd, (struct sockaddr *)&cli_addr, &cli_len);
            if (conn_fd < 0) {
                perror("accept");
                continue;
            }
            printf("新客户端:%s:%d, fd=%d\n", inet_ntoa(cli_addr.sin_addr),
                   ntohs(cli_addr.sin_port), conn_fd);

            // 加入 client_fds
            for (i = 0; i < MAX_CLIENTS; i++) {
                if (client_fds[i] < 0) {
                    client_fds[i] = conn_fd;
                    break;
                }
            }
            if (i == MAX_CLIENTS) {
                printf("已达最大客户端数,拒绝连接 fd=%d\n", conn_fd);
                close(conn_fd);
            } else {
                FD_SET(conn_fd, &all_set);
                if (conn_fd > max_fd) max_fd = conn_fd;
            }
            if (--nready <= 0) continue;
        }

        // 4. 遍历现有客户端,处理可读事件
        for (i = 0; i < MAX_CLIENTS; i++) {
            int sockfd = client_fds[i];
            if (sockfd < 0) continue;
            if (FD_ISSET(sockfd, &read_set)) {
                // 处理一条消息
                if (handle_one_message(sockfd) < 0) {
                    // 发生错误或对端关闭
                    close(sockfd);
                    FD_CLR(sockfd, &all_set);
                    client_fds[i] = -1;
                }
                if (--nready <= 0) break;
            }
        }
    }

    // 清理
    for (i = 0; i < MAX_CLIENTS; i++) {
        if (client_fds[i] >= 0) close(client_fds[i]);
    }
    close(listen_fd);
    return 0;
}
  • 整个服务器进程在单线程中通过 select 监听 多个客户端套接字
  • 对于每个就绪的客户端 sockfd,调用 handle_one_message 完整地“读取并解析”一条消息;
  • 如果解析过程出错(协议不对、CRC 校验失败、对端关闭等),立即关闭对应连接并在 select 集合中清理。

7.2 客户端实现要点

  1. 连接服务器socket()connect()
  2. 读取用户输入:先读取“用户名”(一次即可),然后进入循环:

    • 如果标准输入有文本,则构造文本消息并调用 send_text_message()
    • 如果 30 秒内未输入任何信息,则构造心跳包并发送;
    • 同时 select 监听服务器回送的数据(如心跳响应或其他提醒)。
  3. 心跳包构造:与文本消息类似,只不过:

    • msg_type = 2
    • user_len = 用户名长度
    • body_len = 0
    • Checksum 同样需要计算。
#include <arpa/inet.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>
#include <zlib.h>
#include <stdint.h>

#pragma pack(push, 1)
typedef struct {
    uint16_t magic;
    uint8_t  version;
    uint8_t  msg_type;
    uint32_t msg_seq;
    uint8_t  user_len;
} PacketHeader;
#pragma pack(pop)

#define MAGIC_NUMBER 0xABCD
#define PROTOCOL_VERSION 0x01

/**
 * 构造并发送心跳包
 */
int send_heartbeat(int sockfd, const char *username, uint32_t seq) {
    size_t username_len = strlen(username);

    // total_len = Header(9B) + username + bodylen(2B, 0) + checksum(4B)
    size_t total_len = sizeof(PacketHeader) + username_len + 2 + 0 + 4;
    uint8_t *buf = (uint8_t *)malloc(total_len);
    if (!buf) return -1;

    PacketHeader header;
    header.magic    = htons(MAGIC_NUMBER);
    header.version  = PROTOCOL_VERSION;
    header.msg_type = 2; // 心跳
    header.msg_seq  = htonl(seq);
    header.user_len = username_len;

    memcpy(buf, &header, sizeof(PacketHeader));
    memcpy(buf + sizeof(PacketHeader), username, username_len);

    // BodyLen = 0
    uint16_t net_body_len = htons((uint16_t)0);
    size_t offset_bodylen = sizeof(PacketHeader) + username_len;
    memcpy(buf + offset_bodylen, &net_body_len, sizeof(uint16_t));
    // 没有 Body

    // 计算 CRC32(不包含最后 4B)
    uLong crc = crc32(0L, Z_NULL, 0);
    crc = crc32(crc, buf, (uInt)(total_len - 4));
    uint32_t net_crc = htonl((uint32_t)crc);
    memcpy(buf + total_len - 4, &net_crc, sizeof(uint32_t));

    ssize_t sent = send(sockfd, buf, total_len, 0);
    free(buf);
    return (sent == (ssize_t)total_len) ? 0 : -1;
}

int send_text_message(int sockfd, const char *username, const char *message, uint32_t seq) {
    size_t username_len = strlen(username);
    size_t body_len     = strlen(message);
    if (username_len > 255 || body_len > 65535) return -1;

    size_t total_len = sizeof(PacketHeader) + username_len + 2 + body_len + 4;
    uint8_t *buf = (uint8_t *)malloc(total_len);
    if (!buf) return -1;

    PacketHeader header;
    header.magic    = htons(MAGIC_NUMBER);
    header.version  = PROTOCOL_VERSION;
    header.msg_type = 1; // 文本
    header.msg_seq  = htonl(seq);
    header.user_len = (uint8_t)username_len;

    memcpy(buf, &header, sizeof(PacketHeader));
    memcpy(buf + sizeof(PacketHeader), username, username_len);

    uint16_t net_body_len = htons((uint16_t)body_len);
    size_t offset_bodylen = sizeof(PacketHeader) + username_len;
    memcpy(buf + offset_bodylen, &net_body_len, sizeof(uint16_t));
    memcpy(buf + offset_bodylen + sizeof(uint16_t), message, body_len);

    uLong crc = crc32(0L, Z_NULL, 0);
    crc = crc32(crc, buf, (uInt)(total_len - 4));
    uint32_t net_crc = htonl((uint32_t)crc);
    memcpy(buf + total_len - 4, &net_crc, sizeof(uint32_t));

    ssize_t sent = send(sockfd, buf, total_len, 0);
    free(buf);
    return (sent == (ssize_t)total_len) ? 0 : -1;
}

int recv_nbytes(int sockfd, void *buf, size_t len);

int handle_one_message(int sockfd) {
    // 同服务器端 handle_one_message 函数,可参考上文,这里略去
    return 0;
}

int main(int argc, char *argv[]) {
    if (argc != 4) {
        printf("Usage: %s <server_ip> <server_port> <username>\n", argv[0]);
        return -1;
    }
    const char *server_ip   = argv[1];
    int         server_port = atoi(argv[2]);
    const char *username    = argv[3];
    size_t      username_len= strlen(username);
    if (username_len == 0 || username_len > 255) {
        printf("用户名长度需在 1~255 之间\n");
        return -1;
    }

    // 1. 连接服务器
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    struct sockaddr_in serv_addr;
    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family      = AF_INET;
    serv_addr.sin_port        = htons(server_port);
    inet_pton(AF_INET, server_ip, &serv_addr.sin_addr);
    if (connect(sockfd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {
        perror("connect");
        return -1;
    }
    printf("已连接服务器 %s:%d,用户名=%s\n", server_ip, server_port, username);

    // 2. 设置 sockfd、stdin 为非阻塞,以便同时监听用户输入与服务器回复
    int flags = fcntl(sockfd, F_GETFL, 0);
    fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
    flags = fcntl(STDIN_FILENO, F_GETFL, 0);
    fcntl(STDIN_FILENO, F_SETFL, flags | O_NONBLOCK);

    fd_set read_set;
    uint32_t seq = 0;
    time_t last_send_time = time(NULL);

    while (1) {
        FD_ZERO(&read_set);
        FD_SET(sockfd, &read_set);
        FD_SET(STDIN_FILENO, &read_set);
        int max_fd = sockfd > STDIN_FILENO ? sockfd : STDIN_FILENO;

        struct timeval timeout;
        timeout.tv_sec  = 1;  // 每秒检查一次是否需要心跳
        timeout.tv_usec = 0;

        int nready = select(max_fd + 1, &read_set, NULL, NULL, &timeout);
        if (nready < 0) {
            if (errno == EINTR) continue;
            perror("select");
            break;
        }

        // 3. 检查服务器回送
        if (FD_ISSET(sockfd, &read_set)) {
            // 这里可以用 handle_one_message 解析服务器消息
            handle_one_message(sockfd);
        }

        // 4. 检查用户输入
        if (FD_ISSET(STDIN_FILENO, &read_set)) {
            char input_buf[1024];
            ssize_t n = read(STDIN_FILENO, input_buf, sizeof(input_buf) - 1);
            if (n > 0) {
                input_buf[n] = '\0';
                // 去掉换行
                if (input_buf[n - 1] == '\n') input_buf[n - 1] = '\0';

                if (strlen(input_buf) > 0) {
                    // 发文本消息
                    send_text_message(sockfd, username, input_buf, seq++);
                    last_send_time = time(NULL);
                }
            }
        }

        // 5. 检查是否超过 30 秒未发送消息,需要发心跳
        time_t now = time(NULL);
        if (now - last_send_time >= 30) {
            send_heartbeat(sockfd, username, seq++);
            last_send_time = now;
        }
    }

    close(sockfd);
    return 0;
}
  • 客户端在主循环中同时监听 sockfd(服务器推送)与 STDIN_FILENO(用户输入),通过 select 实现非阻塞地“同时等待”两种事件;
  • 如果 30 秒内没有新的用户输入,则发送一次心跳包;
  • handle_one_message() 负责处理服务器的任何回包,包括心跳响应、其他用户的消息通知等。

7.3 示意图:客户端 ↔ 服务器 流程

Client                                      Server
  |---------------- TCP Connect ----------->|
  |                                         |
  |-- send "Hello, World!" as Text Message->|
  |                                         |  recv Header(9B) -> parse (msg_type=1)
  |                                         |  recv UsernameLen & Username
  |                                         |  recv BodyLen & Body
  |                                         |  recv Checksum -> 校验
  |                                         |  打印 “收到消息 user=..., body=...”
  |                                         |  (如需ACK,可自定义回应)
  |<------------ recv  Heartbeat Response--|
  |                                         |
  |-- (30s超时) send Heartbeat ------------>|
  |                                         |  recv Header -> parse(msg_type=2)
  |                                         |  心跳解析完成 -> 立即 构造心跳响应
  |<------------ send 心跳响应 -------------|
  |                                         |
  | ...                                     |
  1. 连接阶段:客户端 connect() → 服务器 accept()
  2. 消息阶段:客户端使用 send_text_message() 打包“文本消息”,服务器 recv 分段读取并解析后打印;
  3. 心跳阶段:若客户端 30 秒内无数据,则调用 send_heartbeat(),服务器收到后直接构造心跳响应;
  4. 双向心跳:服务器发送心跳响应,客户端在 select 中收到后也可以计算“服务器在线”,若超时可自行重连。

8. 常见注意事项与优化建议

8.1 网络不定长包的处理

  • TCP 粘包/拆包:TCP 并不保证一次 send() 对应一次 recv()

    • 可能在发送端发出一条 100B 的消息,接收端会在两次 recv(60) + recv(40) 中获取完整内容;
    • 也可能两条小消息“粘在”一起,从一次 recv(200) 一次性读到。

解决措施

  1. 先读固定长度包头:用 recv_nbytes(..., 9);即便数据还没完全到达,该函数也会循环等待,直到完整;
  2. 根据包头中的长度字段:再去读 username\_len、body\_len、checksum 等,不多读也不少读;
  3. 对粘包:假设一口气读到了 2 条或多条消息的头,recv_nbytes() 只负责“把头部先读满”,之后通过“剩余字节”继续循环解析下一条消息;

示意:两条消息粘在一起

TCP 接收缓冲区:
+-----------------------------------------------------------+
| [Msg1: Header + Username + Body + Crc] [Msg2: Header + ... |
+-----------------------------------------------------------+

recv_nbytes(sockfd, head_buf, 9); // 先将 Msg1 的头部 9B 读出
parse 出 user_len, body_len 后,继续 recv 剩余 Msg1
读取完成 Msg1 后,缓冲区中还有 Msg2

下一次调用 recv_nbytes(sockfd, head_buf, 9),会立刻从 Msg2 读数据,不会等待

8.2 缓冲区管理与内存对齐

  • 手动内存管理:示例中用 malloc()/free() 来管理 Username 与 Body 缓冲区,

    • 若并发连接数多,应考虑使用 缓冲池(Buffer Pool)避免频繁 malloc/free 的性能开销;
  • 字节对齐#pragma pack(1) 确保了 Header 结构不含填充字节,但若部分字段超过 1 字节应谨慎使用字节指针计算偏移,

    • 推荐定义常量偏移,如 offset_username = sizeof(PacketHeader),避免“魔法数字”;
  • 栈 vs 堆:Header 结构可放在栈上 PacketHeader header;;对于 Username/Body 大小在几 KB 范围内,可考虑栈上局部数组 char buf[4096],并手动控制偏移。但若长度可达数十 KB,需放到堆。

8.3 心跳包与超时重连机制

  • 客户端每隔 T 秒发送一次心跳,保证服务器知道客户端在线;
  • 服务器也可以向客户端周期性发送心跳,客户端可用来检测“服务器断线”;
  • 超时判断:如果某方连续 N 次未收到对方心跳,则判定“对方已下线/掉线”,并关闭连接或尝试重连;
  • 心跳频率:既要低于业务消息频率,避免过度消耗带宽;又要保证足够频繁,一旦断连能及时发现。

8.4 使用高层序列化库(Protobuf/FlatBuffers)简介

  • 如果业务场景不希望手写“渐进式序列化与反序列化”,也可考虑使用Google Protocol Buffers(Protobuf)FlatBuffersCap’n Proto 等成熟方案;
  • 优点:自动生成代码,支持多语言,内置版本兼容、校验、压缩等;
  • 缺点:引入额外依赖,生成代码体积较大,性能和灵活度略逊于自定义二进制协议;

示例(Protobuf):

syntax = "proto3";
package chat;

// 文本消息
message TextMsg {
  uint32 seq       = 1;
  string username  = 2;
  string body      = 3;
}

// 心跳包
message Heartbeat {
  uint32 seq       = 1;
  string username  = 2;
}

// 顶层消息(用于包含不同类型)
message ChatPacket {
  oneof payload {
    TextMsg    txt_msg   = 1;
    Heartbeat  hb_msg    = 2;
  }
}
  • 然后用 protoc --cpp_out=. / protoc --csharp_out=. 等指令生成对应语言的序列化/反序列化代码;
  • 发送端只需 ChatPacket packet; packet.set_txt_msg(...); packet.SerializeToArray(buf, size); send(sockfd, buf, size, 0);
  • 接收端只需读取长度字段、RecvBytes(...) 得到完整二进制,再 packet.ParseFromArray(buf, size);

若对手工实现的协议维护成本较高,可考虑切换到 Protobuf。但对于轻量级、极低延迟的场景,自定义协议往往能获取更好的性能。


9. 总结

本文以“简易聊天协议”为例,详细讲解了在 Linux C 网络编程中,如何:

  1. 设计自定义二进制协议,包含包头、变长字段、可选校验;
  2. 序列化(发送端):手动打包 Header、字段、正文,并做网络字节序转换与 CRC 校验,保证数据在网络中可靠传输;
  3. 反序列化(接收端):先 recv 定长头部,解析长度信息,再循环读取后续可变长字段,最后校验 CRC 后交由业务逻辑;
  4. 完整示例:给出了服务器与客户端完整架构,展示了如何在 单线程 + select 的框架下同时兼顾 文本消息 与 心跳包;
  5. 常见注意事项:对 TCP 粘包/拆包、缓冲区管理、心跳超时、字节对齐等细节进行了深入分析,并简要介绍了高层序列化库的取舍。

掌握自定义协议与手动序列化/反序列化,不仅能帮助你在轻量、高性能场景下游刃有余,还能让你更深刻地理解底层网络编程原理。当你以后需要针对特定业务做更灵活的定制时,这套技术栈无疑是核心能力之一。


后续拓展

  1. epollkqueue 优化多连接性能;
  2. 增加 加密(如 AES-CBC)与混淆,保障传输安全;
  3. 将心跳改为“异步 I/O + 定时器”架构;
  4. 在消息体中引入二进制文件分片传输,实现大文件断点续传。

图解回顾

  • 协议整体结构:Header → Username → BodyLen → Body → Checksum
  • TCP 粘包/拆包处理流程:先定长读头 → 根据长度再读变长 → 校验 → 处理 → 继续下一条
  • 客户端/服务器交互示意:文本消息与心跳包双向穿插。
2024-11-30

Socket TCP 和 UDP 编程基础(Python)

网络编程是现代计算机科学的重要组成部分,而 socket 是 Python 提供的一个模块,用于实现网络通信。本文将详细介绍 TCP 和 UDP 两种通信协议的基础知识,配合 Python 代码示例和图解,让你轻松上手。


一、TCP 和 UDP 概述

1.1 TCP 协议

  • 特点:面向连接、可靠传输、适合数据量大且顺序重要的场景。
  • 应用:网页浏览、文件传输。
  • 通信模型

    • 客户端发起连接。
    • 服务端接受连接。
    • 双方建立可靠的数据传输通道。

1.2 UDP 协议

  • 特点:无连接、快速、不保证可靠传输。
  • 应用:视频直播、在线游戏。
  • 通信模型

    • 无需建立连接。
    • 数据直接发送到目标。

二、Python Socket 基础

socket 模块是 Python 提供的网络通信接口,可以实现 TCP 和 UDP 通信。

2.1 常用方法

方法说明
socket()创建一个新的 socket 对象
bind(address)绑定到指定的 IP 和端口
listen(backlog)在 TCP 中监听连接
accept()接受客户端连接
connect(address)客户端连接到服务器
send(data)发送数据
recv(buffer_size)接收数据
sendto(data, addr)向指定地址发送数据(UDP)
recvfrom(buffer_size)接收数据(UDP)
close()关闭 socket

三、TCP 编程

3.1 TCP 服务端代码

import socket

# 创建 TCP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("127.0.0.1", 65432))  # 绑定 IP 和端口
server_socket.listen(5)  # 最大连接数

print("服务端已启动,等待连接...")
conn, addr = server_socket.accept()  # 等待客户端连接
print(f"连接来自: {addr}")

# 接收并发送数据
data = conn.recv(1024).decode("utf-8")
print(f"收到客户端消息: {data}")
conn.send("你好,客户端!".encode("utf-8"))

# 关闭连接
conn.close()
server_socket.close()

3.2 TCP 客户端代码

import socket

# 创建 TCP socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(("127.0.0.1", 65432))  # 连接服务器

# 发送数据
client_socket.send("你好,服务端!".encode("utf-8"))

# 接收数据
data = client_socket.recv(1024).decode("utf-8")
print(f"收到服务端消息: {data}")

# 关闭连接
client_socket.close()

运行结果

  • 服务端输出

    服务端已启动,等待连接...
    连接来自: ('127.0.0.1', 50000)
    收到客户端消息: 你好,服务端!
  • 客户端输出

    收到服务端消息: 你好,客户端!

TCP 通信模型图解

客户端               服务端
  |   connect()       |
  | ----------------> |
  |                   |
  |     accept()       |
  | <---------------- |
  |                   |
  |     send()         |
  | ----------------> |
  |                   |
  |     recv()         |
  | <---------------- |
  |                   |

四、UDP 编程

4.1 UDP 服务端代码

import socket

# 创建 UDP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind(("127.0.0.1", 65432))  # 绑定 IP 和端口
print("UDP 服务端已启动,等待数据...")

# 接收数据
data, addr = server_socket.recvfrom(1024)
print(f"收到来自 {addr} 的消息: {data.decode('utf-8')}")

# 发送响应
server_socket.sendto("你好,客户端!".encode("utf-8"), addr)

server_socket.close()

4.2 UDP 客户端代码

import socket

# 创建 UDP socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 发送数据
client_socket.sendto("你好,服务端!".encode("utf-8"), ("127.0.0.1", 65432))

# 接收响应
data, addr = client_socket.recvfrom(1024)
print(f"收到服务端消息: {data.decode('utf-8')}")

client_socket.close()

运行结果

  • 服务端输出

    UDP 服务端已启动,等待数据...
    收到来自 ('127.0.0.1', 50000) 的消息: 你好,服务端!
  • 客户端输出

    收到服务端消息: 你好,客户端!

UDP 通信模型图解

客户端               服务端
  |   sendto()        |
  | ----------------> |
  |                   |
  |     recvfrom()     |
  | <---------------- |

五、TCP 与 UDP 的区别

特性TCPUDP
是否连接面向连接无连接
数据可靠性高,保证数据按顺序到达不可靠,不保证顺序
速度较慢,需建立连接和确认快,无需连接
应用场景文件传输、网页浏览视频流、在线游戏

六、综合练习:多人聊天室

服务端代码

import socket
import threading

# 创建服务端
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("127.0.0.1", 65432))
server_socket.listen(5)
clients = []

def handle_client(conn, addr):
    print(f"新客户端加入: {addr}")
    while True:
        try:
            data = conn.recv(1024).decode("utf-8")
            if not data:
                break
            print(f"{addr}: {data}")
            for client in clients:
                if client != conn:
                    client.send(f"{addr}: {data}".encode("utf-8"))
        except:
            break
    conn.close()
    clients.remove(conn)

print("聊天室启动中...")
while True:
    conn, addr = server_socket.accept()
    clients.append(conn)
    threading.Thread(target=handle_client, args=(conn, addr)).start()

客户端代码

import socket
import threading

def receive_messages(client_socket):
    while True:
        try:
            message = client_socket.recv(1024).decode("utf-8")
            print(message)
        except:
            print("连接已关闭")
            break

client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(("127.0.0.1", 65432))

threading.Thread(target=receive_messages, args=(client_socket,)).start()

print("加入聊天室!输入消息开始聊天...")
while True:
    msg = input()
    client_socket.send(msg.encode("utf-8"))

七、总结

通过本文,你应该已经掌握了 Python 中基于 socket 模块进行 TCP 和 UDP 编程的基本方法。无论是构建简单的客户端-服务端模型,还是开发复杂的网络应用,都可以从这些基础知识入手。希望你通过练习能够熟练运用这些技术!

2024-09-09

解释:

这个错误通常表明Redis服务器试图绑定到端口6379时出现问题,但是没有提供具体的错误信息。在Unix-like系统中,这可能意味着端口已经被占用,或者是权限问题。

解决方法:

  1. 检查端口是否已被占用:

    执行 netstat -tulnp | grep 6379 查看端口是否被其他进程占用。

  2. 如果端口被占用,你需要找到占用进程并终止它,或者更改Redis配置文件中的端口号。
  3. 如果是权限问题,确保Redis进程有权限绑定到所需的端口。通常,低于1024的端口需要特权用户(如root)才能绑定。你可以尝试使用更高的端口号,或者以root用户运行Redis服务。
  4. 检查Redis配置文件(通常名为redis.conf),确保bind指令(如果有的话)不会阻止绑定到所需接口。
  5. 查看Redis日志文件,它可能包含更多关于错误的信息。
  6. 如果问题依然存在,可以尝试重启服务器,然后再次启动Redis服务。
  7. 如果你在云环境中,确保安全组或网络策略允许对应端口的流量。
  8. 如果上述步骤都不能解决问题,可以尝试重新安装Redis,并确保按照官方文档进行配置。
2024-09-06

在Ubuntu 22.04上安装和使用Libmodbus库的步骤如下:

  1. 安装libmodbus库:



sudo apt-update
sudo apt-get install libmodbus-dev
  1. 编写一个简单的Modbus TCP客户端示例代码:



#include <stdio.h>
#include <modbus/modbus.h>
 
int main() {
    int rc;
    modbus_t *ctx = modbus_new_tcp("127.0.0.1", 502);
    if (ctx == NULL) {
        fprintf(stderr, "Unable to allocate libmodbus context\n");
        return -1;
    }
 
    if (modbus_connect(ctx) == -1) {
        fprintf(stderr, "Unable to connect: %s\n", modbus_strerror(errno));
        modbus_free(ctx);
        return -1;
    }
 
    uint16_t tab_reg[3];
 
    rc = modbus_read_registers(ctx, 0, 3, tab_reg);
    if (rc == -1) {
        fprintf(stderr, "%s\n", modbus_strerror(errno));
        return -1;
    }
 
    printf("Register 0: %d\nRegister 1: %d\n", tab_reg[0], tab_reg[1]);
 
    modbus_close(ctx);
    modbus_free(ctx);
 
    return 0;
}
  1. 编译示例代码:



gcc -o modbus-tcp-client modbus-tcp-client.c -lmodbus
  1. 运行客户端程序:



./modbus-tcp-client

确保Modbus服务器端在运行,并且客户端的IP地址和端口与服务器端的信息匹配。上述代码示例展示了如何创建一个简单的Modbus TCP客户端,连接到服务器,并读取3个寄存器的值。

2024-09-04

以下是一个简单的Spring Boot TCP服务器和客户端的示例。

TCP服务器端 (Spring Boot):

  1. 添加依赖 spring-boot-starter-integrationspring-boot-starter-webpom.xml
  2. 配置服务器端口和TCP连接工厂。
  3. 创建TCP连接接收器。
  4. 启动Spring Boot应用程序。



import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
import org.springframework.integration.ip.tcp.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.serializer.ByteArraySerializer;
 
@Configuration
public class TcpServerConfig {
 
    @Bean
    public TcpNetServerConnectionFactory serverConnectionFactory() {
        TcpNetServerConnectionFactory factory = new TcpNetServerConnectionFactory(12345);
        factory.setSerializer(new ByteArraySerializer());
        factory.setDeserializer(new ByteArraySerializer());
        return factory;
    }
 
    @Bean
    public TcpInboundGateway tcpInboundGateway() {
        TcpInboundGateway gateway = new TcpInboundGateway();
        gateway.setConnectionFactory(serverConnectionFactory());
        return gateway;
    }
 
    @ServiceActivator
    public void clientMessageHandler() {
        // 处理接收到的消息
    }
}

TCP客户端 (Spring Boot):

  1. 添加依赖 spring-boot-starter-integrationspring-boot-starter-webpom.xml
  2. 配置服务器地址和端口。
  3. 发送消息。
  4. 启动Spring Boot应用程序。



import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.ip.tcp.TcpOutboundGateway;
import org.springframework.integration.ip.tcp.TcpNetClientConnectionFactory;
import org.springframework.integration.ip.tcp.serializer.ByteArraySerializer;
 
@Configuration
public class TcpClientConfig {
 
    @Bean
    public TcpNetClientConnectionFactory clientConnectionFactory() {
        TcpNetClientConnectionFactory factory = new TcpNetClientConnectionFactory("localhost", 12345);
        factory.setSerializer(new ByteArraySerializer());
        factory.setDeserializer(new ByteArraySerializer());
        return factory;
    }
 
    @Bean
    @ServiceActivator
    public TcpOutboundGateway tcpOutboundGateway() {
        Tc
2024-09-04



import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
 
public class TCPServer {
 
    private int port;
 
    public TCPServer(int port) {
        this.port = port;
    }
 
    public void run() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline pipeline = ch.pipeline();
                     pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                     pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                     pipeline.addLast("decoder", new StringDecoder());
                     pipeline.addLast("encoder", new StringEncoder());
                     pipeline.addLast("handler", new TCPServerHandler());
                 }
             });
 
            ChannelFuture f = b.bind(port).sync();
            System.out.println("服务器启动,监听端口:" + port);
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
 
    public static void main(String[] args) throws Exception {
        TCPServer server = new TCPServer(8080);
        server.run();
    }
}

这个示例代码展示了如何使用Netty框架来创建一个基本的TCP长连接服务器。服务器监听8080端口,并使用LengthFieldBasedFrameDecoderLengthFieldPrepender处理帧解码和编码,确保消息边界被正确识别。同时,它展示了如何添加StringDecoderStringEncoder来处理字符串消息,以及如何自定义处理器TCPServerHandler来处理具体的业务逻辑。这个代码提供了一个简单的模板,可以根据具体需求进行扩展和修改。

2024-09-04



import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
 
public class MyMessageEncoder extends MessageToByteEncoder<MyMessage> {
 
    @Override
    protected void encode(ChannelHandlerContext ctx, MyMessage msg, ByteBuf out) throws Exception {
        // 将消息转换为字节,并写入ByteBuf
        out.writeInt(msg.getLength()); // 写入消息长度
        out.writeBytes(msg.getData()); // 写入消息数据
    }
}
 
class MyMessage {
    private int length;
    private byte[] data;
 
    // 构造函数、getter和setter省略
 
    public int getLength() {
        return length;
    }
 
    public byte[] getData() {
        return data;
    }
}

这个简单的示例展示了如何使用MessageToByteEncoder来编码自定义的消息对象。在这个例子中,我们假设MyMessage有一个整数字段表示数据的长度和一个字节数组表示实际的数据内容。在encode方法中,我们将消息的长度和数据写入ByteBuf中,以便它可以被发送。这个编码器可以被用在Netty的编码器链中,与LengthFieldPrepender一起用于解决TCP粘包和拆包问题。

2024-09-04

在Oracle数据库网络配置中,DCD(Dynamic Connections Directory)是一种动态监听器注册机制,可以使数据库客户端在连接时间较长时,监听器能够自动识别到数据库实例的存在。Linux TCP KeepAlive是一种TCP/IP的网络配置选项,可以在网络连接空闲时发送保活包,以防止连接超时。而硬件防火墙通常用于保护网络边界免受未授权访问。

以下是这三者可能的配置方式和相关的代码示例:

  1. DCD配置:

    listener.ora文件中配置DCD,如下所示:




LISTENER =
  (DESCRIPTION_LIST =
    (DESCRIPTION =
      (ADDRESS = (PROTOCOL = TCP)(HOST = myhost.example.com)(PORT = 1521))
      (ADDRESS = (PROTOCOL = IPC)(KEY = EXTPROC1521))
    )
  )
 
SID_LIST_LISTENER =
  (SID_LIST =
    (SID_DESC =
      (GLOBAL_DBNAME = mydb.example.com)
      (ORACLE_HOME = /u01/app/oracle/product/11.2.0/db_1)
      (SID_NAME = mydb)
    )
    (SID_DESC =
      (SID_NAME = plsextproc)
      (ORACLE_HOME = /u01/app/oracle/product/11.2.0/db_1)
      (PROGRAM = extproc)
    )
  )
  1. Linux TCP KeepAlive配置:

    /etc/sysctl.conf文件中配置TCP KeepAlive参数,如下所示:




# 启用TCP KeepAlive
net.ipv4.tcp_keepalive_time = 30
# 每个包的发送间隔
net.ipv4.tcp_keepalive_intvl = 10
# 在认为连接失效之前发送的包数
net.ipv4.tcp_keepalive_probes = 5

然后执行sysctl -p使配置生效。

  1. 硬件防火墙配置:

    硬件防火墙通常通过其管理界面进行配置,这通常涉及定义允许通过的网络流量规则,以及可能的异常规则。具体配置取决于硬件防火墙的品牌和型号。

这些配置通常由网络管理员或IT专家进行,因此不需要详细的代码示例。然而,对于TCP KeepAlive,可以使用sysctl命令来配置Linux系统参数,对于DCD和监听器注册,需要在Oracle的监听器配置文件中进行相应设置。硬件防火墙的配置通常在防火墙的管理界面上进行。

2024-09-04

解释:

这个错误表明Redis服务器在尝试监听TCP连接时无法在本地地址127.0.0.1的6380端口创建服务器套接字。这通常是因为端口已被占用或权限问题。

解决方法:

  1. 检查端口是否被其他应用占用:

    • 在Linux上,可以使用lsof -i :6380netstat -tulnp | grep 6380
    • 在Windows上,可以使用netstat -ano | findstr :6380
  2. 如果端口被占用,请找到占用端口的应用程序并停止它,或者更改Redis配置文件中的端口号。
  3. 如果端口未被占用,检查Redis配置文件中的bind指令是否包含127.0.0.1,如果不是,添加它或者移除bind指令以允许任意地址绑定。
  4. 确保当前用户有权限在该端口上创建套接字。
  5. 如果更改配置后问题依旧,请检查防火墙设置,确保没有阻止Redis的端口。
  6. 重启Redis服务。

如果以上步骤无法解决问题,请查看Redis日志文件以获取更多信息。

2024-09-02

在C#中使用ModeBus TCP读取汇川Easy521 PLC的数据,你需要一个支持ModeBus协议的库,比如LibUA (开源的ModeBus库) 或者使用第三方库,如EasyModbus。

以下是使用EasyModbus库进行ModeBus TCP通信的简单示例:

首先,安装EasyModbus库:




Install-Package EasyModbus

然后,使用以下代码读取PLC的寄存器:




using EasyModbus;
using System;
using System.Net;
 
namespace ModbusTCPExample
{
    class Program
    {
        static void Main(string[] args)
        {
            // PLC的IP地址
            string plcIp = "192.168.1.10";
            // PLC的端口,ModeBus TCP通常是502
            int port = 502;
            // 创建Modbus TCP客户端
            ModbusClient modbusClient = new ModbusClient(IPAddress.Parse(plcIp), port);
            try
            {
                // 连接PLC
                modbusClient.Connect();
                // 读取起始地址为0的寄存器,长度为10的数据
                int[] registers = modbusClient.ReadHoldingRegisters(0, 10);
                for (int i = 0; i < registers.Length; i++)
                {
                    Console.WriteLine($"Register {i}: {registers[i]}");
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
            finally
            {
                // 断开连接
                if (modbusClient.Connected)
                {
                    modbusClient.Disconnect();
                }
            }
        }
    }
}

确保PLC的IP地址、端口和寄存器地址是正确的,并且你有足够的权限去访问PLC的资源。上述代码中的ReadHoldingRegisters方法用于读取保持寄存器,如果你需要读取输入寄存器或者线圈状态,可以使用ReadInputRegistersReadCoils方法。