RedisでPubSubをする場合に https://github.com/go-redis/redis を使ってみることにした。このライブラリのクライアントを、SubscribeとPublishで共用できるのかどうか、コードを読んで確認してみた。

結論だけ言えば、共用できる。

Redisの制約

Redisのドキュメントによれば、Subscribeしているコネクションが発行してよいのは以下のコマンドに限られる。

SUBSCRIBE
PSUBSCRIBE
UNSUBSCRIBE
PUNSUBSCRIBE
PING
QUIT

See: http://redis.io/topics/pubsub

A client subscribed to one or more channels should not issue commands, although it can subscribe and unsubscribe to and from other channels. The replies to subscription and unsubscription operations are sent in the form of messages, so that the client can just read a coherent stream of messages where the first element indicates the type of message. The commands that are allowed in the context of a subscribed client are SUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, PUNSUBSCRIBE, PING and QUIT.

Subscribeでメッセージを待ち受けている時に、他のコマンドのレスポンスが受信されるとマズイことになるので、これは当然の仕様である。

コードを読む

概要

このライブラリにはClientというstructがあり、それを経由して各種のコマンドを送信する。Clientは内部にコネクションプールがあり、そのプール内のコネクションが実際のネットワークの送受信を行っている。

Subscribe

Client#SubscribeでSubscribeを開始する。SubscribeはPubSubオブジェクトを生成し、Subscribeでのメッセージ受信はこれを経由して行うことになる。

このとき、PubSubオブジェクトはClientのコネクションプールを共用するようにしているが、stickyConnPoolというstructでラップしている。

return &PubSub{
	baseClient: &baseClient{
		opt:      c.opt,
		connPool: newStickyConnPool(c.connPool, false),
	},
}

Subscribeを開始したコネクションがプールに戻されないのであれば、Clientで他のコマンドを発行できる。

コネクションのGet

Subscribeのコマンドを送信する場合は、コネクションプールからコネクションをGetして行うことになる。stickyConnPoolのGetは以下のようになっている。

if p.cn != nil {
	cn = p.cn
	return
}

cn, isNew, err = p.pool.Get()
if err != nil {
	return
}
p.cn = cn
return

すでに利用開始しているコネクションp.cnがある場合は、それを返すようになっている。初回の呼び出しでは大もとのコネクションプールからGetし、それをp.cnに入れて覚えておくようにしている。

Subscribe、メッセージの受信、PingなどのPubSubオブジェクト経由のコマンドは、上記のように、Subscribe開始時のコネクションを使い回すようになっている。

Put

メッセージの受信待ちに以下のようなコードがある。

cn, _, err := c.conn()
cmd := NewSliceCmd()
err = cmd.readReply(cn)
c.putConn(cn, err)

c.conn()でSubscribe中のコネクションが取得され、コネクションからデータが受信されるのを待っているというコードだが、受信後にc.putConn()でコネクションを戻すかのようにも見える。

putConnの内部ではコネクションプールのPutが呼び出される。stickyConnPoolのPutは以下のようになっている。

if p.cn != cn {
	panic("p.cn != cn")
}
return nil

というわけで、利用中のコネクションをプールに戻すことはせず、何もしていない(ライブラリのバグを検出するためのアサーションのためのチェックがあるだけ)。

いつプールに戻すのか？

戻さない。Subscribeを開始したコネクションは、それが終了してもコネクションプールには戻らないようになっている。

メッセージ受信中にネットワークのエラーが発生した場合、ライブラリ内部では新しい接続を行ってSubscribeを再開する。またPubSub#Closeでもコネクションを破棄して、プールには戻さないようになっていた。

まとめ

Subscribeを開始したコネクションはプールに戻らない。Clientはプールに他のコネクションを持っているので、同じClientオブジェクトを使って、Publishやその他のコマンドを発行しても問題ない。