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
以上です。