ElastiCache で Redis キースペース通知を実装するにはどうすればよいですか?

簡単な説明

Redis キースペース通知を使用すると、PubSub チャネルをサブスクライブできます。Redis コマンドまたはデータの変更が発生すると、クライアントはチャネルを介してパブリッシュされたイベントを受信します。これらの通知は、特定の単一または複数のキーに保存されている値に発生した変更に対して、アプリケーションが応答する必要がある場合に便利です。また、Redis キースペース通知を使用して、データセットの変更を追跡することもできます。キースペース通知機能は、バージョン 2.8.6 以降で利用できます。

Redis キースペース通知をトリガーするコマンドが実行されると、モニタリング中のクライアントが応答する次の 2 つのイベントが発生します。

• キースペース通知
• キーイベント通知

キースペースチャネルはイベントの名前を、キーイベントチャネルはキーの名前を、メッセージとして受信します。

重要: 次の解決策は Redis (クラスターモード無効) クラスターにのみ適用されます。解像度は Redis (クラスターモード有効) クラスターには適用されません。両方のクラスタータイプのキースペース通知については、Redis サイトの「Redis キースペース通知」を参照してください。

解決方法

Redis キースペース通知を有効にする

ElastiCache では、Redis キースペース通知はデフォルトで無効になっています。カスタムキャッシュパラメータグループで notify-keyspace-events パラメータを使用して、キースペース通知を有効にすることができます。notify-keyspace-events の値には、使用するチャネル (keyspace イベントまたは key-event イベント) と、これらのチャネルに投稿する情報の決定に複数のパラメータを使用します。詳細については、「Redis 2.8.6 で追加されたパラメータ」を参照してください。

1.    ElastiCache コンソールを開きます。

2.    利用可能なパラメータグループすべての一覧を表示するには、ナビゲーションペインで [Parameter Groups] (パラメータグループ) を選択します。

3.    変更するパラメータグループを選択します。
注: デフォルトのパラメータグループは変更できません。表示されているパラメータグループがすべてデフォルトの場合は、[Create Parameter Group] (パラメータグループの作成) を選択して新しいパラメータグループを作成し、[Family] (ファミリー)、[Create] (作成) の順に選択します。

4.    [パラメータの編集] を選択します。

5.    notify-keyspace-events までスクロールし、[Value] (値) フィールドに AKE と入力します。許可される値の詳細については、 Redis 2.8.6 追加のパラメータを参照してください。

6.    [変更の保存] を選択します。変更は、クラスターを再起動しなくても直ちに有効になります。
注: 変更した新しいパラメータグループを Redis クラスターに割り当ててください。

redis-cli を使用してイベントを生成する

redis-cli を使用している場合、次のコマンドを実行して、指定したコマンドがどのようにイベントを生成するかを確認できます。

#./redis-cli -c -h testdisabled.xxxxx.ng.0001.use1.cache.amazonaws.com -p 6379 --csv psubscribe '__key*__:*'

任意のコマンドセットを実行して、キースペースイベントを作成します。次のコマンドは、Redisキーの値と有効期限を 3 秒に設定し、期限切れイベントのサブスクライバーを設定します。

./redis-cli -c -h testdisabled.xxxxx.ng.0001.use1.cache.amazonaws.com -p 6379
testdisabled.xm4oz6.ng.0001.use1.cache.amazonaws.com:6379>
testdisabled.xm4oz6.ng.0001.use1.cache.amazonaws.com:6379> SET key1 value1 ex 5
testdisabled.xm4oz6.ng.0001.use1.cache.amazonaws.com:6379>

別のセッションで、イベントが生成されていることを確認します。イベントからの出力例を次に示します。

# ./redis-cli -h testdisabled.xxxxx.ng.0001.use1.cache.amazonaws.com -p 6379 --csv psubscribe '*'
Reading messages... (press Ctrl-C to quit)
"psubscribe","*",1
"pmessage","*","__keyspace@0__:key1","set"
"pmessage","*","__keyevent@0__:set","key1"
"pmessage","*","__keyspace@0__:key1","expire"
"pmessage","*","__keyevent@0__:expire","key1"
"pmessage","*","__keyspace@0__:key1","expired"
"pmessage","*","__keyevent@0__:expired","key1"

この出力例では、チャネルに正常にサブスクライブできたことが最初のイベントによって示されています。数字「1」は、現在サブスクライブしているチャネルの数を表しています。

