Redis中的订阅发布和事务(一)

订阅发布

PUBSUB NUMSUB

PUBSUB NUMSUB [channel-1 channel-2… channel-n]子命令接受任意多个频道作为输入参数，并返回这些频道的订阅者数量。
这个子命令是通过pubsub_channels字典中找到频道对应的订阅者链表，然后返回订阅者链表的长度来实现的(订阅者链表的长度
就是频道订阅者的数量)，这个过程可以用以下伪代码来描述:

def pubsub_numsub(*all_input_channels):
# 遍历输入的所有频道
for channel in all_input_channels:
# 如果pubsub_channels字典中没有channel这个键
# 那么说明channel频道没有任何订阅者
if channel not in server.pubsub_channels:
# 返回频道名
reply_channel_name(channel)
# 订阅者数量为0
reply_subscribe_count(0)

# 如果pubsub_channels字典中存在channel键
# 那么说明channel频道至少有一个订阅者
else:
# 返回频道名
reply_channel_name(channel)
# 订阅者链表的长度就是订阅者数量
reply_subscribe_count(len(server.pubsub_channels(channel)))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

例子

• 举个例子。对于图中所示的pubsub_channels字典来说，对字典中的四个频道执行PUBSUB NUMSUB命令将获得以下回复

redis>PUBSUB NUMSUB news.it news.sport news.business news.movie
1)."news.it"
2)."3"
3)."news.sport"
4)."2"
5)."news.business"
6)."2"
7)."news.movie"
8)."1"
1
2
3
4
5
6
7
8
9

PUBSUB NUMPAT

PUBSUB NUMPAT子命令用于返回服务器当前被订阅模式的数量。这个子命令是通过返回pubsub_patterns链表的长度来实现的，因为这个链表的长度就是服务器被订阅模式的数量，这个过程可以用以下伪代码来描述:

def pubsub_numpat():
# pubsub_patterns链表的长度即是被订阅模式的数量
reply_pattern_count(len(server.pubsub_patterns))
1
2
3

例子

• 举个例子。对于图中所示的pubsub_patterns链表来说，执行PUBSUB NUMPAT命令将返回3:

redis>PUBSUB NUMPAT
(integer) 3
1
2

事务

概述

Redis通过MULTI、EXEC、WATCH等命令来实现事务(transaction)功能。事务提供了一种将多个命令请求打包,然后一次性、按顺序地执行多个命令的机制，并且在事务执行期间，服务器不会中断事务而改去执行其他客户端的命令请求，它会将事务中的所有命令都执行完毕，然后采取执行其他客户端的命令请求。

例子

• 举个例子。事务首先以一个MULTI命令为开始，接着将多个命令放入事务当中，最后由EXEC命令将这个事务提交(commit)给服务器执行:

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> SET "name" "Practical Common Lisp"
QUEUED
127.0.0.1:6379> GET "name"
QUEUED
127.0.0.1:6379> SET "author" "Peter Seibel"
QUEUED
127.0.0.1:6379> GET "author"
QUEUED
127.0.0.1:6379> EXEC
1) OK
2) "Practical Common Lisp"
3) OK
4) "Peter Seibel"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

事务的实现

一个事务从开始到结束通常会经历以下三个阶段:

• 1.事务开始
• 2.命令入队
• 3.事务执行

事务开始

MULTI命令的执行标志着事务的开始:

redis> MULTI
OK
1
2

MULTI命令可以将执行该命令的客户端从非事务状态切换至事务状态，这一切换是通过在客户端状态的flags属性中打开REDIS_MULTI标识来完成的，MULTI命令的实现可以用以下伪代码来表示:

def MULTI():
# 打开事务表示
client.flags |= REDIS_MULTI

# 返回OK回复
replyOK()
1
2
3
4
5
6

命令入队

当一个客户端处于非事务状态时，这个客户端发送的命令会立即被服务器执行:

127.0.0.1:6379> SET "name" "Practical Common Lisp"
OK
127.0.0.1:6379> GET "name"
"Practical Common Lisp"
127.0.0.1:6379> SET "author" "Peter Seibel"
OK
127.0.0.1:6379> GET "author"
"Peter Seibel
1
2
3
4
5
6
7
8

与此不同的是，当一个客户端切换到事务状态之后，服务器会根据这个客户端法拉的不同命令执行不同的操作:

• 1.如果客户端发送的命令为EXEC、DISCARD、WATCH、MULTI四个命令的其中一个，那么服务器立即执行这个命令
• 2.与此相反，如果客户端发送的命令是EXEC、DISCARD、WATCH、MULTI四个命令以外的其他命令，那么服务器并不立即执行这个命令，而是将这个命令放入一个事务队列里面，然后向客户端返回QUEUED回复。

服务器判断命令是该入队还是该立即执行的过程可以用流程图来描述

事务队列

每个Redis客户端都由自己的事务状态，这个事务状态保存在客户端状态的mstate属性里面:

typedef struct redisClient {
// ...

// 事务状态
multiState mstate; // MULTI/EXEC state

// ...
}redisClient;
1
2
3
4
5
6
7
8

事务状态包含一个事务队列，以及一个已入队命令的计数器(也可以说是事务队列的长度):

typedef struct multiState {
// 事务队列, FIFO顺序
multiCmd *commands;

// 已入队命令计数
int count;
} multiState;
1
2
3
4
5
6
7

事务队列是一个multiCmd类型的数组，数组中的每个multiCmd结构都保存了一个已入队命令的相关信息，包括指向命令实现函数的指针、命令的参数，以及参数的数量:

typedef struct multiCmd {
// 参数
robj **argv;

// 参数数量
int argc;

// 命令指针
struct redisCommand *cmd;
}multiCmd;
1
2
3
4
5
6
7
8
9
10

事务队列以先进先出(FIFO)的方式保存入队的命令，较先入队的命令会被放到数组的前面，而较后入队的命令则会被放到数组的后面

例子

• 举个例子。如果客户端执行以下命令:

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> SET "name" "Practical Common Lisp"
QUEUED
127.0.0.1:6379> GET "name"
QUEUED
127.0.0.1:6379> SET "author" "Peter Seibel"
QUEUED
127.0.0.1:6379> GET "author"
QUEUED
1
2
3
4
5
6
7
8
9
10

那么服务器将为客户端创建如图所示的事务状态:
1.最先入队的SET命令被放在了事务队列的索引0位置上
2.第二入队的GET命令被放在了事务队列的索引1位置上
3.第三入队的另一个SET命令被放在了事务队列的索引2位置上
4.最后入队的另一个GET命令被放在了事务队列的索引3位置上