① 一个发布者多个订阅者模型
② 多个发布者一个订阅者模型
③ 多个发布者多个订阅者模型
一个发布者多个订阅者模型
# 先订阅到一个频道
127.0.0.1:6379> SUBSCRIBE FM888
1) "subscribe"
2) "FM888"
3) (integer) 1
# 发布者发布消息
127.0.0.1:6379> PUBLISH FM888 test1
(integer) 1
127.0.0.1:6379> PUBLISH FM888 test2
(integer) 2 (代表订阅个数)
# 订阅者们接收消息
127.0.0.1:6379(subscribed mode)> SUBSCRIBE FM888
1) "subscribe"
2) "FM888"
3) (integer) 1
1) "message"
2) "FM888"
3) "test1"
1) "message"
2) "FM888"
3) "test2"
多个发布者个一个订阅者模型
# 订阅者订阅频道
127.0.0.1:6379(subscribed mode)> SUBSCRIBE FM888
# 发布者1
127.0.0.1:6379> PUBLISH FM888 test2
(integer) 1
# 发布者2
127.0.0.1:6379> PUBLISH FM888 test3
(integer) 1
# 订阅者
127.0.0.1:6379(subscribed mode)> SUBSCRIBE FM888
1) "subscribe"
2) "FM888"
3) (integer) 1
1) "message"
2) "FM888"
3) "test2"
1) "message"
2) "FM888"
3) "test3"
多个发布者个多个订阅者模型
# 订阅者们订阅频道
127.0.0.1:6379> PSUBSCRIBE *
# 发布者1
127.0.0.1:6379> PUBLISH FM3 test10
(integer) 1
# 发布者2
127.0.0.1:6379> PUBLISH FM888 test10
(integer) 1
# 发布者3
127.0.0.1:6379> PUBLISH FM888 test1
(integer) 1
# 发布者4
127.0.0.1:6379> PUBLISH FM12 test12
(integer) 1
# 订阅者们收到的消息
127.0.0.1:6379> PSUBSCRIBE *
1) "psubscribe"
2) "*"
3) (integer) 1
1) "pmessage"
2) "*"
3) "FM3"
4) "test10"
1) "pmessage"
2) "*"
3) "FM888"
4) "test10"
1) "pmessage"
2) "*"
3) "FM888"
4) "test1"
1) "pmessage"
2) "*"
3) "FM12"
4) "test12"
Lua是一种轻量小巧的脚本语言,用标准C语言编写并以源代码形式开放。Redis 2.6 版本通过内嵌支持 Lua 环境。也就是说一般的运用,是不需要单独安装Lua。
使用 Lua 脚本最大的好处是 Redis 会将整个脚本作为一个整体执行
,不会被其他请求打断,可以保持原子性
且减少了网络开销。
常用的命令不多,就下面这几个:
eval
script
numkeys
key [key ...]
arg [arg ...]
script 参数
: 是一段 Lua 脚本程序,它可以是一段程序也可以是一个文件。
numkeys参数
: 指定后续参数有几个key,即:key [key …]中key的个数。如没有key,则为0
key [key …]
: 从 EVAL 的第三个参数
开始算起,表示在脚本中所用到的那些 Redis 键(key)。在Lua脚本中通过KEYS[1], KEYS[2]
获取。
arg [arg …]
: 附加参数
,通过ARGV[1],ARGV[2]
获取
// 例1:numkeys=1,keys数组只有1个元素key1,arg数组无元素
127.0.0.1:6379> EVAL "return KEYS[1]" 1 key1
"key1"
// 例2:numkeys=0,keys数组无元素,arg数组元素中有1个元素value1
127.0.0.1:6379> EVAL "return ARGV[1]" 0 value1
"value1"
// 例3:numkeys=2,keys数组有两个元素key1和key2,arg数组元素中有两个元素first和second
127.0.0.1:6379> eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 first second
1) "key1"
2) "key2"
3) "first"
4) "second"
Lua 脚本中执行 Redis 命令
redis.call(command, key [key …] argv [argv…])
command:Redis 中的命令,如 set、get 等。
key:操作 Redis 中的 key 值,相当于我们调用方法时的形参。
param:代表参数,相当于我们调用方法时的实参。
SCRIPT LOAD命令格式:SCRIPT LOAD script
EVALSHA命令格式:EVALSHA sha1 numkeys key [key …] arg [arg …]
SCRIPT LOAD 将脚本 script 添加到Redis服务器的脚本缓存
中,并不立即执行这个脚本,而是会立即对输入的脚本进行求值。并返回
给定脚本的 SHA1
校验和。如果给定的脚本已经在缓存里面了,那么不执行任何操作。
在脚本被加入到缓存之后,在任何客户端通过EVALSHA
命令,可以使用脚本的 SHA1 校验和来调用这个脚本。脚本可以在缓存
中保留无限长的时间,直到执行SCRIPT FLUSH为止。
简单来说,就是执行
SCRIPT LOAD
命令会存到缓存,执行EVALSHA
会校验缓存中是否有,有就执行缓存的。
## SCRIPT LOAD加载脚本,并得到sha1值
127.0.0.1:6379> SCRIPT LOAD "redis.call('SET', KEYS[1], ARGV[1]);redis.call('EXPIRE', KEYS[1], ARGV[2]); return 1;"
"6aeea4b3e96171ef835a78178fceadf1a5dbe345"
## EVALSHA使用sha1值,并拼装和EVAL类似的numkeys和key数组、arg数组,调用脚本。
127.0.0.1:6379> EVALSHA 6aeea4b3e96171ef835a78178fceadf1a5dbe345 1 userAge 10 60
(integer) 1
127.0.0.1:6379> get userAge
"10"
SCRIPT EXISTS sha1 [sha1 …]
判断脚本是否在缓存中。
SCRIPT FLUSH sha
清除Redis服务端所有 Lua 脚本缓存
SCRIPT FLUSH
杀死当前正在运行的 Lua 脚本
// Redis_CompareAndSet.lua 文件
local key = KEYS[1]
local val = redis.call("GET", key);
if val == ARGV[1]
then
redis.call('SET', KEYS[1], ARGV[2])
return 1
else
return 0
end
// 执行命令
如:redis-cli -a 123456 --eval ./Redis_CompareAndSet.lua userName , zhangsan lisi
## Redis客户端执行
127.0.0.1:6379> set userName zhangsan
OK
127.0.0.1:6379> get userName
"zhangsan"
例如:同一IP在10秒内最多访问三次
// Redis_LimitIpVisit.lua
local visitNum = redis.call('incr', KEYS[1])
if visitNum == 1 then
redis.call('expire', KEYS[1], ARGV[1])
end
if visitNum > tonumber(ARGV[2]) then
return 0
end
return 1;
## LimitIP:127.0.0.1为key, 10 3表示:同一IP在10秒内最多访问三次
## 前三次返回1,代表未被限制;第四、五次返回0,代表127.0.0.1这个ip已被拦截
[root@vm01 learn_lua]# redis-cli -a 123456 --eval Redis_LimitIpVisit.lua LimitIP:127.0.0.1 , 10 3
(integer) 1
[root@vm01 learn_lua]# redis-cli -a 123456 --eval Redis_LimitIpVisit.lua LimitIP:127.0.0.1 , 10 3
(integer) 1
[root@vm01 learn_lua]# redis-cli -a 123456 --eval Redis_LimitIpVisit.lua LimitIP:127.0.0.1 , 10 3
(integer) 1
[root@vm01 learn_lua]# redis-cli -a 123456 --eval Redis_LimitIpVisit.lua LimitIP:127.0.0.1 , 10 3
(integer) 0
[root@vm01 learn_lua]# redis-cli -a 123456 --eval Redis_LimitIpVisit.lua LimitIP:127.0.0.1 , 10 3
(integer) 0
Springboot 继承 Redis 使用 Lua
# Redis数据库地址
spring.redis.host=127.0.0.1
# Redis端口
spring.redis.port=6379
# Redis密码(如果没有密码不用填写)
spring.redis.password=
// RedisCRUD.lua脚本
-- set
if KEYS[1] and ARGV[1] then
redis.call('SET', KEYS[1], ARGV[1])
return 1
end
-- get
if KEYS[1] and not ARGV[1] then
return redis.call('GET', KEYS[1])
end
-- delete
if KEYS[1] and not ARGV[1] then
redis.call('DEL', KEYS[1])
return 1
end
-- exists
if KEYS[1] and not ARGV[1] then
if redis.call('EXISTS', KEYS[1]) == 1 then
return true
else
return false
end
end
-- hset
if KEYS[1] and ARGV[1] and ARGV[2] and ARGV[3] then
redis.call('HSET', KEYS[1], ARGV[1], ARGV[2])
redis.call('EXPIRE', KEYS[1], ARGV[3])
return 1
end
-- hget
if KEYS[1] and ARGV[1] and not ARGV[2] then
return redis.call('HGET', KEYS[1], ARGV[1])
end
-- hdelete
if KEYS[1] and ARGV[1] and not ARGV[2] then
redis.call('HDEL', KEYS[1], ARGV[1])
return 1
end
-- hexists
if KEYS[1] and ARGV[1] and not ARGV[2] then
if redis.call('HEXISTS', KEYS[1], ARGV[1]) == 1 then
return true
else
return false
end
end
// 在RedisTemplate的Bean中添加
@Bean
public RedisScript<Long> redisScript() {
RedisScript<Long> redisScript = new DefaultRedisScript<>();
redisScript.setLocation(new ClassPathResource("lua/RedisCRUD.lua"));
redisScript.setResultType(Long.class);
return redisScript;
}
// 实现RedisService
@Service
public class RedisServiceImpl implements RedisService {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
@Autowired
private RedisScript<Long> redisScript;
public void set(String key, Object value) {
redisTemplate.opsForValue().set(key, value);
}
public Object get(String key) {
return redisTemplate.opsForValue().get(key);
}
public void delete(String key) {
redisTemplate.delete(key);
}
public Boolean exists(String key) {
return redisTemplate.hasKey(key);
}
public Long hset(String key, String field, Object value) {
return redisTemplate.opsForHash().put(key, field, value);
}
public Object hget(String key, String field) {
return redisTemplate.opsForHash().get(key, field);
}
public void hdelete(String key, String... fields) {
redisTemplate.opsForHash().delete(key, fields);
}
public Boolean hexists(String key, String field) {
return redisTemplate.opsForHash().hasKey(key, field);
}
public Long eval(String script, List<String> keys, List<Object> args) {
return redisTemplate.execute(RedisScript.of(script), keys, args.toArray());
}
public Long eval(List<String> keys, List<Object> args) {
return redisTemplate.execute(redisScript, keys, args.toArray());
}
}
//测试RedisService
@RunWith(SpringRunner.class)
@SpringBootTest
public class RedisServiceImplTest {
@Autowired
private RedisService redisService;
@Test
public void test() {
//第一种方式:执行string的lua
redisService.eval("redis.call('SET', KEYS[1], ARGV[1])",Collections.singletonList(hashKey), Collections.singletonList(hashValue));
//第二种方式:执行lua脚本
String key ="key";
String value ="value";
redisService.eval(Collections.singletonList(hashKey), Collections.singletonList(hashValue));
}
又称计数器算法:在指定周期内累加访问次数,当
访问次数达到设定的阈值时
,触发限流策略,当进入下一个时间周期时进行访问次数的清零。如图所示,我们要求3秒内的请求不要超过150次:
但是,貌似看似很“完美”的流量统计方式其实存在一个非常严重的
临界问题
,即:如果第2到3秒内产生了150次请求,而第3到4秒内产生了150次请求,那么其实在第2秒到第4秒这两秒内,就已经发生了300次请求了,远远大于
我们要求的3秒内的请求不要超过150次
这个限制。
重试框架,就是当我们方法失败后的重试机制
@Service
@Slf4j
public class SpringRetryAnnotationService {
@Autowired
CommonService commonService;
/**
* 如果失败,定义重试3次,重试间隔为3s,指定恢复名称,指定监听器
*/
@Retryable(value = RuntimeException.class, maxAttempts = 3, backoff = @Backoff(value = 3000L), recover = "testRecover", listeners = {"myRetryListener"})
public void test() {
commonService.test("注解式");
}
@Recover
public void testRecover(RuntimeException runtimeException) {
commonService.recover("注解式");
}
}