bottleneckとは

タスクスケジューラーとレート制限の機能を提供するnpmパッケージです。
https://www.npmjs.com/package/bottleneck

例えば、Backends For Frontendsから後方のAPIを呼び出す時に、「この閾値以上で呼び出すとAPIサーバーが死ぬから止めてね！」という時など、一気に処理を流したくない時に使えます。（API側で制限かけなよ！っていうツッコミは置いておいて）

テスト用APIの準備

テスト用のAPIをexpressで用意しておきます。

app.ts

import * as moment from "moment";
import * as express from "express";
const app = express();
const sleep = msec => new Promise(resolve => setTimeout(resolve, msec));

app.get("/", async (req, res) => {
  console.log(moment().format("HH:mm:ss.SSS") + " Start " + req.query.sleep);
  await sleep(req.query.sleep);
  console.log(moment().format("HH:mm:ss.SSS") + " End   " + req.query.sleep);
  return res.send(req.query.sleep);
});

app.listen(3000, () => console.log("Example app listening on port 3000"));

何も制限せずにリクエストした場合

test0.ts

import axios from "axios";

async function request(sleep: number): Promise<void> {
  const res = await axios({
    method: "get",
    url: `http://localhost:3000/?sleep=${sleep}`
  });
  console.log(res.data);
}

function main() {
  for (let idx = 1; idx <= 20; idx++) {
    request(idx);
  }
}

main();

APIのログを見ると、一気にリクエストが来たのが分かります。

15:22:00.718 Start 1
15:22:00.721 End   1
15:22:00.727 Start 2
15:22:00.728 Start 3
15:22:00.728 Start 4
15:22:00.728 Start 5
15:22:00.728 Start 6
15:22:00.729 Start 7
15:22:00.729 Start 8
15:22:00.729 Start 9
15:22:00.730 Start 10
15:22:00.730 Start 11
15:22:00.730 Start 12
15:22:00.731 Start 13
15:22:00.731 Start 14
15:22:00.735 Start 15
15:22:00.735 Start 16
15:22:00.736 Start 17
15:22:00.736 Start 18
15:22:00.736 Start 19
15:22:00.736 Start 20
15:22:00.738 End   2
15:22:00.738 End   3
15:22:00.739 End   4
15:22:00.739 End   5
15:22:00.740 End   6
15:22:00.740 End   7
15:22:00.740 End   8
15:22:00.741 End   9
15:22:00.741 End   10
15:22:00.742 End   11
15:22:00.743 End   12
15:22:00.744 End   13
15:22:00.749 End   14
15:22:00.750 End   15
15:22:00.752 End   16
15:22:00.753 End   17
15:22:00.754 End   18
15:22:00.755 End   19
15:22:00.757 End   20

QPSで制限した場合

「このQPS以下でリクエストしてね！」という要件の場合は、minTimeオプションを使用します。
ジョブが開始して、次のジョブが開始するまでの待機時間をミリ秒で指定します。
5QPSに制限する場合は「1000ms / 5request = 200ms」で200になります。

test1.ts

import axios from "axios";
import Bottleneck from "bottleneck";

const limiter = new Bottleneck({
  minTime: 200
});

async function request(sleep: number): Promise<void> {
  const res = await limiter.schedule(() =>
    axios({
      method: "get",
      url: `http://localhost:3000/?sleep=${sleep}`
    })
  );
  console.log(res.data);
}

function main() {
  for (let idx = 1; idx <= 20; idx++) {
    request(idx);
  }
}

main();

実行すると、こんな感じ。

15:39:28.530 Start 1
15:39:28.532 End   1
15:39:28.711 Start 2
15:39:28.714 End   2
15:39:28.909 Start 3
15:39:28.913 End   3
15:39:29.109 Start 4
15:39:29.114 End   4
15:39:29.310 Start 5
15:39:29.316 End   5
15:39:29.510 Start 6
15:39:29.516 End   6
15:39:29.710 Start 7
15:39:29.719 End   7
15:39:29.909 Start 8
15:39:29.918 End   8
15:39:30.109 Start 9
15:39:30.119 End   9
15:39:30.309 Start 10
15:39:30.320 End   10
15:39:30.509 Start 11
15:39:30.521 End   11
15:39:30.710 Start 12
15:39:30.723 End   12
15:39:30.909 Start 13
15:39:30.923 End   13
15:39:31.110 Start 14
15:39:31.125 End   14
15:39:31.310 Start 15
15:39:31.326 End   15
15:39:31.510 Start 16
15:39:31.527 End   16
15:39:31.710 Start 17
15:39:31.728 End   17
15:39:31.910 Start 18
15:39:31.930 End   18
15:39:32.110 Start 19
15:39:32.130 End   19
15:39:32.310 Start 20
15:39:32.338 End   20

同時実行数で制限した場合

「同時に大量にリクエストされるとサーバーのスレッドがががが」とお困りの場合は、maxConcurrentオプションを使いましょう。

test2.ts

import axios from "axios";
import Bottleneck from "bottleneck";

const limiter = new Bottleneck({
  maxConcurrent: 2
});

async function request(sleep: number): Promise<void> {
  const res = await limiter.schedule(() =>
    axios({
      method: "get",
      url: `http://localhost:3000/?sleep=${sleep}`
    })
  );
  console.log(res.data);
}

function main() {
  for (let idx = 1; idx <= 20; idx++) {
    request(2000 + idx);
  }
}

main();

実行すると、こんな感じ。

