Node.js+Redis实现定时器任务
标签（空格分隔）： Node.js
[TOC]
简介
在 Redis 的 2.8.0 版本之后，其推出了一个新的特性——键空间消息（Redis Keyspace Notifications），它配合 2.0.0 版本之后的 subscribe 就能完成这个定时任务的操作了，不过定时的单位是秒。
Redis的发布与订阅(Publish / Subscribe)
Redis 在 2.0.0 之后推出了 Pub / Sub 的指令，大致就是说发布端给 Redis 的特定频道发送消息，订阅端从Redis的特定频道取值——形成了一个简易的消息队列
比如
发布端可以往 foo 频道推一个消息bar，那么就可以直接：
PUBLISH foo bar
订阅端在客户端订阅 foo 频道就能接受到这个消息了。
举个例子，如果在 Node.js 里面使用 ioredis 这个包那么看起来就会像这样：
var Redis = require("ioredis");
var sub = new Redis(/** 连接信息 */);
sub.once("connect", function() {
// 假设我们需要选择 redis 的 db，因为实际上我们不会去污染默认的 db 0
sub.select(DB_NUMBER, function(err) {
if(err) process.exit(4);
sub.subscribe("foo", function() {
//... 订阅频道成功
});
});
});
// 监听从 `foo` 来的消息
sub.on("message", function(channel, msg) {
console.log(channel, msg);
});
Redis的键空间提醒(Redis Keyspace Notifications)
在 Redis 里面有一些事件，比如键到期、键被删除等。然后我们可以通过一些让 Redis 一旦触发这些事件的时候就往特定的 Channel 推一条消息。
消息类型及频道名的命名规则
频道名的命名规则
频道名的命名规则是__keyspace@+dbnumber+__:+key_name.
例如：在db0数据库中键名为foo的频道名为__keyspace@0__:foo
消息类型
在redis.conf文件中找到notify-keyspace-events.它的值可以是Ex、Klg等，这些字母的具体含义如下
K，表示 keyspace 事件，有这个字母表示会往 __keyspace@<db>__ 频道推消息。
E，表示 keyevent 事件，有这个字母表示会往 __keyevent@<db>__ 频道推消息。
g，表示一些通用指令事件支持，如 DEL、EXPIRE、RENAME 等等。
$，表示字符串（String）相关指令的事件支持。
l，表示列表（List）相关指令事件支持。
s，表示集合（Set）相关指令事件支持。
h，哈希（Hash）相关指令事件支持。
z，有序集（Sorted Set）相关指令事件支持。
x，过期事件，与 g 中的 EXPIRE 不同的是，g 的 EXPIRE 是指执行 EXPIRE key ttl 这条指令的时候顺便触发的事件，而这里是指那个 key 刚好过期的这个时间点触发的事件。
e，驱逐事件，一个 key 由于内存上限而被驱逐的时候会触发的事件。
A，g$lshzxe 的别名。也就是说 AKE 的意思就代表了所有的事件。
配置redis
修改redis.conf配置文件，打开以下配置项
notify-keyspace-events Ex // 开启键到期提醒配置
示例代码
const redis = require('redis');
const uuid = require('node-uuid');
const redis_config = {
host: "172.16.203.138",
port: 6379,
password: 'user',
db: 1
}
// 数据库连接
const redis_server = redis.createClient(redis_config); // 发布者
redis_server.on('connect', () => {
console.log('redis_server connected')
})
redis_server.on('error', () => {
console.log('error occured from redis_server')
})
const redis_cli = redis.createClient(redis_config);
redis_cli.on('connect', () => {
console.log("redis_cli connected")
})
redis_cli.on('error', () => {
console.log("error occured from redis_cli")
})
// 编写一个接收到消息以后要执行的方法
const test = (message) => {
console.log("执行了test方法" + message)
}
const funcs = {
test
}
// 添加消息队列
const SampleTaskMaker = (message, func_name, timeout) => {
let message_obj = {
message: message,
func_name: func_name,
timeout: timeout
}
let key = JSON.stringify(message_obj);
let content = "";
redis_server.multi()
.set(key, content)
.expire(key, timeout)
.exec((error) => {
if (error) {
console.log("任务添加失败");
return;
} else {
console.log("任务添加成功")
}
});
console.log({
key
})
return key;
}
// 创建任务
SampleTaskMaker("这是测试的定时任务信息", "test", 2);
// 创建订阅通知号
const SampleOnExpired = (channel, key) => {
console.log("接收到了推送")
let message_obj = JSON.parse(key);
let message = message_obj.message;
let func_name = message_obj.func_name;
funcs[func_name](message);
}
// 设置订阅信息。这里的@1是和所用数据库有关的，因为用的是db1.所以用@1
let subscribeKey = "__keyevent@1__:expired";
// 订阅服务器创建订阅客户端
redis_cli.once('connect', () => {
redis_cli.subscribe(subscribeKey, () => {
console.log("订阅成功");
})
})
// 监听从‘__keyevent@2__:expired’发来的消息
redis_cli.on("message", SampleOnExpired);
网友评论