2 つ目のイベントは、keyspace 通知です。キースペースチャネル __keyspace@0__:key1 には、イベントの名前である「set」がメッセージとして含まれています。

3 つ目のイベントは、key-event 通知です。キーイベントチャネル __keyevent@0__:set には、キーの名前である「key1」がメッセージとして含まれています。

5 秒後に、出力内の期限切れメッセージを見ることができます。メッセージには微妙な違いがあります。一方のメッセージは __keyevent@0__:expire で、もう一方のメッセージは __keyevent@0__:expired です。

PubSub インスタンスから読み取られる各メッセージは、次のキーを含む辞書のようなものです。

• Type - 次のいずれか: subscription、unsubscribe、psubscribe、punsubscribe、message、pmessage
• Pattern - 公開されたメッセージのチャネルに一致するパターン (すべてのケースで一致しない pmessage タイプは除外)
• Channel - メッセージが投稿されるチャネル、またはメッセージがサブスクライブされるチャネル
• Data - メッセージデータ

Redis-py を使用してイベントを生成する

1.    通知にサブスクライブします。次の例では、すべてのキースペース通知にサブスクライブし、受信したイベントを出力します。Python2 または Python3 のいずれかに最適なスクリプトを使用してください。

Python 2 の場合:

# cat subscribe.py 
import redis
import datetime
import time
r = redis.Redis(host='testdisabled.xxxxx.ng.0001.use1.cache.amazonaws.com', port=6379, db=0)
#Subscribing for events matching pattern "__key*__:*"
p = r.pubsub()
p.psubscribe('__key*__:*')
print('Starting message loop')
while True:
    message = p.get_message()
    if message:
        print datetime.datetime.now(), message
    else:
        time.sleep(0.01)

Python 3 の場合:

import redis
from datetime import datetime
import time
r = redis.Redis(host = 'testdisabled.xxxxx.ng.0001.use1.cache.amazonaws.com', port = 6379, db = 0)
#Subscribing for events matching pattern "__key*__:*
p = r.pubsub()
p.psubscribe('__key*__:*')
print('Starting message loop')
while True:
    message = p.get_message()
    if message:
        print (datetime.now(), message)
    else:
        time.sleep(0.01)

注意: 値が小さいかスリープがない場合、チャネルがクライアントによって常にポーリングされ、サーバーが過負荷になる場合があります。値が大きいほど、サーバー側のメモリ使用率が増加します。上記の例では、time.sleep を (0.01) に設定しています。

2.    キーを設定するか期限切れにします。

別のセッションでクラスターに接続し、createkey.py コマンドを実行します。key2 キーの値を value2 に、有効期限を 3 秒に設定します。

# cat createkey.py
import redis
  r = redis.Redis(host='testdisabled.xxxxx.ng.0001.use1.cache.amazonaws.com', port=6379, db=0)
  r.set('key2,’value2’)
  r.expire('key2’,3)

別のセッションで subscribe.py コマンドを実行します。以下が出力例です。

# python2 subscribe.py

Starting message loop
2020-08-17 18:16:23.902762 {u'pattern': None, u'type': 'psubscribe', u'channel': '__key*__:*', u'data': 1L}
2020-08-17 18:16:50.375281 {u'pattern': '__key*__:*',
    u'type': 'pmessage', u'channel': '__keyspace@0__:key2', u'data': 'set'}
2020-08-17 18:16:50.375365 {u'pattern': '__key*__:*', u'type': 'pmessage', u'channel': '__keyevent@0__:set', u'data': 'key2'}
2020-08-17 18:16:50.375442 {u'pattern': '__key*__:*',
    u'type': 'pmessage', u'channel': '__keyspace@0__:key2', u'data': 'expire'}
2020-08-17 18:16:50.375499 {u'pattern': '__key*__:*', u'type': 'pmessage', u'channel': '__keyevent@0__:expire', u'data': 'key2'}
2020-08-17 18:16:53.458622 {u'pattern':
    '__key*__:*', u'type': 'pmessage', u'channel': '__keyspace@0__:key2', u'data': 'expired'}
2020-08-17 18:16:53.458856 {u'pattern': '__key*__:*', u'type': 'pmessage', u'channel': '__keyevent@0__:expired', u'data': 'key2'}

Redis キーの有効期限リスナーの準備ができました。

---