15:44:46.696 Start 2001
15:44:46.699 Start 2002
15:44:48.697 End   2001
15:44:48.701 End   2002
15:44:48.718 Start 2003
15:44:48.718 Start 2004
15:44:50.721 End   2003
15:44:50.723 End   2004
15:44:50.728 Start 2005
15:44:50.729 Start 2006
15:44:52.733 End   2005
15:44:52.741 End   2006
15:44:52.755 Start 2007
15:44:52.756 Start 2008
15:44:54.763 End   2007
15:44:54.765 End   2008
15:44:54.772 Start 2009
15:44:54.773 Start 2010
15:44:56.781 End   2009
15:44:56.784 End   2010
15:44:56.792 Start 2011
15:44:56.793 Start 2012
15:44:58.803 End   2011
15:44:58.812 End   2012
15:44:58.824 Start 2013
15:44:58.827 Start 2014
15:45:00.837 End   2013
15:45:00.841 End   2014
15:45:00.860 Start 2015
15:45:00.868 Start 2016
15:45:02.882 End   2015
15:45:02.884 End   2016
15:45:02.889 Start 2017
15:45:02.895 Start 2018
15:45:04.912 End   2017
15:45:04.913 End   2018
15:45:04.922 Start 2019
15:45:04.923 Start 2020
15:45:06.942 End   2019
15:45:06.943 End   2020

キューの長さを制限する

実行待ちのジョブが入っているキューの長さは、デフォルトでは制限がありません。
例えば5QPSで制限した場合、キューの50番目に登録されているジョブが始まるのは10秒後になります。
Webアプリケーションであれば、応答が遅くなるのが分かっていれば「503 Service Unavailable」を返したくなると思いますので、そういう場合はキューの長さをhighWaterで指定します。

test3.ts

import axios from "axios";
import Bottleneck from "bottleneck";

const limiter = new Bottleneck({
  maxConcurrent: 2,
  highWater: 10
});

async function request(sleep: number): Promise<void> {
  try {
    const res = await limiter.schedule(() =>
      axios({
        method: "get",
        url: `http://localhost:3000/?sleep=${sleep}`
      })
    );
    console.log(res.data);
  } catch (e) {
    console.log(sleep, e.message);
  }
}

function main() {
  for (let idx = 1; idx <= 20; idx++) {
    request(2000 + idx);
  }
}

main();

実行すると、こんな感じ。リクエスト数が少ないですね。

15:58:59.913 Start 2001
15:58:59.917 Start 2002
15:59:01.914 End   2001
15:59:01.919 End   2002
15:59:01.947 Start 2011
15:59:01.948 Start 2012
15:59:03.958 End   2011
15:59:03.960 End   2012
15:59:03.966 Start 2013
15:59:03.967 Start 2014
15:59:05.980 End   2013
15:59:05.982 End   2014
15:59:05.991 Start 2015
15:59:05.996 Start 2016
15:59:08.006 End   2015
15:59:08.012 End   2016
15:59:08.029 Start 2017
15:59:08.030 Start 2018
15:59:10.046 End   2017
15:59:10.050 End   2018
15:59:10.056 Start 2019
15:59:10.057 Start 2020
15:59:12.076 End   2019
15:59:12.077 End   2020

リクエスト側のログを見ると、３番～１０番がキャンセルされています。
実はキャンセルされるジョブの優先度はstrategyオプションで指定できるのですが、デフォルトではキューの長さより多いジョブが登録された場合、古いジョブの方からキャンセルされます。

2003 'This job has been dropped by Bottleneck'
2004 'This job has been dropped by Bottleneck'
2005 'This job has been dropped by Bottleneck'
2006 'This job has been dropped by Bottleneck'
2007 'This job has been dropped by Bottleneck'
2008 'This job has been dropped by Bottleneck'
2009 'This job has been dropped by Bottleneck'
2010 'This job has been dropped by Bottleneck'
2001
2002
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020

キューが溢れたら新しいジョブからキャンセルする

デフォルトのstrategyでは古い方からキャンセルされました。
逆にキューが一杯になった時に、新しく入れないようにしたい場合は、Bottleneck.strategy.OVERFLOWを指定します。

test4.ts

import axios from "axios";
import Bottleneck from "bottleneck";

const limiter = new Bottleneck({
  maxConcurrent: 2,
  highWater: 10,
  strategy: Bottleneck.strategy.OVERFLOW
});

async function request(sleep: number): Promise<void> {
  try {
    const res = await limiter.schedule(() =>
      axios({
        method: "get",
        url: `http://localhost:3000/?sleep=${sleep}`
      })
    );
    console.log(res.data);
  } catch (e) {
    console.log(sleep, e.message);
  }
}

function main() {
  for (let idx = 1; idx <= 20; idx++) {
    request(2000 + idx);
  }
}

main();

実行すると、新しい方のジョブがキャンセルされているのが分かります。

2013 'This job has been dropped by Bottleneck'
2014 'This job has been dropped by Bottleneck'
2015 'This job has been dropped by Bottleneck'
2016 'This job has been dropped by Bottleneck'
2017 'This job has been dropped by Bottleneck'
2018 'This job has been dropped by Bottleneck'
2019 'This job has been dropped by Bottleneck'
2020 'This job has been dropped by Bottleneck'
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012

おわりに

以前はQPS制限をするのにlimiterとasyncのキューを組み合わせて実装していましたが、bottleneckを使うことで簡単に実装できるようになりました。

今回使用したテストコードはこちらになります。
https://github.com/proyuki02/study-of-bottleneck

以上です。