**适用范围：**Redis7.2.6核心C源码；对照6.2.14。
**本章目标：**按命令链、数据结构和故障路径阅读，先避开版本混用。
0.1 源码基线与本书的使用方式
本书以Redis7.2.6固定提交为主线，对照6.2.14解释关键演进；两者是阅读基线，不是最新版本或部署选型建议。所有源码链接指向固定提交，不会随分支更新改变行号。Redis8.x只在版本导航说明中讨论，不能把其后新增能力代入7.2.6代码。
不需要编译C、不需要运行redis-server、不需要做实验。每节按机制图→连续源码节选→字段与分支→纸面案例阅读。窗口可能在方法中间结束，完整实现以链接为准；这里只统一显示缩进，未用伪代码替换算法。
图01
源码对照：server.c · 7.2.6 · L7042–L7371。连续节选，窗口可能结束于函数中间；完整函数及调用方见链接。
Redis 7.2.6 · src/server.c · L7042–L7371
int main(int argc, char **argv) {
struct timeval tv;
int j;
char config_from_stdin = 0;
#ifdef REDIS_TEST
if (argc >= 3 && !strcasecmp(argv[1], "test")) {
int flags = 0;
for (j = 3; j < argc; j++) {
char *arg = argv[j];
if (!strcasecmp(arg, "--accurate")) flags |= REDIS_TEST_ACCURATE;
else if (!strcasecmp(arg, "--large-memory")) flags |= REDIS_TEST_LARGE_MEMORY;
else if (!strcasecmp(arg, "--valgrind")) flags |= REDIS_TEST_VALGRIND;
}
if (!strcasecmp(argv[2], "all")) {
int numtests = sizeof(redisTests)/sizeof(struct redisTest);
for (j = 0; j < numtests; j++) {
redisTests[j].failed = (redisTests[j].proc(argc,argv,flags) != 0);
}
/* Report tests result */
int failed_num = 0;
for (j = 0; j < numtests; j++) {
if (redisTests[j].failed) {
failed_num++;
printf("[failed] Test - %s\n", redisTests[j].name);
} else {
printf("[ok] Test - %s\n", redisTests[j].name);
}
}
printf("%d tests, %d passed, %d failed\n", numtests,
numtests-failed_num, failed_num);
return failed_num == 0 ? 0 : 1;
} else {
redisTestProc *proc = getTestProcByName(argv[2]);
if (!proc) return -1; /* test not found */
return proc(argc,argv,flags);
}
return 0;
}
#endif
/* We need to initialize our libraries, and the server configuration. */
#ifdef INIT_SETPROCTITLE_REPLACEMENT
spt_init(argc, argv);
#endif
tzset(); /* Populates 'timezone' global. */
zmalloc_set_oom_handler(redisOutOfMemoryHandler);
/* To achieve entropy, in case of containers, their time() and getpid() can
* be the same. But value of tv_usec is fast enough to make the difference */
gettimeofday(&tv,NULL);
srand(time(NULL)^getpid()^tv.tv_usec);
srandom(time(NULL)^getpid()^tv.tv_usec);
init_genrand64(((long long) tv.tv_sec * 1000000 + tv.tv_usec) ^ getpid());
crc64_init();
/* Store umask value. Because umask(2) only offers a set-and-get API we have
* to reset it and restore it back. We do this early to avoid a potential
* race condition with threads that could be creating files or directories.
*/
umask(server.umask = umask(0777));
uint8_t hashseed[16];
getRandomBytes(hashseed,sizeof(hashseed));
dictSetHashFunctionSeed(hashseed);
char *exec_name = strrchr(argv[0], '/');
if (exec_name == NULL) exec_name = argv[0];
server.sentinel_mode = checkForSentinelMode(argc,argv, exec_name);
initServerConfig();
ACLInit(); /* The ACL subsystem must be initialized ASAP because the
basic networking code and client creation depends on it. */
moduleInitModulesSystem();
connTypeInitialize();
/* Store the executable path and arguments in a safe place in order
* to be able to restart the server later. */
server.executable = getAbsolutePath(argv[0]);
server.exec_argv = zmalloc(sizeof(char*)*(argc+1));
server.exec_argv[argc] = NULL;
for (j = 0; j < argc; j++) server.exec_argv[j] = zstrdup(argv[j]);
/* We need to init sentinel right now as parsing the configuration file
* in sentinel mode will have the effect of populating the sentinel
* data structures with master nodes to monitor. */
if (server.sentinel_mode) {
initSentinelConfig();
initSentinel();
}
/* Check if we need to start in redis-check-rdb/aof mode. We just execute
* the program main. However the program is part of the Redis executable
* so that we can easily execute an RDB check on loading errors. */
if (strstr(exec_name,"redis-check-rdb") != NULL)
redis_check_rdb_main(argc,argv,NULL);
else if (strstr(exec_name,"redis-check-aof") != NULL)
redis_check_aof_main(argc,argv);
if (argc >= 2) {
j = 1; /* First option to parse in argv[] */
sds options = sdsempty();
/* Handle special options --help and --version */
if (strcmp(argv[1], "-v") == 0 ||
strcmp(argv[1], "--version") == 0) version();
if (strcmp(argv[1], "--help") == 0 ||
strcmp(argv[1], "-h") == 0) usage();
if (strcmp(argv[1], "--test-memory") == 0) {
if (argc == 3) {
memtest(atoi(argv[2]),50);
exit(0);
} else {
fprintf(stderr,"Please specify the amount of memory to test in megabytes.\n");
fprintf(stderr,"Example: ./redis-server --test-memory 4096\n\n");
exit(1);
}
} if (strcmp(argv[1], "--check-system") == 0) {
exit(syscheck() ? 0 : 1);
}
/* Parse command line options
* Precedence wise, File, stdin, explicit options -- last config is the one that matters.
*
* First argument is the config file name? */
if (argv[1][0] != '-') {
/* Replace the config file in server.exec_argv with its absolute path. */
server.configfile = getAbsolutePath(argv[1]);
zfree(server.exec_argv[1]);
server.exec_argv[1] = zstrdup(server.configfile);
j = 2; // Skip this arg when parsing options
}
sds *argv_tmp;
int argc_tmp;
int handled_last_config_arg = 1;
while(j < argc) {
/* Either first or last argument - Should we read config from stdin? */
if (argv[j][0] == '-' && argv[j][1] == '\0' && (j == 1 || j == argc-1)) {
config_from_stdin = 1;
}
/* All the other options are parsed and conceptually appended to the
* configuration file. For instance --port 6380 will generate the
* string "port 6380\n" to be parsed after the actual config file
* and stdin input are parsed (if they exist).
* Only consider that if the last config has at least one argument. */
else if (handled_last_config_arg && argv[j][0] == '-' && argv[j][1] == '-') {
/* Option name */
if (sdslen(options)) options = sdscat(options,"\n");
/* argv[j]+2 for removing the preceding `--` */
options = sdscat(options,argv[j]+2);
options = sdscat(options," ");
argv_tmp = sdssplitargs(argv[j], &argc_tmp);
if (argc_tmp == 1) {
/* Means that we only have one option name, like --port or "--port " */
handled_last_config_arg = 0;
if ((j != argc-1) && argv[j+1][0] == '-' && argv[j+1][1] == '-' &&
!strcasecmp(argv[j], "--save"))
{
/* Special case: handle some things like `--save --config value`.
* In this case, if next argument starts with `--`, we will reset
* handled_last_config_arg flag and append an empty "" config value
* to the options, so it will become `--save "" --config value`.
* We are doing it to be compatible with pre 7.0 behavior (which we
* break it in #10660, 7.0.1), since there might be users who generate
* a command line from an array and when it's empty that's what they produce. */
options = sdscat(options, "\"\"");
handled_last_config_arg = 1;
}
else if ((j == argc-1) && !strcasecmp(argv[j], "--save")) {
/* Special case: when empty save is the last argument.
* In this case, we append an empty "" config value to the options,
* so it will become `--save ""` and will follow the same reset thing. */
options = sdscat(options, "\"\"");
}
else if ((j != argc-1) && argv[j+1][0] == '-' && argv[j+1][1] == '-' &&
!strcasecmp(argv[j], "--sentinel"))
{
/* Special case: handle some things like `--sentinel --config value`.
* It is a pseudo config option with no value. In this case, if next
* argument starts with `--`, we will reset handled_last_config_arg flag.
* We are doing it to be compatible with pre 7.0 behavior (which we
* break it in #10660, 7.0.1). */
options = sdscat(options, "");
handled_last_config_arg = 1;
}
else if ((j == argc-1) && !strcasecmp(argv[j], "--sentinel")) {
/* Special case: when --sentinel is the last argument.
* It is a pseudo config option with no value. In this case, do nothing.
* We are doing it to be compatible with pre 7.0 behavior (which we
* break it in #10660, 7.0.1). */
options = sdscat(options, "");
}
} else {
/* Means that we are passing both config name and it's value in the same arg,
* like "--port 6380", so we need to reset handled_last_config_arg flag. */
handled_last_config_arg = 1;
}
sdsfreesplitres(argv_tmp, argc_tmp);
} else {
/* Option argument */
options = sdscatrepr(options,argv[j],strlen(argv[j]));
options = sdscat(options," ");
handled_last_config_arg = 1;
}
j++;
}
loadServerConfig(server.configfile, config_from_stdin, options);
if (server.sentinel_mode) loadSentinelConfigFromQueue();
sdsfree(options);
}
if (server.sentinel_mode) sentinelCheckConfigFile();
/* Do system checks */
#ifdef __linux__
linuxMemoryWarnings();
sds err_msg = NULL;
if (checkXenClocksource(&err_msg) < 0) {
serverLog(LL_WARNING, "WARNING %s", err_msg);
sdsfree(err_msg);
}
#if defined (__arm64__)
int ret;
if ((ret = checkLinuxMadvFreeForkBug(&err_msg)) <= 0) {
if (ret < 0) {
serverLog(LL_WARNING, "WARNING %s", err_msg);
sdsfree(err_msg);
} else
serverLog(LL_WARNING, "Failed to test the kernel for a bug that could lead to data corruption during background save. "
"Your system could be affected, please report this error.");
if (!checkIgnoreWarning("ARM64-COW-BUG")) {
serverLog(LL_WARNING,"Redis will now exit to prevent data corruption. "
"Note that it is possible to suppress this warning by setting the following config: ignore-warnings ARM64-COW-BUG");
exit(1);
}
}
#endif /* __arm64__ */
#endif /* __linux__ */
/* Daemonize if needed */
server.supervised = redisIsSupervised(server.supervised_mode);
int background = server.daemonize && !server.supervised;
if (background) daemonize();
serverLog(LL_NOTICE, "oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo");
serverLog(LL_NOTICE,
"Redis version=%s, bits=%d, commit=%s, modified=%d, pid=%d, just started",
REDIS_VERSION,
(sizeof(long) == 8) ? 64 : 32,
redisGitSHA1(),
strtol(redisGitDirty(),NULL,10) > 0,
(int)getpid());
if (argc == 1) {
serverLog(LL_WARNING, "Warning: no config file specified, using the default config. In order to specify a config file use %s /path/to/redis.conf", argv[0]);
} else {
serverLog(LL_NOTICE, "Configuration loaded");
}
initServer();
if (background || server.pidfile) createPidFile();
if (server.set_proc_title) redisSetProcTitle(NULL);
redisAsciiArt();
checkTcpBacklogSettings();
if (server.cluster_enabled) {
clusterInit();
}
if (!server.sentinel_mode) {
moduleInitModulesSystemLast();
moduleLoadFromQueue();
}
ACLLoadUsersAtStartup();
initListeners();
if (server.cluster_enabled) {
clusterInitListeners();
}
InitServerLast();
if (!server.sentinel_mode) {
/* Things not needed when running in Sentinel mode. */
serverLog(LL_NOTICE,"Server initialized");
aofLoadManifestFromDisk();
loadDataFromDisk();
aofOpenIfNeededOnServerStart();
aofDelHistoryFiles();
if (server.cluster_enabled) {
serverAssert(verifyClusterConfigWithData() == C_OK);
}
for (j = 0; j < CONN_TYPE_MAX; j++) {
connListener *listener = &server.listeners[j];
if (listener->ct == NULL)
continue;
serverLog(LL_NOTICE,"Ready to accept connections %s", listener->ct->get_type(NULL));
}
if (server.supervised_mode == SUPERVISED_SYSTEMD) {
if (!server.masterhost) {
redisCommunicateSystemd("STATUS=Ready to accept connections\n");
} else {
redisCommunicateSystemd("STATUS=Ready to accept connections in read-only mode. Waiting for MASTER <-> REPLICA sync\n");
}
redisCommunicateSystemd("READY=1\n");
}
} else {
sentinelIsRunning();
if (server.supervised_mode == SUPERVISED_SYSTEMD) {
redisCommunicateSystemd("STATUS=Ready to accept connections\n");
redisCommunicateSystemd("READY=1\n");
}
}
/* Warning the user about suspicious maxmemory setting. */
if (server.maxmemory > 0 && server.maxmemory < 1024*1024) {
serverLog(LL_WARNING,"WARNING: You specified a maxmemory value that is less than 1MB (current value is %llu bytes). Are you sure this is what you really want?", server.maxmemory);
}
redisSetCpuAffinity(server.server_cpulist);
setOOMScoreAdj(-1);
aeMain(server.el);
aeDeleteEventLoop(server.el);
return 0;
}**逐段阅读抓手：**main是生命周期入口；具体命令路径要进入networking.c、server.c和各t_*.c。
0.2 Redis的单线程到底指什么
7.2.6核心普通命令执行与共享键空间修改主要在主线程，网络读写可以在配置条件下使用I/O线程；BIO线程处理部分fsync、关闭文件和惰性释放，RDB/AOF重写可用子进程。模块等扩展还有自身约束。
因此“Redis进程只有一个线程”和“开I/O线程后所有命令都并行执行”都不准确。要看线程在哪个阶段接管，以及结果何时回到主线程执行。长命令、脚本和大对象处理依然可能阻塞核心命令路径。
图02
源码对照：networking.c · 7.2.6 · L4173–L4217。连续节选，窗口可能结束于函数中间；完整函数及调用方见链接。
Redis 7.2.6 · src/networking.c · L4173–L4217
void *IOThreadMain(void *myid) {
/* The ID is the thread number (from 0 to server.io_threads_num-1), and is
* used by the thread to just manipulate a single sub-array of clients. */
long id = (unsigned long)myid;
char thdname[16];
snprintf(thdname, sizeof(thdname), "io_thd_%ld", id);
redis_set_thread_title(thdname);
redisSetCpuAffinity(server.server_cpulist);
makeThreadKillable();
while(1) {
/* Wait for start */
for (int j = 0; j < 1000000; j++) {
if (getIOPendingCount(id) != 0) break;
}
/* Give the main thread a chance to stop this thread. */
if (getIOPendingCount(id) == 0) {
pthread_mutex_lock(&io_threads_mutex[id]);
pthread_mutex_unlock(&io_threads_mutex[id]);
continue;
}
serverAssert(getIOPendingCount(id) != 0);
/* Process: note that the main thread will never touch our list
* before we drop the pending count to 0. */
listIter li;
listNode *ln;
listRewind(io_threads_list[id],&li);
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
if (io_threads_op == IO_THREADS_OP_WRITE) {
writeToClient(c,0);
} else if (io_threads_op == IO_THREADS_OP_READ) {
readQueryFromClient(c->conn);
} else {
serverPanic("io_threads_op value is unknown");
}
}
listEmpty(io_threads_list[id]);
setIOPendingCount(id, 0);
}
}**逐段阅读抓手：**关注IO_THREADS_OP_READ/WRITE以及CLIENT_PENDING_COMMAND；线程参与I/O不等于并行修改数据库。
0.3 目录怎样对应职责
server.c负责初始化、周期任务、命令调度和传播；networking.c负责客户端与协议；db.c负责键空间；object.c和各结构文件管理编码；t_string/t_hash等实现命令；rdb/aof负责持久化；replication、sentinel、cluster分别负责复制、监控切换和分片。
从一个命令的proc函数出发，沿lookup→结构操作→signalModifiedKey→dirty→传播读，能同时理解命令语义与后台影响。redisObject的type和encoding是跨文件阅读的共同坐标。
图03
源码对照：server.h · 7.2.6 · L900–L918。连续节选，窗口可能结束于函数中间；完整函数及调用方见链接。
Redis 7.2.6 · src/server.h · L900–L918
struct redisObject {
unsigned type:4;
unsigned encoding:4;
unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
* LFU data (least significant 8 bits frequency
* and most significant 16 bits access time). */
int refcount;
void *ptr;
};
/* The string name for an object's type as listed above
* Native types are checked against the OBJ_STRING, OBJ_LIST, OBJ_* defines,
* and Module types have their registered name returned. */
char *getObjectTypeName(robj*);
/* Macro used to initialize a Redis object allocated on the stack.
* Note that this macro is taken near the structure definition to make sure
* we'll update it when the structure is changed, to avoid bugs like
* bug #85 introduced exactly in this way. */**逐段阅读抓手：**type是用户数据类型，encoding是当前底层表示；相同type可有多个encoding。
0.4 6.x与7.x对照：6.x已有I/O线程，7.x也不等于命令多线程
| 维度 | 固定6.2.14 | 固定7.2.6 |
|---|---|---|
| 实现 | 6.2.14可配置网络I/O线程；命令执行回到主线程。 | 7.2.6保留此分工，并演进线程管理、连接和读写处理；本文线程结论限定这两份基线。 |
**变化原因（固定源码分析）：**网络读写与共享键空间执行需要分清，前者可以分担CPU而后者仍按核心模型协调。不能把I/O线程功能首次引入时间错误归到7.x。
**边界：**只看进程线程数不能判定普通命令是否并行；8.x后续实现需重新固定版本读。
图04
6.2.14源码：networking.c · L3763–L3838，连续节选。
Redis 6.2.14 · src/networking.c · L3763–L3838
/* When threaded I/O is also enabled for the reading + parsing side, the
* readable handler will just put normal clients into a queue of clients to
* process (instead of serving them synchronously). This function runs
* the queue using the I/O threads, and process them in order to accumulate
* the reads in the buffers, and also parse the first command available
* rendering it in the client structures. */
int handleClientsWithPendingReadsUsingThreads(void) {
if (!server.io_threads_active || !server.io_threads_do_reads) return 0;
int processed = listLength(server.clients_pending_read);
if (processed == 0) return 0;
/* Distribute the clients across N different lists. */
listIter li;
listNode *ln;
listRewind(server.clients_pending_read,&li);
int item_id = 0;
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
int target_id = item_id % server.io_threads_num;
listAddNodeTail(io_threads_list[target_id],c);
item_id++;
}
/* Give the start condition to the waiting threads, by setting the
* start condition atomic var. */
io_threads_op = IO_THREADS_OP_READ;
for (int j = 1; j < server.io_threads_num; j++) {
int count = listLength(io_threads_list[j]);
setIOPendingCount(j, count);
}
/* Also use the main thread to process a slice of clients. */
listRewind(io_threads_list[0],&li);
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
readQueryFromClient(c->conn);
}
listEmpty(io_threads_list[0]);
/* Wait for all the other threads to end their work. */
while(1) {
unsigned long pending = 0;
for (int j = 1; j < server.io_threads_num; j++)
pending += getIOPendingCount(j);
if (pending == 0) break;
}
/* Run the list of clients again to process the new buffers. */
while(listLength(server.clients_pending_read)) {
ln = listFirst(server.clients_pending_read);
client *c = listNodeValue(ln);
c->flags &= ~CLIENT_PENDING_READ;
listDelNode(server.clients_pending_read,ln);
serverAssert(!(c->flags & CLIENT_BLOCKED));
if (processPendingCommandsAndResetClient(c) == C_ERR) {
/* If the client is no longer valid, we avoid
* processing the client later. So we just go
* to the next. */
continue;
}
processInputBuffer(c);
/* We may have pending replies if a thread readQueryFromClient() produced
* replies and did not install a write handler (it can't).
*/
if (!(c->flags & CLIENT_PENDING_WRITE) && clientHasPendingReplies(c))
clientInstallWriteHandler(c);
}
/* Update processed count on server */
server.stat_io_reads_processed += processed;
return processed;
}7.2.6源码：networking.c · L4431–L4523，连续节选。
Redis 7.2.6 · src/networking.c · L4431–L4523
/* When threaded I/O is also enabled for the reading + parsing side, the
* readable handler will just put normal clients into a queue of clients to
* process (instead of serving them synchronously). This function runs
* the queue using the I/O threads, and process them in order to accumulate
* the reads in the buffers, and also parse the first command available
* rendering it in the client structures.
* This function achieves thread safety using a fan-out -> fan-in paradigm:
* Fan out: The main thread fans out work to the io-threads which block until
* setIOPendingCount() is called with a value larger than 0 by the main thread.
* Fan in: The main thread waits until getIOPendingCount() returns 0. Then
* it can safely perform post-processing and return to normal synchronous
* work. */
int handleClientsWithPendingReadsUsingThreads(void) {
if (!server.io_threads_active || !server.io_threads_do_reads) return 0;
int processed = listLength(server.clients_pending_read);
if (processed == 0) return 0;
/* Distribute the clients across N different lists. */
listIter li;
listNode *ln;
listRewind(server.clients_pending_read,&li);
int item_id = 0;
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
int target_id = item_id % server.io_threads_num;
listAddNodeTail(io_threads_list[target_id],c);
item_id++;
}
/* Give the start condition to the waiting threads, by setting the
* start condition atomic var. */
io_threads_op = IO_THREADS_OP_READ;
for (int j = 1; j < server.io_threads_num; j++) {
int count = listLength(io_threads_list[j]);
setIOPendingCount(j, count);
}
/* Also use the main thread to process a slice of clients. */
listRewind(io_threads_list[0],&li);
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
readQueryFromClient(c->conn);
}
listEmpty(io_threads_list[0]);
/* Wait for all the other threads to end their work. */
while(1) {
unsigned long pending = 0;
for (int j = 1; j < server.io_threads_num; j++)
pending += getIOPendingCount(j);
if (pending == 0) break;
}
io_threads_op = IO_THREADS_OP_IDLE;
/* Run the list of clients again to process the new buffers. */
while(listLength(server.clients_pending_read)) {
ln = listFirst(server.clients_pending_read);
client *c = listNodeValue(ln);
listDelNode(server.clients_pending_read,ln);
c->pending_read_list_node = NULL;
serverAssert(!(c->flags & CLIENT_BLOCKED));
if (beforeNextClient(c) == C_ERR) {
/* If the client is no longer valid, we avoid
* processing the client later. So we just go
* to the next. */
continue;
}
/* Once io-threads are idle we can update the client in the mem usage */
updateClientMemUsageAndBucket(c);
if (processPendingCommandAndInputBuffer(c) == C_ERR) {
/* If the client is no longer valid, we avoid
* processing the client later. So we just go
* to the next. */
continue;
}
/* We may have pending replies if a thread readQueryFromClient() produced
* replies and did not put the client in pending write queue (it can't).
*/
if (!(c->flags & CLIENT_PENDING_WRITE) && clientHasPendingReplies(c))
putClientInPendingWriteQueue(c);
}
/* Update processed count on server */
server.stat_io_reads_processed += processed;
return processed;
}**对照抓手：**如果只是字段重排或函数拆分，说明语义延续；如果新增后端、确认点或协议，则明确它何时启用、状态存在哪里、失败怎样收尾。
本章纸面推演
第一轮只读SET→对象→db字典→回复与传播；第二轮补RDB/AOF、复制、过期与淘汰；第三轮读Sentinel/Cluster、事务与Streams。遇到结构转换，先写明对象type和encoding，别只背命令名。