Pub/Sub Internal

データ構造 Data Structures

Pub/Sub システムは Redis サーバーとクライアントにチャネルとパターンを登録します。
まず簡単に説明すると、チャネルは dict(Hash table) に保存され、パターンはリンクトリスト(linked list)に保存されます。

まずチャネルを保存する dict(Hash table) を見てみましょう。

チャネルを保存するサーバー Dict(Hash table) データ構造

サーバー構造体(redisServer struct)には pubsub_channels フィールドと pubsub_patterns フィールドがあります。   Pubsub_channels フィールドはチャネルを保存する dict 構造体を指します。   Dict 構造体から dictEntry までは Redis で使うハッシュテーブルです。
DictEntry の key フィールドが channel を指します。   ひとつの channel を複数クライアントが subscribe できるため dictEntry の value フィールドはリスト(linked list)を指します。   リストは複数の listNode を持ち、各ノードはクライアントを指します。
PUBLISH channel message コマンドはまず Hash table から channel を探し、 リストに保存されているクライアントへひとつずつメッセージを送ります。   その次にパターンを登録したクライアントへメッセージを送ります。
パターンを保存するリンクトリスト構造は次のとおりです。

パターンを保存するサーバー リンクトリスト データ構造

サーバー構造体の pubsub_patterns フィールドはパターンを保存するリストを指します。   リストの各ノードは pubsubPattern 構造体を指し、 この構造体はクライアントとパターンを持ちます。
PUBLISH channel message コマンドはまず channel 名でクライアントにメッセージを送った後、 channel に該当するパターンを探してそのクライアントにメッセージを送ります。

チャネルを保存するクライアント Dict(Hash table) データ構造

クライアント構造体(redisClient struct)にもサーバー構造体と同様に pubsub_channels フィールドと pubsub_patterns フィールドがあります。   クライアント構造体の pubsub_channels フィールドもチャネルを保存する dict 構造体を指します。
Dict 構造体から dictEntry まではサーバー構造体のときと同じで dictEntry の key フィールドが channel を指すところまでサーバーのときと同じです。   ただしクライアントは value を持っていません。
Subscribe コマンドが実行されるとクライアント構造体にチャネルを保存し、 サーバー構造体にもチャネルを保存します。   Publish コマンドが実行されると該当チャネルを探してクライアントにメッセージを送るのは サーバー構造体を使います。
クライアント構造体はクライアントが該当チャネルを登録し、 クライアントを pubsub モードへ切り替える役割をします。   クライアントが normal モードなら client buffer は無制限ですが pubsub モードなら hard limit は 32mb、soft limit は 8mb に制限します。   Unsubscribe または punsubscribe コマンドが実行され、クライアント構造体にチャネルや パターンがひとつもなければクライアントを pubsub モードから normal モードへ切り替えます。

パターンを保存するクライアント リンクトリスト データ構造

クライアント構造体の pubsub_patterns フィールドはパターンを保存するリストを指します。   リストの各ノードはサーバーと異なり pattern 構造体を直接指します。
Psubscribe コマンドが実行されるとクライアント構造体であるリストにパターンを保存し、 サーバー構造体にもパターンを保存し、 クライアントを pubsub モードへ切り替える役割もします。

データ構造を説明したので、次は functions を説明します。

FUNCTIONS

Redis pubsub は6つのコマンドで構成されています。
メッセージを受ける subscribe と psubscribe があり、 メッセージ受信を止める unsubscribe と punsubscribe があります。   メッセージを送る publish があり、 登録されたチャネル リストと個数を照会しパターン個数を照会する pubsub コマンドがあります。
ひとつずつ内部構造を見てみましょう。

SUBSCRIBE

チャネルを登録してメッセージを受けるコマンドです。
SUBSCRIBE コマンドが実行されるとサーバー構造体(redisServer struct)と クライアント構造体(redisClient struct)にチャネルを登録します。
説明する Function の引数(argument)名は理解を助けるため元ソースと異なり 構造体名に変えたところがあります。   上の図に出る function は太い文字で表示しました。

1. まず dictAdd(client->pubsub_channels,channel,NULL) を実行して redisClient.pubsub_channels の dict 構造体にチャネル名を登録します。   3番目の引数が値で、NULL を渡すことで値(value)は NULL が入ります。   図 1-2 を見ると理解しやすいです。
2. ここからはサーバー構造体に登録する過程です。
   dictFind(server.pubsub_channels,channel) を実行してチャネルがなければ list = listCreate() を実行してリストを生成します。   Channel がすでに登録されていれば、 list = dictGetVal(dictEntry) を実行してリストを取得し、4番へ進みます。
3. dictAdd(server.pubsub_channels,channel,list) を実行してチャネルとリストを dict に登録します。
4. listAddNodeTail(list,client) を実行してリストにクライアントを登録します。

図 1-1 と合わせて見ると役立ちます。

PSUBSCRIBE

