如何在NodeJS游戏服务器中防止Redis GameStatus修改时被覆盖并实现Pub/Sub广播

针对多用户并发修改Redis中GameStatus导致的覆盖问题，可以通过以下Redis原生机制结合NodeJS实现来避免：

1. 用WATCH+事务实现乐观锁 #

Redis的WATCH命令可以监听指定key，在事务执行前如果该key被其他客户端修改，事务会自动失败。你可以在失败后重试操作，确保基于最新的数据进行修改。

示例代码（使用NodeJS的redis库）：

const redis = require('redis');
const client = redis.createClient();

async function updateGameStatus(gameId, updateFn) {
  while (true) {
    try {
      // 监听GameStatus key
      await client.watch(`game:${gameId}:status`);
      // 获取当前状态
      const currentStatus = JSON.parse(await client.get(`game:${gameId}:status`));
      // 执行自定义修改逻辑
      const newStatus = updateFn(currentStatus);
      // 开启事务并提交
      const multi = client.multi();
      multi.set(`game:${gameId}:status`, JSON.stringify(newStatus));
      const result = await multi.exec();
      
      if (result) {
        // 事务执行成功，广播新状态
        await client.publish(`game:${gameId}:updates`, JSON.stringify(newStatus));
        break;
      }
      // 事务失败，重试
    } catch (err) {
      console.error('Update failed, retrying:', err);
    } finally {
      // 解除监听
      await client.unwatch();
    }
  }
}

// 使用示例：修改玩家分数
updateGameStatus('123', (status) => {
  status.players['user1'].score += 10;
  return status;
});

2. 拆分状态为哈希结构，使用原子字段更新 #

如果GameStatus可以拆分为独立字段（比如玩家分数、房间人数等），改用Redis哈希（Hash）存储，用HSET、HINCRBY等原子命令修改单个字段，避免全量覆盖整个状态。

示例代码：

// 初始化哈希结构
await client.hSet(`game:123:status`, {
  'player1_score': 100,
  'player2_score': 80,
  'room_count': 2
});

// 用户1增加分数（原子操作，不会被覆盖）
await client.hIncrBy(`game:123:status`, 'player1_score', 10);
// 用户2修改房间人数（原子操作）
await client.hSet(`game:123:status`, 'room_count', 3);

// 广播更新（可以定期或操作后广播哈希全量数据）
const updatedStatus = await client.hGetAll(`game:123:status`);
await client.publish(`game:123:updates`, JSON.stringify(updatedStatus));

3. 使用分布式锁实现悲观锁 #

通过Redis的SETNX（SET if Not eXists）命令实现分布式锁，确保同一时间只有一个客户端能修改GameStatus。注意设置锁的超时时间，避免客户端崩溃导致死锁。

示例代码：

async function acquireLock(lockKey, timeout = 5000) {
  const lockValue = Date.now() + timeout + 1;
  const result = await client.set(lockKey, lockValue, {
    NX: true,
    EX: Math.ceil(timeout / 1000)
  });
  return result === 'OK' ? lockValue : null;
}

async function releaseLock(lockKey, lockValue) {
  const currentValue = await client.get(lockKey);
  if (currentValue === lockValue.toString()) {
    await client.del(lockKey);
  }
}

async function updateGameStatusWithLock(gameId, updateFn) {
  const lockKey = `game:${gameId}:lock`;
  let lockValue = null;
  try {
    lockValue = await acquireLock(lockKey);
    if (!lockValue) {
      // 没抢到锁，重试
      setTimeout(() => updateGameStatusWithLock(gameId, updateFn), 100);
      return;
    }
    // 获取并修改状态
    const currentStatus = JSON.parse(await client.get(`game:${gameId}:status`));
    const newStatus = updateFn(currentStatus);
    await client.set(`game:${gameId}:status`, JSON.stringify(newStatus));
    // 广播
    await client.publish(`game:123:updates`, JSON.stringify(newStatus));
  } finally {
    if (lockValue) {
      await releaseLock(lockKey, lockValue);
    }
  }
}

4. 用Redis Stream实现串行化修改请求 #

如果修改请求需要按顺序处理，可以将修改任务写入Redis Stream，用消费者组串行处理每个请求，确保状态修改不会并发冲突。

示例代码：

// 初始化消费者组（仅第一次执行）
await client.xGroupCreate(`game:123:stream`, `game_updates_group`, '$', { MKSTREAM: true });

// 生产者：发送修改请求
async function sendUpdateRequest(gameId, updateCmd) {
  // 实际项目建议传递结构化修改指令，而非函数
  await client.xAdd(`game:${gameId}:stream`, '*', updateCmd);
}

// 消费者：串行处理修改请求
async function processUpdates(gameId) {
  while (true) {
    const messages = await client.xReadGroup(`game_updates_group`, `client_1`, [{ key: `game:${gameId}:stream`, id: '>' }], { COUNT: 1, BLOCK: 5000 });
    if (!messages) continue;
    
    for (const message of messages[0].messages) {
      const { playerId, scoreDelta } = message.message;
      const currentStatus = JSON.parse(await client.get(`game:${gameId}:status`));
      // 执行修改
      currentStatus.players[playerId].score += parseInt(scoreDelta);
      await client.set(`game:${gameId}:status`, JSON.stringify(currentStatus));
      await client.publish(`game:${gameId}:updates`, JSON.stringify(currentStatus));
      // 确认消息已处理
      await client.xAck(`game:${gameId}:stream`, `game_updates_group`, message.id);
    }
  }
}

// 启动消费者
processUpdates('123');
// 发送修改请求示例
sendUpdateRequest('123', { playerId: 'user1', scoreDelta: '10' });

内容的提问来源于stack exchange，提问作者Alex Choi