Redis服务器源码解析:手写极简版搞定版本升级痛点
上周刚把生产环境的 Redis 从 4.0 升到 7.0，结果一堆老代码直接报错。MULTI 命令的行为变了，过期键的处理逻辑也不对劲，改了一下午才搞定。这种“版本升级后 API 全变了”的坑，其实只要懂底层原理，根本不用慌。
很多人只知道 Redis 快，但不知道它为什么快。今天我们就动手写一个极简的 Redis 服务器。不追求功能全，只为了搞懂内存结构、事件循环和协议解析。通过这份源码解析，你会发现那些莫名其妙的 API 行为，其实都是底层逻辑决定的。
项目目标与核心思路
我们的目标不是造轮子去替代 Redis，而是为了“看懂”。
真正的 Redis 有几百万行 C 代码，包含持久化、集群、Lua 脚本等复杂功能。对于理解核心机制来说，这些都是噪音。我们要剥离掉这些，只保留最核心的三块：网络层：如何接收客户端连接？
协议层：如何解析 RESP 协议？
数据层：内存中数据怎么存？怎么查？选 Python 来实现，不是为了性能，而是为了代码可读性。Python 的字典、列表、线程模型，足够模拟 Redis 的核心概念。如果你能用 Python 写出一个能跑通 GET、SET、DEL 的服务，再去读 C 源码时，就不会觉得是天书了。
关键原则：单线程模型：Redis 核心操作是单线程的，我们也模拟这一点，避免并发复杂度干扰思路。
RESP 协议：必须严格遵守 Redis 扩展协议，这样我们可以直接用 redis-cli 测试。
内存哈希表：用 Python 的 dict 模拟 Redis 的 dict.c 中的哈希表。目录结构规划
项目结构保持扁平化，方便阅读。
mini-redis/
├── main.py # 入口文件，启动服务器
├── protocol.py # RESP 协议解析与序列化
├── command.py # 命令处理逻辑
├── storage.py # 内存存储模拟
└── README.md # 说明文档protocol.py 负责和客户端打交道，把字节流变成 Python 对象。
storage.py 负责数据持久化在内存中，模拟 Redis 的内存空间。
command.py 是业务逻辑层，处理 SET、GET 等具体指令。
main.py 是主循环，把三者串联起来。核心代码实现
1. 存储层：模拟 Redis 内存
Redis 的数据结构非常经典，String、List、Hash、Set、ZSet。我们先只实现 String，因为它是基础。
# storage.pyclass RedisStorage:def __init__(self):# 模拟 Redis 的 dict，key 是 bytes，value 是 bytes# 为什么用 bytes？因为网络传输和 Redis 内部都是二进制self.data = {}# 模拟 TTL，key 是 key，value 是过期时间戳self.ttl = {}def set(self, key: bytes, value: bytes, ex: int = None):self.data[key] = valueif ex:import timeself.ttl[key] = time.time() + exelse:self.ttl.pop(key, None)def get(self, key: bytes):if key not in self.data:return None# 检查过期if key in self.ttl:import timeif time.time() self.ttl[key]:self.delete(key)return Nonereturn self.data[key]def delete(self, key: bytes):self.data.pop(key, None)self.ttl.pop(key, None)return 1 if key in self.data else 0解析重点：Key/Value 都是 bytes：这是很多初学者忽略的细节。Redis 内部存储全是二进制，不区分字符串类型。我们在 Python 中也要保持这个习惯，否则序列化时会出 bug。
懒删除策略：注意 get 方法里，我们是在读取时检查过期，而不是后台线程定期删除。这就是 Redis 的“惰性过期”策略。官方文档里提到，这种策略能最大化 CPU 效率，因为只有在访问时才判断，如果键一直没人访问，就不浪费资源去删除它。2. 协议层：解析 RESP
Redis 使用 RESP (REdis Serialization Protocol)。最简单的形式是 *2\r\n$3\r\nSET\r\n$5\r\nhello\r\n$5\r\nworld\r\n。
我们需要把这种二进制流解析成 Python 列表 [[b'SET'], [b'hello'], [b'world']]。
# protocol.pyimport structclass RESPParser:def __init__(self):self.buffer = b''def feed(self, data: bytes):self.buffer += datadef parse(self):解析一条完整的 RESP 命令返回 None 表示数据不完整，需要等待更多数据if not self.buffer:return None# 尝试读取第一行，确定类型# 简单起见，我们假设客户端一次发送完整命令# 实际项目中需要处理半包问题，这里简化处理lines = self.buffer.split(b'\r\n')# 如果最后不是空字符串，说明数据没传完if lines[-1] != b'':return None# 去掉最后的空字符串lines = lines[:-1]if not lines:return None# 清除缓冲区已处理部分（简化版，实际需精确计算长度）self.buffer = b''first_line = lines[0]# 简单判断：如果是简单命令（如 PING），直接返回# 标准 RESP 命令以 * 开头if first_line.startswith(b'*'):try:count = int(first_line[1:])args = []for i in range(count):if lines[1 + i*2] != b'': # 长度行len_str = lines[1 + i*2]val = lines[2 + i*2]args.append(val)return argsexcept (ValueError, IndexError):return Noneelse:# 兼容非标准协议或简单文本return [first_line]避坑指南：半包问题：网络传输是不保证完整性的。上面的 parse 方法做了简化，实际开发中必须维护一个状态机，逐字节解析，直到读取完整长度。
二进制安全：千万不要用 str 处理 key/value，必须用 bytes。因为 Redis 的 key 可以包含 \x00 等控制字符，字符串处理会报错或截断。3. 命令层与主循环
现在把存储和协议连起来。
# command.pyfrom storage import RedisStoragedef execute_command(cmd: list, storage: RedisStorage):if not cmd:return b':-1\r\n' # Errorcommand = cmd[0].upper()args = cmd[1:]if command == b'SET':if len(args) 2:return b':-ERR wrong number of arguments for \'set\' command\r\n'key = args[0]value = args[1]ex = None# 简单解析 EX 参数for i in range(2, len(args), 2):if args[i] == b'EX' and i+1 len(args):ex = int(args[i+1])storage.set(key, value, ex)return b'+OK\r\n'elif command == b'GET':if len(args) != 1:return b':-ERR wrong number of arguments for \'get\' command\r\n'key = args[0]val = storage.get(key)if val is None:return b'$-1\r\n' # Nullelse:# 序列化返回return b'$' + str(len(val)).encode() + b'\r\n' + val + b'\r\n'elif command == b'DEL':if len(args) == 0:return b':-ERR wrong number of arguments for \'del\' command\r\n'count = 0for key in args:if storage.delete(key):count += 1return b':' + str(count).encode() + b'\r\n'elif command == b'PING':return b'+PONG\r\n'return b':-ERR unknown command \'' + command + b'\'\r\n'主入口：
# main.pyimport socket
import threading
from protocol import RESPParser
from command import execute_command
from storage import RedisStoragedef handle_client(conn, storage):parser = RESPParser()while True:try:data = conn.recv(1024)if not data:breakparser.feed(data)cmd = parser.parse()if cmd:response = execute_command(cmd, storage)conn.sendall(response)except Exception as e:print(fClient error: {e})breakconn.close()def start_server(host='127.0.0.1', port=6379):storage = RedisStorage()server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)server.bind((host, port))server.listen(5)print(fMini Redis running on {host}:{port})while True:conn, addr = server.accept()print(fNew connection: {addr})t = threading.Thread(target=handle_client, args=(conn, storage))t.daemon = Truet.start()if __name__ == '__main__':start_server()运行与测试
启动服务后，用 redis-cli 连接测试。
# 终端 1
python main.py# 终端 2
redis-cli -p 6379测试用例：基本读写：
127.0.0.1:6379 SET name zhangsan
OK
127.0.0.1:6379 GET name
zhangsan过期测试：
127.0.0.1:6379 SET temp hello EX 5
OK
127.0.0.1:6379 GET temp
hello
# 等待 6 秒
127.0.0.1:6379 GET temp
(nil)删除测试：
127.0.0.1:6379 DEL name
(integer) 1
127.0.0.1:6379 GET name
(nil)调试技巧：
如果 GET 返回乱码，检查 protocol.py 中的序列化部分。RESP 协议要求返回的 String 必须带长度头 $len\r\ndata\r\n。漏掉长度头，客户端会解析失败。
优化扩展与避坑
这个极简版离生产 Redis 还差得远，但有几个点值得深挖：多线程 vs 单线程：
上面的代码用了 threading，每个连接一个线程。真正的 Redis 核心命令是单线程的，网络 IO 也是多线程（6.0 引入）。为什么 Redis 核心单线程？ 避免锁竞争，CPU 瓶颈通常在内存访问而非网络。
Python 的 GIL：Python 有全局解释器锁，多线程并不能真正并行执行 CPU 密集型任务。如果模拟高并发，建议用 asyncio 重写网络层，保持单线程处理命令，这样更接近 Redis 的真实模型。内存碎片：
Redis 在长时间运行后，内存占用会高于实际数据量。这是因为内存分配器（jemalloc）的碎片化。我们的 Python 版由 Python 解释器管理内存，碎片化问题由解释器内部处理，感知不到。但在 C 实现中，这是运维关注的重点。持久化：
当前版本重启数据丢失。进阶练习可以加入 RDB 快照：每隔 N 秒，将 storage.data 序列化保存到文件。
启动时，如果文件存在，加载到内存。
注意：RDB 是二进制格式，不能直接存 Python 的 dict，需要自定义序列化或转储为 JSON（测试用）。大 Key 问题：
如果 SET 一个 10MB 的 Value，上面的代码会阻塞主线程。真实 Redis 中，大 Key 操作会阻塞其他命令。优化方案是将大 Key 拆分，或者在后台线程异步处理（但要注意一致性）。小结
通过手写这个 Mini Redis，我们把黑盒变成了白盒。版本升级 API 变化的本质：往往是底层数据结构或内存管理机制的调整。比如 4.0 到 7.0，集群模式的变化、过期策略的优化，都直接影响命令行为。
源码阅读方法：不要从头读到尾。先找 server.c 的主循环，看事件如何触发；再看 dict.c，看哈希表如何冲突解决；最后看具体命令的实现。
官方文档的价值：Redis 官方文档的“Internals”章节，虽然简短，但指出的方向非常准确。结合代码看，事半功倍。你更常用哪种写法？是喜欢用 Python 这种脚本语言快速模拟逻辑，还是直接啃 C 源码？评论区交流一下，看看大家是怎么入门 Redis 底层的。 Redis服务器源码解析:手写极简版搞定版本升级痛点
Redis服务器源码解析:手写极简版搞定版本升级痛点
上周刚把生产环境的 Redis 从 4.0 升到 7.0，结果一堆老代码直接报错。MULTI 命令的行为变了，过期键的处理逻辑也不对劲，改了一下午才搞定。这种“版本升级后 API 全变了”的坑，其实只要懂底层原理，根本不用慌。
很多人只知道 Redis 快，但不知道它为什么快。今天我们就动手写一个极简的 Redis 服务器。不追求功能全，只为了搞懂内存结构、事件循环和协议解析。通过这份源码解析，你会发现那些莫名其妙的 API 行为，其实都是底层逻辑决定的。
项目目标与核心思路
我们的目标不是造轮子去替代 Redis，而是为了“看懂”。
真正的 Redis 有几百万行 C 代码，包含持久化、集群、Lua 脚本等复杂功能。对于理解核心机制来说，这些都是噪音。我们要剥离掉这些，只保留最核心的三块：网络层：如何接收客户端连接？
协议层：如何解析 RESP 协议？
数据层：内存中数据怎么存？怎么查？选 Python 来实现，不是为了性能，而是为了代码可读性。Python 的字典、列表、线程模型，足够模拟 Redis 的核心概念。如果你能用 Python 写出一个能跑通 GET、SET、DEL 的服务，再去读 C 源码时，就不会觉得是天书了。
关键原则：单线程模型：Redis 核心操作是单线程的，我们也模拟这一点，避免并发复杂度干扰思路。
RESP 协议：必须严格遵守 Redis 扩展协议，这样我们可以直接用 redis-cli 测试。
内存哈希表：用 Python 的 dict 模拟 Redis 的 dict.c 中的哈希表。目录结构规划
项目结构保持扁平化，方便阅读。
mini-redis/
├── main.py # 入口文件，启动服务器
├── protocol.py # RESP 协议解析与序列化
├── command.py # 命令处理逻辑
├── storage.py # 内存存储模拟
└── README.md # 说明文档protocol.py 负责和客户端打交道，把字节流变成 Python 对象。
storage.py 负责数据持久化在内存中，模拟 Redis 的内存空间。
command.py 是业务逻辑层，处理 SET、GET 等具体指令。
main.py 是主循环，把三者串联起来。核心代码实现
1. 存储层：模拟 Redis 内存
Redis 的数据结构非常经典，String、List、Hash、Set、ZSet。我们先只实现 String，因为它是基础。
# storage.pyclass RedisStorage:def __init__(self):# 模拟 Redis 的 dict，key 是 bytes，value 是 bytes# 为什么用 bytes？因为网络传输和 Redis 内部都是二进制self.data = {}# 模拟 TTL，key 是 key，value 是过期时间戳self.ttl = {}def set(self, key: bytes, value: bytes, ex: int = None):self.data[key] = valueif ex:import timeself.ttl[key] = time.time() + exelse:self.ttl.pop(key, None)def get(self, key: bytes):if key not in self.data:return None# 检查过期if key in self.ttl:import timeif time.time() self.ttl[key]:self.delete(key)return Nonereturn self.data[key]def delete(self, key: bytes):self.data.pop(key, None)self.ttl.pop(key, None)return 1 if key in self.data else 0解析重点：Key/Value 都是 bytes：这是很多初学者忽略的细节。Redis 内部存储全是二进制，不区分字符串类型。我们在 Python 中也要保持这个习惯，否则序列化时会出 bug。
懒删除策略：注意 get 方法里，我们是在读取时检查过期，而不是后台线程定期删除。这就是 Redis 的“惰性过期”策略。官方文档里提到，这种策略能最大化 CPU 效率，因为只有在访问时才判断，如果键一直没人访问，就不浪费资源去删除它。2. 协议层：解析 RESP
Redis 使用 RESP (REdis Serialization Protocol)。最简单的形式是 *2\r\n$3\r\nSET\r\n$5\r\nhello\r\n$5\r\nworld\r\n。
我们需要把这种二进制流解析成 Python 列表 [[b'SET'], [b'hello'], [b'world']]。
# protocol.pyimport structclass RESPParser:def __init__(self):self.buffer = b''def feed(self, data: bytes):self.buffer += datadef parse(self):解析一条完整的 RESP 命令返回 None 表示数据不完整，需要等待更多数据if not self.buffer:return None# 尝试读取第一行，确定类型# 简单起见，我们假设客户端一次发送完整命令# 实际项目中需要处理半包问题，这里简化处理lines = self.buffer.split(b'\r\n')# 如果最后不是空字符串，说明数据没传完if lines[-1] != b'':return None# 去掉最后的空字符串lines = lines[:-1]if not lines:return None# 清除缓冲区已处理部分（简化版，实际需精确计算长度）self.buffer = b''first_line = lines[0]# 简单判断：如果是简单命令（如 PING），直接返回# 标准 RESP 命令以 * 开头if first_line.startswith(b'*'):try:count = int(first_line[1:])args = []for i in range(count):if lines[1 + i*2] != b'': # 长度行len_str = lines[1 + i*2]val = lines[2 + i*2]args.append(val)return argsexcept (ValueError, IndexError):return Noneelse:# 兼容非标准协议或简单文本return [first_line]避坑指南：半包问题：网络传输是不保证完整性的。上面的 parse 方法做了简化，实际开发中必须维护一个状态机，逐字节解析，直到读取完整长度。
二进制安全：千万不要用 str 处理 key/value，必须用 bytes。因为 Redis 的 key 可以包含 \x00 等控制字符，字符串处理会报错或截断。3. 命令层与主循环
现在把存储和协议连起来。
# command.pyfrom storage import RedisStoragedef execute_command(cmd: list, storage: RedisStorage):if not cmd:return b':-1\r\n' # Errorcommand = cmd[0].upper()args = cmd[1:]if command == b'SET':if len(args) 2:return b':-ERR wrong number of arguments for \'set\' command\r\n'key = args[0]value = args[1]ex = None# 简单解析 EX 参数for i in range(2, len(args), 2):if args[i] == b'EX' and i+1 len(args):ex = int(args[i+1])storage.set(key, value, ex)return b'+OK\r\n'elif command == b'GET':if len(args) != 1:return b':-ERR wrong number of arguments for \'get\' command\r\n'key = args[0]val = storage.get(key)if val is None:return b'$-1\r\n' # Nullelse:# 序列化返回return b'$' + str(len(val)).encode() + b'\r\n' + val + b'\r\n'elif command == b'DEL':if len(args) == 0:return b':-ERR wrong number of arguments for \'del\' command\r\n'count = 0for key in args:if storage.delete(key):count += 1return b':' + str(count).encode() + b'\r\n'elif command == b'PING':return b'+PONG\r\n'return b':-ERR unknown command \'' + command + b'\'\r\n'主入口：
# main.pyimport socket
import threading
from protocol import RESPParser
from command import execute_command
from storage import RedisStoragedef handle_client(conn, storage):parser = RESPParser()while True:try:data = conn.recv(1024)if not data:breakparser.feed(data)cmd = parser.parse()if cmd:response = execute_command(cmd, storage)conn.sendall(response)except Exception as e:print(fClient error: {e})breakconn.close()def start_server(host='127.0.0.1', port=6379):storage = RedisStorage()server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)server.bind((host, port))server.listen(5)print(fMini Redis running on {host}:{port})while True:conn, addr = server.accept()print(fNew connection: {addr})t = threading.Thread(target=handle_client, args=(conn, storage))t.daemon = Truet.start()if __name__ == '__main__':start_server()运行与测试
启动服务后，用 redis-cli 连接测试。
# 终端 1
python main.py# 终端 2
redis-cli -p 6379测试用例：基本读写：
127.0.0.1:6379 SET name zhangsan
OK
127.0.0.1:6379 GET name
zhangsan过期测试：
127.0.0.1:6379 SET temp hello EX 5
OK
127.0.0.1:6379 GET temp
hello
# 等待 6 秒
127.0.0.1:6379 GET temp
(nil)删除测试：
127.0.0.1:6379 DEL name
(integer) 1
127.0.0.1:6379 GET name
(nil)调试技巧：
如果 GET 返回乱码，检查 protocol.py 中的序列化部分。RESP 协议要求返回的 String 必须带长度头 $len\r\ndata\r\n。漏掉长度头，客户端会解析失败。
优化扩展与避坑
这个极简版离生产 Redis 还差得远，但有几个点值得深挖：多线程 vs 单线程：
上面的代码用了 threading，每个连接一个线程。真正的 Redis 核心命令是单线程的，网络 IO 也是多线程（6.0 引入）。为什么 Redis 核心单线程？ 避免锁竞争，CPU 瓶颈通常在内存访问而非网络。
Python 的 GIL：Python 有全局解释器锁，多线程并不能真正并行执行 CPU 密集型任务。如果模拟高并发，建议用 asyncio 重写网络层，保持单线程处理命令，这样更接近 Redis 的真实模型。内存碎片：
Redis 在长时间运行后，内存占用会高于实际数据量。这是因为内存分配器（jemalloc）的碎片化。我们的 Python 版由 Python 解释器管理内存，碎片化问题由解释器内部处理，感知不到。但在 C 实现中，这是运维关注的重点。持久化：
当前版本重启数据丢失。进阶练习可以加入 RDB 快照：每隔 N 秒，将 storage.data 序列化保存到文件。
启动时，如果文件存在，加载到内存。
注意：RDB 是二进制格式，不能直接存 Python 的 dict，需要自定义序列化或转储为 JSON（测试用）。大 Key 问题：
如果 SET 一个 10MB 的 Value，上面的代码会阻塞主线程。真实 Redis 中，大 Key 操作会阻塞其他命令。优化方案是将大 Key 拆分，或者在后台线程异步处理（但要注意一致性）。小结
通过手写这个 Mini Redis，我们把黑盒变成了白盒。版本升级 API 变化的本质：往往是底层数据结构或内存管理机制的调整。比如 4.0 到 7.0，集群模式的变化、过期策略的优化，都直接影响命令行为。
源码阅读方法：不要从头读到尾。先找 server.c 的主循环，看事件如何触发；再看 dict.c，看哈希表如何冲突解决；最后看具体命令的实现。
官方文档的价值：Redis 官方文档的“Internals”章节，虽然简短，但指出的方向非常准确。结合代码看，事半功倍。你更常用哪种写法？是喜欢用 Python 这种脚本语言快速模拟逻辑，还是直接啃 C 源码？评论区交流一下，看看大家是怎么入门 Redis 底层的。