追記：見逃した方のためにお知らせしておくと、Disqueのソースコードがhttp://github.com/antirez/disqueで公開されています。

ここ数か月、時間の15〜20％ほどを、主に夜や週末から盗み取るようにして、新しいシステムの開発に費やしています。メッセージブローカーで、Disqueと呼んでいます。元の仕様に書かれていた内容の80％は実装済みですが、まだリリースできる状態だとは思えません。出荷できないなら、せめてブログに書こう……というわけで、今回はその始まりと、どんなものなのかを少し紹介します。

〜 最初の一歩 〜

多くの開発者はRedisをメッセージキューとして使っています。Redisの低レベルなプリミティブを隠蔽するライブラリでラップすることもあれば、Redisの生のAPIを使って、単純なその場限りのキューを直接組み立てることもあります。この用途では主に、ブロッキング形式のリスト操作や、リストへのプッシュ操作が使われます。Redisは、そういう使い方をするシステムとしては、どうやら最高であると同時に最悪でもあります。高速で、調査しやすく、導入もしやすく、使いやすいからです。多くの環境では、すでにインフラの一部になっているという利点もあります。一方で、Redisの可変データ構造は不変メッセージとは大きく異なるため、デメリットもあります。RedisのHAやクラスタリングにおけるトレードオフは、大きな可変値を扱うことに完全に最適化されています。しかし、メッセージを扱う場合に同じトレードオフが最善とは限りません。

メッセージブローカーにとって重要な保証の一つは、メッセージが少なくとも1回、あるいは最大でも1回配信されることです。要するに、メッセージが正確に1回だけ配信されたこと（ここでいう配信とは、ワーカーがメッセージを受け取り、かつ処理したことを意味します）を保証するのは実質的に不可能です。そのため選択肢は、メッセージブローカーが0回または1回の配信を保証するか、1回から無限回の配信を保証するかになります。これはそれぞれ「最大1回配送（at-most-once）」セマンティクス、「少なくとも1回配送（at-least-once）」セマンティクスと呼ばれることがよくあります。前者にも用途はありますが、より興味深く実用的なのは後者です。つまり、メッセージが少なくとも1回配信され、障害が発生した場合には複数回配信されることを保証する方式です。

そこで数か月前、レプリケーションもクラスタリングも一切行わないRedisマスターの集合を使い、こうした保証を実現するクライアント側プロトコルについて考え始めました。Redisの使い方を少し変えるだけで、よりよいシステムにできる場合があります。たとえば分散ロックについては、実装自体は簡単でありながら、単一インスタンス＋フェイルオーバー方式より堅牢なアルゴリズムを文書化しようとしました（http://redis.io/topics/distlock）。

しかし数日作業したところ、設計の草案からは、クライアント側のアルゴリズムは複雑すぎて最適でもなく、どうしても実現したかったことの一部は不可能、あるいは非常に困難だということが見えてきました。それなら専用システムとして設計したほうがよさそうでした。Redisにさらに機能を追加するのもよい考えとは思えませんでした。Redisはすでに多くのことをしていますし、メッセージングをきちんと扱うには、Redisの動作方式とはかなり異なるものが必要だったからです。では、世の中にはメッセージブローカーがいくらでもあるのに、なぜ新しいシステムを設計するのでしょうか。多くのユーザーが、この目的のために特別に設計されたシステムではなくRedisを使っていたからです。これは奇妙なことでした。少数なら間違いかもしれません。しかし、これほど多くの人がそうしているなら、何か理由があるはずです。メッセージブローカーの世界を見たとき、多くの人が慣れていたのは、Redisの参入障壁の低さ、簡単なAPI、速度ではなかったのでしょうか。そこに並んでいる解決策は、アプリケーション側に多くを要求する単純すぎるものか、非常に多機能ではあるものの複雑すぎるものか、そのどちらかに見えます。もしかすると、「メッセージングのRedis」に相当するものが入り込む余地はあるのではないでしょうか。

〜 Redisを容赦なくフォークする 〜

人生で初めて、私はいきなりコードを書き始めませんでした。何週間ものあいだ、ときどき設計を見直し、それをRedisのクライアントライブラリではなく新しいシステムへと落とし込み、ユーザーの立場で、どんなメッセージブローカーなら自分はとても満足するだろうかと考えました。最初の用途は変わりませんでした。遅延ジョブです。Disqueは汎用システムですが、設計中の90％の場面で「想定していたユーザー」は、処理すべきジョブになりそうなメッセージを送る問題を解決しようとしている人でした。この用途に反するものは、取り除きました。

