Linux C/C++ 学习日记（65）：Redis（六）：论”单线程“服务器是如何实现高性能的，含部分源码剖析

注：该文用于个人学习记录和知识交流，如有不足，欢迎指点。

一、线程数目

• 操作数据结构的语句（即执行指令）是主线程（单线程，天然具备隔离性、不需要加锁）
• 别的线程主要用于优化：避免IO密集型和CPU密集型过多占用主线程

二、IO密集型层面的优化

1. 磁盘IO

• rdb：通过fork子进程来持久化
• aof：有异步刷盘机制（可选）

2. 网络IO

• 通过reactor模型管理多个连接（主线程）
• 数据的接收，数据发送分摊到多个io子进程中

三、CPU密集型层面的优化

• 分治思想：协议的解析（命令接收完之后）、协议的加密（响应回复前）分摊到多个io子进程
• 数据结构的切换：value的不同数据结构底层存储方式在数据量不同情况下采取不同的策略（在执行效率与空间占比间保持平衡）
• 渐进式数据迁移：rehash的两种方式：
  • 每执行一条指令copy一个槽位
  • 当服务器空闲时，定时1ms进行rehash的数据迁移

四、数据结构层面优化

简单说：数据量小时用 “紧凑编码” 省内存，数据量大时用 “高效编码” 保性能。

1. 哈希 （ziplist → dict）

• ziplist（压缩列表）：
  • 内存上：所有键值对 / 元素连续存储，无冗余指针（dict 每个键值对要存两个指针，skiplist 每个节点有多层指针），内存密度是 dict 的 1/5 以上；
  • 性能上：增删改需要重排整个列表（O (n)），但数据量小（≤512/128 个元素）时，n 很小，性能损耗可忽略；
• dict/skiplist：
  • 内存上：dict 有哈希表开销（桶数组 + 链表） ；
  • 性能上：dict 查改 O (1), 数据量大时远快于 ziplist 的 O (n)。

设计目的：小哈希省内存，大哈希用 dict保证读写性能。

1.1 hashTypeSet

