最好的学习方式：造轮子
Redis 是面试必问、工作必用的中间件。但大多数人停留在"会用""能背八股"的层面。这篇文章带你从零写一个 Mini Redis，彻底搞懂底层原理。
第一层：网络层（IO 多路复用）
Redis 的单线程高性能，核心是 epoll。我们用 Python 的 selectors 模块模拟：
import selectors, socket
sel = selectors.DefaultSelector()
def accept(sock, mask):
conn, addr = sock.accept()
conn.setblocking(False)
sel.register(conn, selectors.EVENT_READ, read)
def read(conn, mask):
data = conn.recv(1024)
if data:
response = process_command(data)
conn.send(response)
else:
sel.unregister(conn)
conn.close()
server = socket.socket()
server.bind(('0.0.0.0', 6379))
server.listen(128)
server.setblocking(False)
sel.register(server, selectors.EVENT_READ, accept)
while True:
events = sel.select()
for key, mask in events:
key.data(key.fileobj, mask)
第二层：协议解析（RESP）
Redis 使用 RESP（REdis Serialization Protocol），非常简单。
def parse_resp(data):
"""解析 RESP 协议"""
lines = data.decode().split('\r\n')
if lines[0][0] == '*': # 数组
count = int(lines[0][1:])
args = []
idx = 1
for _ in range(count):
if lines[idx][0] == '$':
length = int(lines[idx][1:])
idx += 1
args.append(lines[idx])
idx += 1
return args
return [lines[0]]
def encode_resp(data):
"""编码 RESP 响应"""
if isinstance(data, str):
return f"+{data}\r\n".encode()
if data is None:
return "$-1\r\n".encode()
if isinstance(data, int):
return f":{data}\r\n".encode()
第三层：核心数据结构——跳表（Skip List）
Redis 的 Sorted Set 底层是跳表。跳表 = 多层索引的有序链表。
import random
class SkipNode:
def __init__(self, key, value, level):
self.key = key
self.value = value
self.forward = [None] * (level + 1)
class SkipList:
MAX_LEVEL = 16
def __init__(self):
self.head = SkipNode(None, None, self.MAX_LEVEL)
self.level = 0
def random_level(self):
level = 0
while random.random() < 0.25 and level < self.MAX_LEVEL:
level += 1
return level
def insert(self, key, value):
update = [None] * (self.MAX_LEVEL + 1)
current = self.head
for i in range(self.level, -1, -1):
while current.forward[i] and current.forward[i].key < key:
current = current.forward[i]
update[i] = current
level = self.random_level()
if level > self.level:
for i in range(self.level + 1, level + 1):
update[i] = self.head
self.level = level
node = SkipNode(key, value, level)
for i in range(level + 1):
node.forward[i] = update[i].forward[i]
update[i].forward[i] = node
跳表的插入/查询/删除复杂度都是 O(log n)，而且实现比红黑树简单得多。
第四层：过期机制
class MiniRedis:
def __init__(self):
self.store = {} # key -> value
self.expires = {} # key -> timestamp
def get(self, key):
# 惰性删除
if key in self.expires and time.time() > self.expires[key]:
del self.store[key]
del self.expires[key]
return None
return self.store.get(key)
def set(self, key, value, px=None):
self.store[key] = value
if px:
self.expires[key] = time.time() + px / 1000
def clean_expired(self):
"""定期随机抽查删除（Redis 的做法）"""
keys = list(self.expires.keys())
random.shuffle(keys)
now = time.time()
for key in keys[:20]:
if self.expires[key] < now:
del self.store[key]
del self.expires[key]
完整架构图
Client → Socket → selectors (epoll) → RESP Parser
↓
Command Router
/ | \
GET/SET EXPIRE DEL
\ | /
Hash Table
Skip List (ZSET)
List (LIST)
下一步
你可以在 Mini Redis 上继续添加：AOF 持久化、主从复制、事务（MULTI/EXEC）、Pub/Sub、Lua 脚本支持。每一个功能的实现，都会让你对 Redis 的理解上一个台阶。