設計が固まると、ようやくコードを書き始めました。では、どこから始めるか。「vi main.c」でしょうか。幸い、Redisはある意味、Cで分散システムを書くためのフレームワークでもあります。プロトコル、ネットワークライブラリ、クライアント処理、ノード間のメッセージバスがありました。これをすべてゼロから書き直すのは、途方もない無駄に思えます。同時に、必要ならDisqueがあらゆる細部でRedisから完全に分岐できるようにもしたいと考えていましたし、Redis本体に影響を与えないサイドプロジェクトにもしたいと思っていました。そこで、Redisを実際のフレームワークとRedisの実装に分割するという大仕事に挑む代わりに、より実用的な方法を取りました。コードをフォークし、ソースコードからRedis固有のものをすべて取り除き、骨格だけを残したのです。これで、仕様の実装に取りかかる準備ができました。

〜 Disqueとは何か 〜

それから数か月、たいして集中的でもない作業を続け、コミット数もわずか200ほどでしたが、ようやく単なるおもちゃには見えないシステムができました。何週間ものあいだ本当におもちゃのように見えていたので、話題にすることさえ怖かったのです。ソースツリーを丸ごと削除してしまう可能性が高いと思っていましたから。今ではアイデアの大部分が、テスト付きの動くコードになっています。これなら将来リリースできると、ようやく確信できました。そこで、設計で採用したトレードオフについて話そうと思います。

Disqueは、デフォルトで分散システムです。APシステムなので、Redisのように単一ノードモードと分散モードを用意する意味はありません。Disqueの単一ノードは、ノードが1台だけのクラスタという特殊なケースにすぎません。これは設計上、重要なポイントの一つでした。つまり、障害に耐え、ネットワーク分断にも耐性があり、何台のノードが稼働し続けているかにかかわらず利用可能であること、いわゆるAPです。また、さまざまな状況で本質的にスケールできるシステムにもしたいと考えました。多数のプロデューサーとコンシューマーが多数のキューを使う場合にも、逆に、すべてのプロデューサーとコンシューマーが単一のキューに集中し、そのキューを複数ノードに分散する場合にも対応できるようにしたかったのです。

要件が、あることをはっきり告げていました……Disqueは設計上、大きな犠牲を払うことになる、と。メッセージの順序です。Disqueが提供するのは、ベストエフォートの順序性だけです。しかし、この犠牲によって得られるものも多くあります。トレードオフが興味深いのは、ときに設計の可能性を大きく広げるからです。

このままDisqueがどんなものかを語り続けることもできます。しかし数か月前、Hacker NewsでJacques Chesterが書いたコメントを見つけました。https://news.ycombinator.com/item?id=8709146をご覧ください［追記：Adrianの名前を間違えて引用した箇所を、コピー＆ペーストの際に間違えて切り出してしまいました。Adrian、ごめんなさい！］。私と同じくPivotalで働いているJacquesは、メッセージングシステムごとに機能や特性の組み合わせが大きく異なり、詳細を知らなければ、それぞれの選択肢を評価するのはほとんど不可能だと書いていました。あるシステムが別のシステムより速いとしても、それは実装が優れているからなのか、それとも単に保証がずっと少ないからなのか、判断できないのです。そこで彼は、メッセージングシステムを評価するときに尋ねるべき質問をまとめました。ここでは彼の質問を使い、さらにいくつか質問を加えて、Disqueとは何かを説明します。単なる印象論で終わらず、実際の情報を提供できればと思います。

Q：メッセージは少なくとも1回配信されますか？

Disqueでは、少なくとも1回の配信（デフォルト）と、最大1回の配信を選べます。この設定はメッセージごとに指定できます。最大1回の配信は、少なくとも1回の配信の特殊なケースにすぎません。メッセージの「retry」パラメーターを0にし、メッセージを1台のノードにだけレプリケートすればよいのです。

Q：メッセージはコンシューマーによって確認応答されますか？

はい。メッセージが正しく配信されたとシステムに伝える方法は、コンシューマーが確認応答することだけです。

Q：確認応答されなかったメッセージは複数回配信されますか？

はい。Disqueは「retry」時間の経過後、メッセージを自動的に再配信します。これはメッセージの最大TTLに達するまで、無期限に行われます。メッセージが確認応答されると、その応答はメッセージのコピーを持つノードへ伝播されます。システムが全ノードに通知できたと判断すると、メッセージは最終的にガベージコレクションされ、削除されます。確認応答済みのメッセージは、メモリーが圧迫された場合にも追い出されます。

ノードは、同じメッセージを複数回キューに入れないようベストエフォートのアルゴリズムを実行し、単一配信に近づけようとします。しかし障害発生時には、複数のノードが同じメッセージを同時に何度も再配信する可能性があります。

