朝日新聞社・サービス開発部の松山莞太と申します。
Yjsを用いた共同編集ツールを開発している場合、ユーザーがテキストを編集するたびに、サーバーへ「更新内容（update）」を送信します。その結果、大量のDBの書き込みが発生し、サーバーやDBに負荷がかかることがあります。
これに対処するため、DBのスペックを上げる必要がありますが、コストも増大するのでその手段は取りたくないですよね。
今回私は、ユーザーの編集内容の送信間隔を間引くことで対応しました。
送信間隔が開くためリアルタイム性は減ってしまいますが、代わりに負荷を削減できました。
今回の説明ではy-websocketをカスタマイズして、アップデートの送信を間引き、複数の更新を一つにまとめて一括送信する方法を解説します。
他の通信方法を利用する際は、読み替えてカスタマイズしてください。
y-websocketにおける送信部分のデフォルト実装
説明の大前提として、どのようにしてYjsの更新がサーバーに送信されるのかを説明します。
Yjsでは、ユーザーが編集しているドキュメントをYDocオブジェクトとしてクライアント側で管理します。そして、y-websocketは、このYDocが更新されるたびにサーバーへその変更を送信します。
具体的には、doc.on('update', this._updateHandler)というコードで、ドキュメントに更新があったときに、_updateHandler関数が呼ばれるようになっています。
this._updateHandler = (update, origin) => {
// ここでupdateがエンコード（バイナリ化）され、サーバーに送信される。
broadcastMessage(this, encoding.toUint8Array(encoder))
}
const broadcastMessage = (provider, buf) => {
// websocket接続で、データが送信される
const ws = provider.ws
ws.send(buf)
}
this.doc.on('update', this._updateHandler)
つまりは、
- ユーザーが何かを編集し、Docが更新される
-
doc.on("update",...)によって、_updateHandlerが呼ばれる -
_updateHandler内で、update（変更内容）をWebSocketでサーバーへ送信する
この実装では、ユーザーが共同編集を少しでも編集するたびにupdateを送りつけるので、多量の通信が発生し、DBへの負荷が高まります。
DBのCPU稼働率に余裕を持たせたかったり、コストにシビアな場合は、工夫して通信量を削減する必要があります。
負荷を減らす方法
通信負荷を減らすために、短時間に連続して発生するアップデートを「ためて」おき、一定間隔ごとに一括送信します。その際、YjsのmergeUpdatesを使って複数のアップデートを一つにまとめます。
手順は３ステップです。
- アップデートの送信を間引く
- 溜まったアップデートをマージする
- マージされたアップデートを送信する
1. アップデートの送信を間引く
短い間に複数回の編集が行われていても、すぐには送らず、一定間隔でしか送らせないようにします。
これは、スロットリングによって実現できます。myFuncの実行間隔を間引きたいときは次のように書きます。
import { throttle } from 'lodash';
const throttledMyFunc = throttle(
() => {
myFunc()
},
5000 //実行間隔
);
throttledMyFuncは、短期間に繰り返し実行しても5秒に1回しか実行されません。
setInterval(throttledMyFunc, 1000) //1秒ごとに実行
// => 5秒おきにしか実行されない
これにより、関数が高頻度に実行されるのを抑制できます。
共同編集では、編集内容の送信をスロットリングで抑制できます。
const throttledBroadcastMessage = throttle(
() => {
// ...送信処理
},
5000
);
2. 溜まったアップデートをマージする
次に、スロットリングによって溜まったアップデートを一つにマージします。
yjsではmergeUpdatesでアップデートを一つにマージできます。
const mergedUpdate = Y.mergeUpdates([update1, update2, update3])
これを利用します。
updateがトリガーされた際には、まずは_updates配列にアップデートを取り込むようにします。
this._updates = [];
this._updateHandler = (update, origin) => {
// 更新を配列に溜め込む
this._updates.push(update);
this.throttledBroadcastMessage();
};
そして、throttledBroadcastMessage内で_updatesをマージします。
this.throttledBroadcastMessage = () => {
//マージ
const mergedUpdate = Y.mergeUpdates(this._updates)
// リセット
this._updates = [];
// mergedUpdateを送信する処理が続く...
};
3. マッージされたアップデートを送信する
マージしたアップデートを通常通りエンコードして、WebSocketで一括送信します。
this._batchBroadcastUpdates = () => {
// アップデートをマージ
const mergedUpdate = Y.mergeUpdates(this._updates)
this._updates = [];
// エンコード
const encoder = encoding.createEncoder();
encoding.writeVarUint(encoder, messageSync);
syncProtocol.writeUpdate(encoder, mergedUpdate);
// エンコードされたupdateを送信
broadcastMessage(this, encoding.toUint8Array(encoder), false);
};
実装全体
まとめると次のようになります。
// y-websocket
export class WebsocketProvider extends Observable<string> {
constructor(doc, ...) {
// ...省略...
// 溜め込み用の配列
this._updates = [];
// 溜まったアップデートをマージして送信する処理。
this._batchBroadcastUpdates = () => {
// 貯めたアップデートを統合
const mergedUpdate = Y.mergeUpdates(this._updates)
this._updates = [];
// エンコード
const encoder = encoding.createEncoder();
encoding.writeVarUint(encoder, messageSync);
syncProtocol.writeUpdate(encoder, mergedUpdate);
// 一括送信
broadcastMessage(this, encoding.toUint8Array(encoder), false);
};
// スロットリング（5秒に1回だけ実行）
this._throttledBroadcast = throttle(() => {
this._batchBroadcastUpdates();
}, 5000);
// ユーザーが編集した際に動作する関数
this._updateHandler = (update, origin) => {
// 自分以外が原因のupdateの場合のみ処理
if (origin !== this) {
// アップデートを蓄積
this._updates.push(update);
// スロットリングされた関数を呼ぶ
this._throttledBroadcast();
}
};
// YDocのupdateイベントを監視
this.doc.on('update', this._updateHandler);
}
// ...省略...
}
まとめ
デフォルトでは編集ごとに即時にサーバーへ送信するため、DBへの負荷が増大します。
一方で、スロットリングで送信間隔を空けることで、複数アップデートをマージして一括送信することで、負荷とコストを削減できます。
リアルタイム性はやや下がりますが、負荷軽減とコスト抑制できます。
この手法を用いることで、DBへの負荷を緩和して、コストを抑えつつ、共同編集を安定して運用できます。
（サービス開発部・松山莞太）