/* 给哈希对象设置 field-value 键值对 * 参数说明： * o: 目标哈希对象（robj 类型） * field: 哈希的字段（SDS 字符串） * value: 哈希的取值（SDS 字符串） * flags: 控制标志（如是否接管 field/value 的内存所有权） * 返回值：1 表示更新了已有字段，0 表示新增了字段 */ int hashTypeSet(robj *o, sds field, sds value, int flags) { int update = 0; // 标记是否是“更新”操作（1=更新，0=新增） /* 分支1：哈希对象当前是 LISTPACK 编码（紧凑存储，替代旧的 ziplist） */ if (o->encoding == OBJ_ENCODING_LISTPACK) { unsigned char *zl, *fptr, *vptr; zl = o->ptr; // 取出 listpack 指针（o->ptr 指向 listpack 结构体） fptr = lpFirst(zl); // 指向 listpack 第一个元素 if (fptr != NULL) { // listpack 非空 // 在 listpack 中查找指定 field（1 表示“匹配整个字段”） fptr = lpFind(zl, fptr, (unsigned char*)field, sdslen(field), 1); if (fptr != NULL) { // 找到该 field，执行“更新”逻辑 /* fptr 指向 field，vptr 指向 field 对应的 value（listpack 中 field/value 成对存储） */ vptr = lpNext(zl, fptr); serverAssert(vptr != NULL); // 断言：field 必须有对应的 value（防御性编程） update = 1; // 标记为更新操作 /* 替换原有 value：lpReplace 会重新分配 listpack 内存（若需要），返回新的 listpack 指针 */ zl = lpReplace(zl, &vptr, (unsigned char*)value, sdslen(value)); } } if (!update) { // 未找到 field，执行“新增”逻辑 /* 将 field 和 value 依次追加到 listpack 尾部（成对存储） */ zl = lpAppend(zl, (unsigned char*)field, sdslen(field)); // 追加 field zl = lpAppend(zl, (unsigned char*)value, sdslen(value)); // 追加 value } o->ptr = zl; // 更新对象的 listpack 指针（可能是新分配的内存） /* 关键：检查是否需要从 LISTPACK 转换为哈希表（HT） * hash_max_listpack_entries 是配置阈值（默认512），元素数超过阈值则转换 * 目的：listpack 增删改是 O(n)，数据量大时性能差，转为 HT（O(1) 读写） */ if (hashTypeLength(o) > server.hash_max_listpack_entries) hashTypeConvert(o, OBJ_ENCODING_HT); } /* 分支2：哈希对象当前是 HT 编码（哈希表，高效存储） */ else if (o->encoding == OBJ_ENCODING_HT) { // 在哈希表中查找 field 对应的条目 dictEntry *de = dictFind(o->ptr,field); if (de) { // 找到条目，执行“更新”逻辑 sdsfree(dictGetVal(de)); // 释放原有 value 的 SDS 内存 // 根据 flags 决定是否接管 value 的内存所有权 if (flags & HASH_SET_TAKE_VALUE) { dictGetVal(de) = value; // 直接复用 value 指针（不复制） value = NULL; // 标记 value 已被接管，后续无需释放 } else { dictGetVal(de) = sdsdup(value); // 复制 value（保留原 value 所有权） } update = 1; // 标记为更新操作 } else { // 未找到条目，执行“新增”逻辑 sds f,v; // 处理 field 的内存所有权：是否接管（不复制） if (flags & HASH_SET_TAKE_FIELD) { f = field; // 直接复用 field 指针 field = NULL; // 标记 field 已被接管 } else { f = sdsdup(field); // 复制 field } // 处理 value 的内存所有权：是否接管（不复制） if (flags & HASH_SET_TAKE_VALUE) { v = value; // 直接复用 value 指针 value = NULL; // 标记 value 已被接管 } else { v = sdsdup(value); // 复制 value } dictAdd(o->ptr,f,v); // 将 field-value 加入哈希表 } } /* 分支3：未知编码，直接崩溃（防御性编程） */ else { serverPanic("Unknown hash encoding"); } /* 收尾：释放未被接管的 SDS 内存（避免内存泄漏） * 若 flags 要求函数接管 field/value 所有权，但未被使用，则释放 */ if (flags & HASH_SET_TAKE_FIELD && field) sdsfree(field); if (flags & HASH_SET_TAKE_VALUE && value) sdsfree(value); return update; // 返回是否是更新操作 }

（t_hash.c)

1.2 dict结构体

（dict.h）

1.3 rehash

ht_table[0];  // 用户访问的数据结构

ht_table[1] = null; // 用于渐进式数据迁移（rehashidx）

渐进式迁移

• 法1：分摊在增删改查中，每次操作就迁移一个槽位
• 法2：redis空闲时，迁移1ms的时间

操作完之后，释放ht_table[0]，将其指向数据迁移完毕的结构体，然后置ht_table[1]为null

2. 字符串（int → embstr → raw）

• int 编码：数字型字符串直接存成整数，抛弃 SDS 结构，内存从 “redisObject+SDS”（≥23 字节）降到仅 redisObject（16 字节），且整数运算无需解析字符串，效率拉满；
• embstr 编码：短字符串（≤44 字节）用连续内存存储，一次分配 / 释放，CPU 缓存命中率高（64 字节缓存行刚好装下），比 raw 少一次内存申请；
• raw 编码：长字符串无法塞进 64 字节缓存行，只能分离存储，牺牲一点内存效率，保证长字符串的读写性能。

设计目的：数字 / 短字符串极致省内存，长字符串保证功能兼容（最大 512MB）

2.1 string对象的构建

