Node.js 中 Redis 的并发问题

Question

问题

我的 Redis 中有一个名为

stocks

的密钥，其值为 1000。假设有 300 个客户同时请求购买（每个客户 100 只股票）。最终应该只有 10 位顾客能够购买。

解决方案

我知道它不起作用，但假设我的

buy

函数是这样的：

/**
 * @param {import("redis").RedisClientType} instance
 * @param {string} clientId
 * @param {number} n
 */
async function buy(instance, clientId, n) {
  // --- (1) ----
  // Get current number of stocks
  let stocks = await instance.GET("stocks");
  stocks = parseInt(stocks);

  // --- (2) ----
  // Validate if the stocks remaining are enough to be bought
  if (stocks < n) {
    console.log("error: company does not have enough stocks");
    return new Error("error: company does not have enough stocks");
  }

  // --- (3) ----
  // Update the current stocks of the company and log who has bought stocks
  await instance.INCRBY("stocks", -n);
  console.log("client @%s has bought %s stocks successfully!", clientId, n);
}

为了测试它，我编写了一个调用

buy

函数 300 次的函数：

const redis = require("redis");
const crypto = require("crypto");
const { buy } = require("./buy");

async function main(customers = 300) {
  const instance = await redis
    .createClient({ url: "redis://localhost:6379" })
    .connect();

  // --- (1) ----
  // set stocks
  await instance.SET("stocks", 1000);

  // --- (2) ----
  // buy 100 stocks concurrentlly for each customer
  let pool = [];
  for (let i = 0; i < customers; i++) {
    let userId = crypto.randomBytes(4).toString("hex");
    pool.push(buy_v3(instance, userId, 100));
  }
  await Promise.all(pool);

  // --- (3) ----
  // Get the remaining stocks
  let shares = await instance.GET("stocks");
  console.log("the number of free shares the company has is: %s", shares);

  await instance.disconnect();
}
main();

输出：

...
error: company does not have enough stocks
error: company does not have enough stocks
error: company does not have enough stocks
error: company does not have enough stocks
the number of free stocks the company has is: -29000

正如我所说，它不起作用，但为了解决这个问题，我使用了这种方法：

/**
 * @param {import("redis").RedisClientType} instance
 * @param {string} clientId
 * @param {number} n
 */
async function buy(instance, clientId, n) {
  try {
    await instance.executeIsolated(async (client) => {
      // --- (1) ----
      // Get current number of stocks
      let stocks = await client.GET("stocks");
      stocks = parseInt(stocks);

      // --- (2) ----
      // Validate if the stocks remaining are enough to be bought
      if (stocks < n) {
        throw new Error("error: company does not have enough stocks");
      }

      // --- (3) ----
      // Update the current stocks of the company
      await client.INCRBY("stocks", -n);
    });

    console.log("client @%s has bought %s stocks successfully!", clientId, n);
  } catch (err) {
    console.log(err.message);
  }
}

如果你再次测试它，你会看到类似这样的东西：

...
error: company does not have enough stocks
error: company does not have enough stocks
error: company does not have enough stocks
error: company does not have enough stocks
the number of free stocks the company has is: 0

这意味着它可以正常工作。

问题

上述解决方案效果很好，但我对

executeIsolated

功能有点困惑。据我所知，它只是创建一个新连接（您可以看到here），当您想在独占连接（如

watch

命令）上运行命令时，它非常有用。

谁能解释一下

executeIsolated

在我的案例中的确切作用是什么？

Answer 1

问题是，当存在并发请求时，无法保证第 n 个请求的

SET

将在第 n+1 个请求的

GET

之前运行。例如，如果有 2 个并发请求，则应按以下顺序执行命令：

> GET stocks
"100"
> INCRBY stocks -100
(integer) 0
> GET stocks
"0"

但它们可能会按以下顺序执行：

> GET stocks
"100"
> GET stocks
"100"
> INCRBY stocks -100
(integer) 0
> INCRBY stocks -100
(integer) -100

为了修复它，您应该使用 Redis 函数（自 Redis 7.0 起可用）或 Lua 脚本，如下所示：

local stocks = redis.call('GET', KEYS[1])
if stocks < ARGS[1] then
  return redis.error_reply('company does not have enough stocks')
end

redis.call('SET', KEYS[1], stocks - ARGS[1])
return redis.status_reply('OK')

关于为什么使用

executeIsolated

“修复”问题 - 可能有两个原因：

1. 池大小为1，这有效地创建了一个队列
2. 池中没有“空闲”连接，创建新连接所需的时间超过执行所需的时间

   GET

   ..

希望这是有道理的..