Redis WATCH原理 - 元素码农

▶Redis核心
- ▶数据结构
- ▶内存管理
  - 内存分配策略
  - 淘汰算法
▶持久化
- ▶RDB机制
  - 快照生成原理
  - 文件格式解析
- ▶AOF机制
  - 命令追加策略
  - 重写过程分析
▶高可用
- ▶主从复制
  - SYNC原理
  - 增量复制
- ▶哨兵机制
  - 故障检测
  - 领导选举
▶高级特性
- ▶事务系统
  - ACID实现
  - WATCH原理
- ▶Lua脚本
  - 沙盒环境
  - 执行管道
▶实战问题
- ▶缓存问题
- ▶数据一致性
  - 读写一致性
  - 双写一致性
- ▶性能优化
  - 大key处理
  - 热点key优化

发布时间: 2025-03-22 10:49

↑

# Redis WATCH原理

## 引言

WATCH是Redis事务中的乐观锁机制，通过监视一个或多个键的变化来决定事务是否执行。本文将深入分析WATCH命令的实现原理、使用场景以及最佳实践。

## 基本概念

### 1. 乐观锁

```c
// 监视键结构
typedef struct watchedKey {
    robj *key;           // 被监视的键
    redisDb *db;         // 所属数据库
} watchedKey;

// 客户端监视状态
typedef struct redisClient {
    list *watched_keys;   // 监视键列表
    int flags;            // 状态标志
} redisClient;
```

### 2. 命令格式

```redis
# 监视键
WATCH key [key ...]

# 取消监视
UNWATCH

# 事务示例
WATCH mykey
MULTI
SET mykey newvalue
EXEC
```

## 实现原理

### 1. 监视机制

```c
// WATCH命令实现
void watchCommand(redisClient *c) {
    int j;
    
    /* 检查是否已经在事务中 */
    if (c->flags & CLIENT_MULTI) {
        addReplyError(c,"WATCH inside MULTI is not allowed");
        return;
    }
    
    /* 遍历要监视的键 */
    for (j = 1; j < c->argc; j++)
        watchForKey(c,c->argv[j]);
    addReply(c,shared.ok);
}

// 监视键实现
void watchForKey(redisClient *c, robj *key) {
    list *clients = NULL;
    listIter li;
    listNode *ln;
    watchedKey *wk;
    
    /* 检查是否已经监视了该键 */
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {
        wk = listNodeValue(ln);
        if (wk->db == c->db && equalStringObjects(key,wk->key))
            return;
    }
    
    /* 将键添加到监视列表 */
    clients = dictFetchValue(c->db->watched_keys,key);
    if (!clients) {
        clients = listCreate();
        dictAdd(c->db->watched_keys,key,clients);
        incrRefCount(key);
    }
    listAddNodeTail(clients,c);
    
    /* 将监视信息添加到客户端 */
    wk = zmalloc(sizeof(watchedKey));
    wk->key = key;
    wk->db = c->db;
    incrRefCount(key);
    listAddNodeTail(c->watched_keys,wk);
}
```

### 2. 修改检测

```c
// 键修改通知
void touchWatchedKey(redisDb *db, robj *key) {
    list *clients;
    listIter li;
    listNode *ln;
    
    /* 获取监视该键的客户端列表 */
    if ((clients = dictFetchValue(db->watched_keys,key)) != NULL) {
        /* 遍历客户端列表 */
        listRewind(clients,&li);
        while((ln = listNext(&li))) {
            redisClient *c = listNodeValue(ln);
            
            /* 标记客户端的CAS失败 */
            c->flags |= CLIENT_DIRTY_CAS;
        }
    }
}
```

### 3. 事务执行