Q：キューイングは永続的ですか、それとも一時的ですか。

永続的です。

Q：永続性は、すべてのメッセージをまずディスクに書き込むことで実現しますか。それとも、サーバー間でメッセージをレプリケートすることで実現しますか？

Disqueはデフォルトではメモリーのみで動作し、同期レプリケーションによって永続性を実現します（ただし、メッセージごとに非同期レプリケーションを指定することもできます）。必要であれば、Redisと同様にAOFを有効にできます。大量再起動などが起きそうな環境では便利でしょう。システムをアップグレードするときは、通常はディスク永続化を使っていなくても、アップグレードのためだけにAOFをディスクへ書き込み、再起動後に状態を失わないようにすることもできます。

Q：キューイングはサーバー群全体で部分的または完全に一貫していますか。それとも、最大スループットのために分割されていますか？

スループットのために分割されています。ただし、メッセージの順序はベストエフォートで保たれます。各メッセージには不変の「ctime」があります。これはウォールクロックによるミリ秒単位のタイムスタンプに、同じミリ秒内で生成されたメッセージ用の連番を加えたものです。ノードはこのctimeを使って、配信するメッセージをソートします。

Q：負荷が高まった場合、メッセージを完全に破棄できますか？（いわゆるベストエフォートですか？）

いいえ。ただし、メモリーに空きがなければ新しいメッセージを拒否することがあります。メモリー使用量が75％に達すると、メッセージを受信したノードは、コピーを保持せず、外部のノードだけにメッセージをレプリケートしようとします。ただし、他のノードもメモリー不足の状態にある場合は、うまくいかない可能性があります。

Q：コンシューマーやプロデューサーはキューの中身を確認できますか。それとも完全に不透明ですか？

キューを「PEEK」するコマンドがあります。

Q：キューイングは順不同ですか。FIFOですか。それとも優先順位付きですか？

先ほど説明したとおり、ベストエフォートのFIFOに近い方式です。

Q：ブローカーは存在しますか。それともブローカーはありませんか？

マスターの集合としてブローカーが存在します。クライアントは任意のノードと通信できます。

Q：ブローカーは独立した名前付きキュー（トピック、ルートなど）を管理しますか。それとも、プロデューサーとコンシューマーが接続を調整する必要がありますか？

名前付きキューです。プロデューサーとコンシューマーが調整する必要はありません。ノードはフェデレーションを使ってクラスタ内の経路を検出し、コンシューマーが必要とするメッセージを必要に応じて渡します。ただし、より多くのコンシューマーがいる場所へ移動したい場合に備えて、クライアントにはヒントが提供されます。

Q：メッセージの投稿はトランザクション的ですか？

はい。メッセージ追加コマンドが返った時点で、クラスタ内に希望した数のコピーが存在することをシステムが保証します。

Q：メッセージの受信はトランザクション的ですか？

そうではないと思います。確認応答されなければ、Disqueは同じメッセージを再び配信しようとするからです。

Q：コンシューマーは受信時にブロックしますか。それとも新しいメッセージがあるか確認できますか？

どちらの動作もサポートしています。デフォルトではブロックします。

Q：プロデューサーは送信時にブロックしますか。それともキューが満杯か確認できますか？

プロデューサーは、新しいメッセージを追加するとき、メッセージをプッシュするローカルノード上のキュー長が指定した値をすでに超えていれば、エラーを返すよう要求できます。

さらにプロデューサーは、できるだけ早く処理を終えてクラスタに任せたい場合、メッセージを非同期でレプリケートするよう要求できます。クラスタはベストエフォートでメッセージをレプリケートします。

キュー内のメッセージが多すぎるときにコンシューマーをブロックし、メッセージ数が減ったらすぐにブロックを解除する方法はありません。

Q：遅延ジョブはサポートされていますか？

はい。秒単位で指定でき、数年先まで設定できます。ただし、その分メモリーを消費します。

Q：コンシューマーとプロデューサーは別々のノードに接続できますか？

はい。

この投稿で、Disqueが少しは実体のあるものになったことを願っています。もちろん、コードを見なければ判断は難しいでしょう。しかし、最大の機能が公開されたのなら、少なくとも文句を言うことはできます。上で述べたもののうち、すでに実装され、正常に動作しているのはどの程度でしょうか。AOFによるディスク永続化と、APIで改良したい小さな点をいくつか除けば、すべてです。したがって、最初のリリースはそれほど遠くないはずです。ただ、これほどまれにしか作業していないので、非常に速く進めるのは難しいのです。