パターンを登録してメッセージを受けるコマンドです。
PSUBSCRIBE コマンドが実行されるとサーバーとクライアントのリスト(Linked List)にパターンを登録します。

1. listSearchKey(client->pubsub_patterns,pattern) でパターンがすでにあるか確認します。
   なければ listAddNodeTail(client->pubsub_patterns,pattern) を実行してクライアント リストに パターンを登録します。   図 1-4 を見ると役立ちます。
2. listAddNodeTail(server.pubsub_patterns,pat) でサーバーにパターンを登録しますが、 クライアントにはパターンだけ登録するのに対し、サーバーには pubsubPattern 構造体に client と pattern を入れて一緒に登録します。   パターンを登録したクライアントへメッセージを送るためです。   図 1-2 を見ると役立ちます。

UNSUBSCRIBE

登録したチャネルを削除し、これ以上メッセージを受けないようにします。
引数でチャネルを入力しなければ pubsubUnsubscribeAllChannels() を呼び出してクライアントに登録された すべてのチャネルを削除します。   pubsubUnsubscribeAllChannels() で while loop を回しながら pubsubUnsubscribeChannel() を呼び出して dictNect() を使ってチャネルをひとつずつ削除します。
チャネルを入力すれば入力したチャネル個数だけ while loop を回しながら pubsubUnsubscribeChannel() を呼び出して チャネルをひとつずつ削除します。

1. dictDelete(client->pubsub_channels,channel) を実行してクライアントからチャネルを削除します。
2. dictEntry = dictFind(server.pubsub_channels,channel) でサーバーから dictEntry を求めます。
3. list = dictGetVal(dictEntry) で value に保存された list を取得します。
4. listNode = listSearchKey(list,client) を実行してリスト ノードを取得します。
5. listDelNode(list,listNode) を実行してリストからノードを削除します。
6. リストにノードがひとつもなければ dictDelete(server.pubsub_channels,channel) を実行して リストを削除します。

PUNSUBSCRIBE

登録したパターンを削除し、これ以上メッセージを受けないようにします。
引数でパターンを入力しなければ pubsubUnsubscribeAllPatterns() を呼び出してクライアントに登録された すべてのパターンを削除します。   pubsubUnsubscribeAllPatterns() で while loop を回しながら pubsubUnsubscribePattern() を呼び出して listDelNode() を使ってチャネルをひとつずつ削除します。
チャネルを入力すれば入力したチャネル個数だけ while loop を回しながら pubsubUnsubscribePattern() を呼び出して チャネルをひとつずつ削除します。

1. まず listSearchKey(client->pubsub_patterns,pattern) を実行してクライアントからパターンを探します。
2. listDelNode(client->pubsub_patterns,listNode) を実行してパターンを削除します。
3. 次に listSearchKey(server->pubsub_patterns,pubsubPattern) を実行してサーバーからパターンを探します。
4. listDelNode(server->pubsub_patterns,listNode) を実行してパターンを削除します。

PUBLISH

上で見た4つのコマンドのうち前の2つはメッセージを受けるもので、後ろの2つはメッセージをこれ以上受けない コマンドです。
PUBLISH はメッセージを送るコマンドです。
PUBLISH は大きく2つの部分で構成されています。   ひとつはサーバー内のクライアントへメッセージを送ること、 もうひとつはサーバーがクラスタ モードならほかのサーバーへ伝えて、各サーバーに 接続しているクライアントへメッセージを送ることです。
簡単に言うと Pub/Sub はクラスタ モードをサポートします。

サーバー内での動作です。

1. まずサーバーで該当チャネルを探します。 dictFind(server.pubsub_channels,channel)
2. listNext(listNode) でノードがなくなるまで While Loop を回しながら 登録されたクライアントに addReplyBulk(client,message) でメッセージを送ります。
3. 次に listLength(server.pubsub_patterns) で確認してサーバーに登録されたパターンがあれば
4. listNext(listNode) でノードがなくなるまで While Loop を回しながら stringmatchlen() function で比較して合えばクライアントに addReplyBulk(pubsubPattern->client,message) でメッセージを送ります。

クラスタのノードへ伝える過程です。
clusterPropagatePublish() からは cluster.c にあります。
簡単に説明すると、サーバー内にクラスタのサーバー情報は dict に保存されていて、 dictNext() でサーバー情報を得てチャネルとメッセージを送ります。   するとそのサーバーはチャネルとメッセージを受け、上と同じ過程を経て クライアントへメッセージを送ります。

PUBSUB

• Subcommand が channels なら server.pubsub_channels で dictNext() で 移動しながらチャネルを表示します。パターンが入力されていれば stringmatchlen() で パターンと比較し合うチャネルだけ表示します。
• Subcommand が numsub なら該当チャネルに登録されたクライアント個数を listLength(list) で求めて表示します。
• Subcommand が numpat ならリストに登録されたパターン個数を listLength(list) で求めて表示します。