```c
// 事务执行检查
int execCommand(redisClient *c) {
    int j;
    robj **orig_argv;
    int orig_argc;
    struct redisCommand *orig_cmd;
    
    /* 检查是否在事务状态 */
    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c,"EXEC without MULTI");
        return C_ERR;
    }
    
    /* 检查CAS是否失败 */
    if (c->flags & CLIENT_DIRTY_CAS) {
        addReply(c,shared.nullmultibulk);
        discardTransaction(c);
        return C_OK;
    }
    
    /* 执行事务命令 */
    unwatchAllKeys(c);
    orig_argv = c->argv;
    orig_argc = c->argc;
    orig_cmd = c->cmd;
    addReplyMultiBulkLen(c,c->mstate.count);
    for (j = 0; j < c->mstate.count; j++) {
        c->argc = c->mstate.commands[j].argc;
        c->argv = c->mstate.commands[j].argv;
        c->cmd = c->mstate.commands[j].cmd;
        call(c,CMD_CALL_FULL);
    }
    
    /* 清理事务状态 */
    discardTransaction(c);
    return C_OK;
}
```

## 性能优化

### 1. 内存优化

```c
// 监视键清理
void unwatchAllKeys(redisClient *c) {
    listIter li;
    listNode *ln;
    
    if (listLength(c->watched_keys) == 0) return;
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {
        watchedKey *wk = listNodeValue(ln);
        list *clients = dictFetchValue(wk->db->watched_keys,wk->key);
        listNode *ln;
        
        ln = listSearchKey(clients,c);
        listDelNode(clients,ln);
        if (listLength(clients) == 0) {
            dictDelete(wk->db->watched_keys,wk->key);
        }
    }
    listRelease(c->watched_keys);
    c->watched_keys = listCreate();
}
```

### 2. 并发优化

```c
// 并发访问控制
void signalModifiedKey(redisDb *db, robj *key) {
    touchWatchedKey(db,key);
}

// 批量修改优化
void signalFlushedDb(int dbid) {
    touchWatchedKeysOnFlush(dbid);
}
```

### 3. 命令优化

```c
// 命令执行优化
int processCommand(redisClient *c) {
    /* 检查是否需要传播命令 */
    int flags = getCommandFlags(c->cmd);
    
    /* 处理WATCH命令 */
    if (flags & CMD_WATCH)
        c->flags |= CLIENT_MONITOR;
    
    /* 执行命令 */
    call(c,CMD_CALL_FULL);
    
    /* 传播命令 */
    if (flags & CMD_WRITE) {
        signalModifiedKey(c->db,c->argv[1]);
    }
    return C_OK;
}
```

## 监控和维护

### 1. 状态监控

```redis
# 查看事务状态
INFO commandstats | grep watch

# 查看监视键数量
INFO stats | grep watch_keys
```

### 2. 性能监控

```redis
# 监控CAS失败次数
INFO stats | grep cas_misses

# 监控内存使用
INFO memory | grep watch
```

### 3. 错误处理

```redis
# 检查事务错误
INFO stats | grep rejected_exec

# 查看命令错误
INFO stats | grep rejected_commands
```

## 最佳实践

### 1. 使用建议

- 合理设置监视键数量
- 避免长时间持有监视
- 及时清理无用监视

### 2. 性能优化建议

- 减少监视键范围
- 控制事务大小
- 优化命令顺序

### 3. 错误处理建议

- 实现重试机制
- 记录错误日志
- 设置超时控制

### 4. 监控建议

- 监控CAS失败率
- 跟踪性能指标
- 设置告警阈值

## 常见问题

### 1. 事务失败

- 监视键被修改
- 命令语法错误
- 内存不足

### 2. 性能问题

- 监视键过多
- 事务队列过大
- 命令执行慢

### 3. 并发问题

- 键值冲突
- 死锁风险
- 超时处理

## 总结

WATCH命令是Redis实现乐观锁的核心机制，通过监视键的变化来保证事务的一致性。在实际应用中，需要根据业务场景合理使用WATCH命令，并通过优化配置和监控机制来确保系统的可靠性和性能。