Pub/Sub Introduction

メッセージを送る、受信

一般的なデータベースとは違うようにredisはメッセージを与えて、受け機能を提供します。   Publishコマンドで送って、Subscribeコマンドで受けます。
通路は、チャネル(channel)を利用します。   チャンネルは"SET KEY VALUE"で使用する'KEY'と同じものと見てもいいです。
方法は、クライアント1でsubscribe channel_nameを実行して、 クライアント2でpublish channel_name"Message"を実行すると、 クライアント1に"Message"が出ます。
redisのPub/Subシステムはメッセージを保管(queuing)しません。   Publishする時点にすでに実行したsubscribeコマンドで待機している クライアントにだけ伝達されます。

もうコマンドを一つずつチェックします。

SUBSCRIBE channel [channel ...]

指定したチャンネルとして送られたというメッセージを受け取ります。   チャンネルを複数指定できます。

6379> subscribe ch01 ch02
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "ch01"
3) (integer) 1
1) "subscribe"
2) "ch02"
3) (integer) 2

他のクライアントからPUBLISH ch01"Hello!"コマンドでメッセージを送ると、 次のように受けます。

1) "message"
2) "ch01"
3) "Hello!"

PSUBSCRIBE pattern [pattern ...]

チャンネルをパターンとして登録します。 パターンを複数登録することができます。
パターンは以下のようなglob-styleを支援します。

• '?'は一字を代置します。  h?lloはhello、hallo、hxlloのようなことを意味します。
• '*'は空白または複数の文字を代置します。  h*lloはhllo、heeeelloのようなことを意味します。
• h[ae]lloは'a'または'e'だけ可能です。  それでhello、halloは可能で、hilloはいけません。

パターン登録コマンドの例

6379> psubscribe ch*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "ch*"
3) (integer) 1

他のクライアントでPUBLISH ch01"Hello!AlphaGo"コマンドでメッセージを送ると、 次のように受けます。

1) "pmessage"
2) "ch*"
3) "ch01"
4) "Hello! AlphaGo"

UNSUBSCRIBE [channel [channel ...]]

SUBSCRIBEに登録したチャンネルを削除して、これ以上メッセージを受けることはありません。
チャンネル名を入力しなければ、該当クライアントに登録されたすべてのチャンネルを削除します。
入力すると、入力したチャンネル名のみ削除します。
redis-cliではCtrl-Cで終了します。

6379> unsubscribe ch*
1) "unsubscribe"
2) "ch*"
3) (integer) 1

PUNSUBSCRIBE [pattern [pattern ...]]

PSUBSCRIBEに登録したパターンを削除して、これ以上メッセージを受けることはありません。
パターンの名を入力しなければ、該当クライアントに登録されたすべてのパターンを削除します。
入力すると、入力したパターンの名のみ削除します。
redis-cliではCtrl-Cで終了します。

6379> punsubscribe ch*
1) "punsubscribe"
2) "ch*"
3) (integer) 1

PUBLISH channel message

メッセージを指定したチャンネルに送ります.
メッセージを受けるクライアント数をリターンします。

6379> publish ch01 "Hello! AlphaGo"
(integer) 1

PUBSUB subcommand [argument [argument ...]]

PUBSUBコマンドは、サーバーに登録されたチャンネルやパターンを照会します。
三つのsubcommandがあります。   channels、numsub、numpatです。
channels、numsubは、チャネル関連コマンドであり、 numpatはパターン関連コマンドです。

PUBSUB channels [pattern]

Patternを入力しなければ、当該サーバに登録されたすべてのチャンネル名を見せてくれます。
ここでpatternはpsubscribeで登録したpatternがなく、チャンネル名をglob-styleで探すのです。
登録したpatternを示すものはありません。
クライアントで同じ名前のチャンネルを何度も登録してもこのコマンドでは一度だけ見せてくれます。

6379> pubsub channels
1) "ch01"
2) "ch02"

PUBSUB numsub [channel-1 ... channel-n]

当該チャンネルに登録されたクライアント数を示します。
PUBSUB channelsで確認したチャンネルそれぞれにいくつかのクライアントが連結されているのかを示してくれます。

6379> pubsub numsub ch01 ch02
1) "ch01"
2) (integer) 3
3) "ch02"
4) (integer) 1

PUBSUB numpat

サーバーに登録されたpatternの個数を見せてくれます。

6379> pubsub numpat
(integer) 3

클러스터에서 Publish하면 클론을 포함한 모든 노드에게 보낸다. 따라서 마스터에서 publish한 메시지를 클론(슬레이브)에서 subscribe할 수 있다.  마스터 뿐만 아니라 클론에서도 publish할 수 있다. 클론에서 publish한 메시지를 다른 클론에서 subscribe할 수 있다.  Pub/Sub를 중요하게 사용한다면 클러스터를 사용하세요.

データ構造 Data Structures

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

まず、チャンネルを保存する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コマンドは先にchannelを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は六つのコマンドで構成されています。
メッセージを受ける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構造体にチャンネル名を登録します。   三番目の引数が値ですが、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は大きく二つの部分で構成されています。   一つはサーバ内にクライアントにメッセージを送ることで、 他の一つはサーバがクラスターモードなら他のサーバーに伝え、メッセージを各サーバに 接続しているクライアントにメッセージを送るのです。
簡単に説明すれば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)に探して示します。