/* 尝试对字符串对象进行编码优化，以节省内存空间 * 核心逻辑：RAW/EMBSTR → INT（数字型） 或 短字符串→EMBSTR，长字符串优化SDS * 返回值：优化后的对象（可能是新对象，也可能是原对象） */ robj *tryObjectEncoding(robj *o) { long value; // 存储字符串转换后的整数值 sds s = o->ptr; // 取出字符串对象的SDS指针 size_t len; // 字符串长度 /* 断言：当前对象必须是字符串类型（OBJ_STRING） * 这个函数只处理字符串对象的编码优化，其他类型由各自的命令处理 */ serverAssertWithInfo(NULL,o,o->type == OBJ_STRING); /* 只对 RAW/EMBSTR 编码的字符串做优化（这两种编码是基于字符数组的原始存储） * 已经是INT编码的对象无需再优化 */ if (!sdsEncodedObject(o)) return o; /* 共享对象不做编码优化： * 共享对象（如共享整数）可能被整个Redis的对象空间引用，优化后可能导致逻辑异常 * 共享对象仅作为键空间的值使用时才安全，因此引用计数>1时直接返回原对象 */ if (o->refcount > 1) return o; /* 第一步：尝试将字符串转换为长整数（INT编码优化） * 注：长度>20字节的字符串不可能转成32/64位整数，直接跳过 */ len = sdslen(s); // 获取SDS字符串的实际长度 if (len <= 20 && string2l(s,len,&value)) { // 长度≤20且能转成long整数 /* 尝试使用共享整数对象（进一步节省内存） * 例外场景：开启maxmemory且策略不允许共享整数时，不使用共享对象 * 原因：LRU淘汰算法需要每个对象有独立的LRU字段，共享对象会导致LRU统计错误 */ if ((server.maxmemory == 0 || !(server.maxmemory_policy & MAXMEMORY_FLAG_NO_SHARED_INTEGERS)) && value >= 0 && value < OBJ_SHARED_INTEGERS) // OBJ_SHARED_INTEGERS默认是10000，共享0~9999的整数 { decrRefCount(o); // 释放原对象的引用计数 incrRefCount(shared.integers[value]); // 增加共享对象的引用计数 return shared.integers[value]; // 返回共享整数对象 } else { // 不满足共享条件时，直接将原对象转为INT编码 if (o->encoding == OBJ_ENCODING_RAW) { // 原编码是RAW sdsfree(o->ptr); // 释放原SDS内存 o->encoding = OBJ_ENCODING_INT; // 标记为INT编码 o->ptr = (void*) value; // ptr不再指向SDS，而是存储整数值 return o; // 返回优化后的原对象 } else if (o->encoding == OBJ_ENCODING_EMBSTR) { // 原编码是EMBSTR decrRefCount(o); // 释放原EMBSTR对象 return createStringObjectFromLongLongForValue(value); // 创建新的INT编码对象 } } } /* 第二步：短字符串优化（RAW→EMBSTR） * OBJ_ENCODING_EMBSTR_SIZE_LIMIT默认是44字节，长度≤44的RAW字符串转EMBSTR * EMBSTR的优势：redisObject和SDS连续存储，一次分配/释放，缓存命中率高 */ if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT) { robj *emb; // 新的EMBSTR编码对象 if (o->encoding == OBJ_ENCODING_EMBSTR) return o; // 已经是EMBSTR，直接返回 emb = createEmbeddedStringObject(s,sdslen(s)); // 创建EMBSTR对象 decrRefCount(o); // 释放原对象 return emb; // 返回新的EMBSTR对象 } /* 第三步：长字符串SDS优化（无法转INT/EMBSTR时） * 如果SDS尾部有超过10%的空闲空间，裁剪SDS以节省内存 * 仅对长度>44字节的字符串做此优化（因为前面已经处理了短字符串） */ trimStringObjectIfNeeded(o); /* 无法做任何编码优化，返回原对象 */ return o; }

（object.c）

2.2 redisObject结构体

（server.h，LRU_BITS为24 : 结构体占16字节）

2.3 int形式：当string为纯数字时（可以含+-号）且长度 <= 20位。

64 位系统中 void *ptr 本身占 8 字节，但在 int 编码时，Redis 把 ptr 当作 long long 类型（也是 8 字节）来用，而非指针：long long 的取值范围是 -9223372036854775808 ~ 9223372036854775807；这个范围对应的十进制数字的最大长度是 19 位（比如 9223372036854775807 是 19 位），加上正负号就是 20 位

2.4 embstr形式

（sds.h, 结构体占3字节）

2.5. raw编码则是redisObject中成员ptr指向一块另外分配的内存空间

2.6 总结

小于20字节且为数字时，选择int编码。大于20字节小于44字节，选择embstr编码。大于44字节，选择raw编码embstr：嵌入字符串，嵌入到redisObject中，而raw就是在redisObject中维持一个指向堆上的资源cpu cache line最小访问单位为64字节，所以选择64个字节作为分界线。20字节为界的原因：long long int 表示的数字范围：-19位 ~ +19位，即长度为20的字符串44字节为界的原因：buf可用空间为：64-16-3-1 = 44。 （这个1作为'\0'，以便兼容c的字符串库函数）

3. 列表（quicklist 封装 ziplist）

• 早期 Redis 列表用 ziplist（小列表）和 linkedlist（大列表），但 linkedlist 每个节点有指针开销，内存效率低；
• 现在改用 quicklist（双向链表 + 每个节点是 ziplist）：
  • 既保留了 ziplist 的内存紧凑性（每个节点内的元素连续存储）；
  • 又避免了纯 ziplist 重排内存的性能问题（增删只影响单个 ziplist 节点，而非整个列表）。

设计目的：平衡内存效率和增删性能，是对传统 ziplist/linkedlist 的优化升级。

核心说明list 的底层编码是 quicklist（双向链表），而 quicklist 的每个节点又是一个 ziplist（压缩列表节点）；因此，list 并不直接转换为 ziplist，而是 quicklist 内部间接使用 ziplist 来存储每个节点的数据。

/* ========================== 1. quicklist 结构体（列表的顶层结构）========================== * 说明：Redis列表(List)的底层实现，是「双向链表 + 每个节点内嵌 listpack」的混合结构， * 替代了早期的 ziplist（小列表）和 linkedlist（大列表），平衡内存效率和增删性能 */ typedef struct quicklist { quicklistNode *head; // 指向双向链表的头节点 quicklistNode *tail; // 指向双向链表的尾节点 unsigned long count; // 所有节点中 listpack 存储的**总元素数**（快速获取列表长度，无需遍历节点） unsigned long len; // 双向链表中 quicklistNode 的**节点总数** signed int fill : QL_FILL_BITS; // 位域（QL_FILL_BITS 通常是16位）：控制每个节点的 listpack 填充因子 // 对应配置项 list-max-ziplist-size，比如-1=最多4KB，-2=最多8KB，1=最多1个元素等 unsigned int compress : QL_COMP_BITS; // 位域（QL_COMP_BITS 通常是16位）：控制列表的压缩深度 // 对应配置项 list-compress-depth，0=不压缩，1=首尾各1个节点不压缩，其余压缩，以此类推 unsigned int bookmark_count: QL_BM_BITS; // 位域（QL_BM_BITS 通常是3位）：书签的数量，用于快速定位节点 quicklistBookmark bookmarks[]; // 柔性数组：存储书签（优化大列表的随机访问性能，避免遍历整个链表） } quicklist; /* ========================== 2. quicklistNode 结构体（列表的节点结构）========================== * 说明：quicklist 的每个节点，内部封装了 listpack（紧凑存储结构，替代旧的 ziplist）， * 支持压缩（LZF算法），兼顾内存效率和访问性能 */ typedef struct quicklistNode { struct quicklistNode *prev; // 指向双向链表的前一个节点 struct quicklistNode *next; // 指向双向链表的后一个节点 unsigned char *entry; // 指向实际存储数据的指针： // - 未压缩时：指向 listpack 结构 // - 压缩时：指向 LZF 压缩后的数据 size_t sz; // entry 指向数据的**总字节数**（方便内存分配/释放/压缩计算） unsigned int count : 16; // 位域（16位）：当前节点的 listpack 中存储的**元素个数**（最大65535个） unsigned int encoding : 2; // 位域（2位）：当前节点的编码方式 // 1=RAW（未压缩，直接存储 listpack） // 2=LZF（用 LZF 算法压缩 listpack，节省内存） unsigned int container : 2; // 位域（2位）：当前节点的数据容器类型 // 1=PLAIN（早期的普通存储，已废弃） // 2=PACKED（当前唯一使用的类型，即使用 listpack 存储数据） unsigned int recompress : 1; // 位域（1位）：标记该节点**之前是否被压缩过** // 当需要访问压缩节点时，会临时解压，访问完成后需要重新压缩 unsigned int attempted_compress : 1; // 位域（1位）：标记该节点**尝试过压缩但失败** // 原因是节点数据太小，压缩后反而会增加内存开销 unsigned int extra : 10; // 位域（10位）：预留位，用于未来扩展功能（Redis 预留位的常规操作） } quicklistNode;

（quicklist.h)。每个 quicklistNode 内部包含一个 ziplist，用于存储实际数据，这就是 “间接使用 ziplist” 的含义。

4. 集合（intset → dict）

• intset（整数集合）：
  • 内存上：连续存储整数，无任何冗余（比如存储 1,2,3 只占 3×8=24 字节），而 dict 存储同样数据需要哈希表 + 指针，至少占上百字节；
  • 限制：只能存整数，且增删需要重排内存；
• dict：
  • 内存上：有哈希表开销，但支持任意字符串元素；
  • 性能上：增删查 O (1)，无类型限制。

设计目的：纯整数小集合用 intset 省内存，非整数 / 大集合用 dict 保证通用性和性能（比如用户标签集合，可能存字符串标签，只能用 dict）。

5. 有序集合

• ziplist（压缩列表）：
  • 内存上：所有键值对 / 元素连续存储，无冗余指针（dict 每个键值对要存两个指针，skiplist 每个节点有多层指针），内存密度是 dict 的 1/5 以上；
  • 性能上：增删改需要重排整个列表（O (n)），但数据量小（≤512/128 个元素）时，n 很小，性能损耗可忽略；
• skiplist：
  • 内存上：skiplist 有多层指针开销，内存占用是 ziplist 的数倍；
  • 性能上：skiplist 查改 O (logN)，数据量大时远快于 ziplist 的 O (n)。

设计目的 小有序集合用 ziplist 省内存 大有序集合用 skiplist 保证读写性能（比如电商场景的商品排行榜 zset，数据量变大后，skiplist 的排序效率远高于 ziplist）

/* 向有序集合中添加/更新元素（核心函数） * 参数说明： * zobj: 目标有序集合对象（robj 类型） * score: 元素的分数（double 类型） * ele: 元素值（SDS 字符串） * in_flags: 输入控制标志（如 ZADD_IN_INCR/XX/NX/GT/LT 等） * out_flags: 输出标志（返回操作结果，如是否新增/更新/NaN 等） * newscore: 输出参数，返回更新后的分数（NULL 则不返回） * 返回值：1 表示操作成功（新增/更新），0 表示失败（如 NaN 输入） */ int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore) { /* 将输入标志解析为易检查的变量 */ int incr = (in_flags & ZADD_IN_INCR) != 0; // 是否是增量更新（对应 ZINCRBY 或 ZADD INCR） int nx = (in_flags & ZADD_IN_NX) != 0; // 仅当元素不存在时添加（ZADD NX） int xx = (in_flags & ZADD_IN_XX) != 0; // 仅当元素存在时更新（ZADD XX） int gt = (in_flags & ZADD_IN_GT) != 0; // 仅当新分数 > 旧分数时更新（ZADD GT） int lt = (in_flags & ZADD_IN_LT) != 0; // 仅当新分数 < 旧分数时更新（ZADD LT） *out_flags = 0; /* 初始化输出标志 */ double curscore; // 存储元素当前的分数（若存在） /* 输入分数是 NaN 直接返回错误（NaN 无法比较，不能作为有序集合分数） */ if (isnan(score)) { *out_flags = ZADD_OUT_NAN; // 标记输出为 NaN 错误 return 0; } /* 分支1：有序集合当前是 LISTPACK 编码（紧凑存储，替代旧的 ziplist） */ if (zobj->encoding == OBJ_ENCODING_LISTPACK) { unsigned char *eptr; // 指向 listpack 中元素的指针 /* 查找元素：存在则返回元素指针并填充 curscore，不存在则返回 NULL */ if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) { /* NX 模式：元素已存在，直接返回（标记为“无操作”） */ if (nx) { *out_flags |= ZADD_OUT_NOP; return 1; } /* INCR 模式：计算增量后的新分数（score = 旧分数 + 输入增量） */ if (incr) { score += curscore; if (isnan(score)) { // 增量后出现 NaN，返回错误 *out_flags |= ZADD_OUT_NAN; return 0; } } /* GT/LT 模式：仅当新分数满足条件时更新 */ if ((lt && score >= curscore) || (gt && score <= curscore)) { *out_flags |= ZADD_OUT_NOP; // 不满足条件，标记无操作 return 1; } /* 若需要返回新分数，填充 newscore */ if (newscore) *newscore = score; /* 分数变化时：先删除旧元素，再插入新分数的元素（listpack 是有序结构，需重排） */ if (score != curscore) { zobj->ptr = zzlDelete(zobj->ptr,eptr); // 删除旧元素 zobj->ptr = zzlInsert(zobj->ptr,ele,score); // 插入新分数的元素 *out_flags |= ZADD_OUT_UPDATED; // 标记为“更新操作” } return 1; } else if (!xx) { // 元素不存在，且不是 XX 模式（执行新增逻辑） /* 新增前检查：是否需要从 LISTPACK 转为 SKIPLIST * 触发条件： * 1. 元素数+1 超过 zset_max_listpack_entries（默认128） * 2. 元素长度超过 zset_max_listpack_value（默认64） * 3. listpack 剩余空间不足以添加新元素 */ if (zzlLength(zobj->ptr)+1 > server.zset_max_listpack_entries || sdslen(ele) > server.zset_max_listpack_value || !lpSafeToAdd(zobj->ptr, sdslen(ele))) { zsetConvert(zobj,OBJ_ENCODING_SKIPLIST); // 转换为 SKIPLIST 编码 } else { // 无需转换，直接插入新元素到 listpack zobj->ptr = zzlInsert(zobj->ptr,ele,score); if (newscore) *newscore = score; *out_flags |= ZADD_OUT_ADDED; // 标记为“新增操作” return 1; } } else { // 元素不存在，但是 XX 模式（仅更新已存在元素） *out_flags |= ZADD_OUT_NOP; // 标记无操作 return 1; } } /* 注：上面的 listpack 分支要么已返回，要么已将对象转为 skiplist 编码 */ /* 分支2：有序集合当前是 SKIPLIST 编码（高效存储，包含 dict + skiplist） */ if (zobj->encoding == OBJ_ENCODING_SKIPLIST) { zset *zs = zobj->ptr; // zset 结构体（包含 dict + skiplist） zskiplistNode *znode; // 跳表节点指针 dictEntry *de; // 哈希表条目指针 /* 在 dict 中查找元素（dict 用于快速定位元素是否存在） */ de = dictFind(zs->dict,ele); if (de != NULL) { // 元素存在，执行更新逻辑 /* NX 模式：元素已存在，返回无操作 */ if (nx) { *out_flags |= ZADD_OUT_NOP; return 1; } curscore = *(double*)dictGetVal(de); // 获取当前分数（dict 中存的是分数指针） /* INCR 模式：计算增量后的新分数 */ if (incr) { score += curscore; if (isnan(score)) { *out_flags |= ZADD_OUT_NAN; return 0; } } /* GT/LT 模式：检查分数条件 */ if ((lt && score >= curscore) || (gt && score <= curscore)) { *out_flags |= ZADD_OUT_NOP; return 1; } /* 填充新分数（若需要返回） */ if (newscore) *newscore = score; /* 分数变化时：更新跳表中的分数 */ if (score != curscore) { // 更新跳表节点的分数（内部会删除旧节点+插入新节点） znode = zslUpdateScore(zs->zsl,curscore,ele,score); /* 注：dict 中无需删除旧条目，仅更新分数指针即可 */ dictGetVal(de) = &znode->score; // 更新 dict 中的分数指针 *out_flags |= ZADD_OUT_UPDATED; // 标记为更新 } return 1; } else if (!xx) { // 元素不存在，且不是 XX 模式（执行新增逻辑） ele = sdsdup(ele); // 复制元素（接管内存所有权） znode = zslInsert(zs->zsl,score,ele); // 插入跳表 // 将元素和分数指针加入 dict（保证 O(1) 查找） serverAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK); *out_flags |= ZADD_OUT_ADDED; // 标记为新增 if (newscore) *newscore = score; return 1; } else { // 元素不存在，但是 XX 模式 *out_flags |= ZADD_OUT_NOP; return 1; } } else { // 未知编码，直接崩溃（防御性编程） serverPanic("Unknown sorted set encoding"); } return 0; /* 永远不会执行到这里（上面分支已覆盖所有情况） */ }

（t_zet.c)

6. 总结：

跳表redis选择采用32层，兼顾内存占用和查找效率的综合考量。

四、io 多进程的工作原理