来自于《Redis深度历险：核心原理和应用实践》，进行分析。
业务场景：一个id，限制每秒只能访问多少次。
下面的代码使用了redission工具类。
不合理的方式：
1、以下这种方式，没有考虑并发问题，一瞬间所有的请求，同时get，然后还没来得及increment，然后就判断错误了。
2、改善方式，先increment，再get。（不过这样的话，get会得到别人increment后的值，有些业务不合适）。或者，使用下面那样的自旋的方式。
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
String ip = request.getRemoteHost();
String key = IP_PREFIX + ip;
Object value = redisService.get(BLACK_PREFIX + ip);
if(value != null){
// 在黑名单则不让访问
return false;
}
Object o = redisService.get(key);
if(o == null){
redisService.set(key, 1, 2000L);
} else if(o.equals(50)){
// 2秒超过50个请求就屏蔽掉,放入黑名单,1小时候之后不能访问
redisService.set(BLACK_PREFIX + ip, 1, 3600L);
}else{
redisService.increment(key, 1);
}
return true;
}
redis自增，必须要使用StringRedisTemplate，而且自定义好key策略。
@Bean
public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory factory) {
StringRedisTemplate template = new StringRedisTemplate(factory);
//定义key序列化方式
//RedisSerializer<String> redisSerializer = new StringRedisSerializer();//Long类型会出现异常信息;需要我们上面的自定义key生成策略，一般没必要
//定义value的序列化方式
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
// template.setKeySerializer(redisSerializer);
template.setValueSerializer(jackson2JsonRedisSerializer);
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
1、自旋的方式。
这不是书中的方式。
缺点：无法精确的限制，每秒10次，最大的可能会到18次。
优点：内存利用少。
RedissonClient redisson = Redisson.create();
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
String id = "123";
RBucket<Boolean> bucket = redisson.getBucket("blackId:" + id);
// 是否是黑名单得
if(bucket.get() == true){
// 或者返回最近的一个请求的缓存
return;
}
RAtomicLong idAccessNum = redisson.getAtomicLong(id);
while (true){
long num = idAccessNum.get();
// 没有初始化，则设置过期时间
// 这种方式，并非准确10秒10次就到了阈值，可能前10秒已经有9次，然后后10秒又请求了9次，
// 最长可能会有18次。但是我们不需要准确的10次，所以无影响。
if(num == 0){
idAccessNum.expire(10, TimeUnit.SECONDS);
}
// 超过了10次，则进入黑名单
if(num > 10){
// 加入黑名单，30秒之后不能再访问
bucket.set(true, 30, TimeUnit.SECONDS);
// 或者返回最近的一个请求的缓存
return;
}
// 这里自旋，只针对同一个用户id，因为理论上同一个用户的请求是线性的，所以这里自旋
// 不会对性能造成很大的影响
boolean updateSuccess = idAccessNum.compareAndSet(num, num + 1);
// 是否更新成功
if(updateSuccess){
break;
}
}
二、使用zset
这是书中的方式。
优点：能够精确的统计次数。
缺点：比较浪费内存。每请求一次，就ad一个数。
public class SimpleRateLimiter {
private Jedis jedis;
public SimpleRateLimiter(Jedis jedis) {
this.jedis = jedis;
}
public boolean isActionAllowed(String userId, String actionKey, int period, int maxCount) {
String key = String.format("hist:%s:%s", userId, actionKey);
long nowTs = System.currentTimeMillis();
Pipeline pipe = jedis.pipelined();
pipe.multi();
// value 没有实际的意义，保证唯一就可以
pipe.zadd(key, nowTs, "" + nowTs);
pipe.zremrangeByScore(key, 0, nowTs - period * 1000);
Response<Long> count = pipe.zcard(key);
pipe.expire(key, period + 1);
pipe.exec();
pipe.close();
return count.get() <= maxCount;
}
public static void main(String[] args) {
Jedis jedis = new Jedis();
SimpleRateLimiter limiter = new SimpleRateLimiter(jedis);
for (int i = 0; i < 20; i++) {
System.out.println(limiter.isActionAllowed("laoqian", "reply", 60, 5));
}
}
}
如果使用redission。
则代码大概这么写
public static void main(String[] args) {
RedissonClient redisson = Redisson.create();
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
String id = "123";
RBucket<Boolean> bucket = redisson.getBucket("blackId:" + id);
// 是否是黑名单得
if(bucket.get() == true){
// 或者返回最近的一个请求的缓存
return;
}
long nanoTime = System.nanoTime();
RScoredSortedSet<Object> zset = redisson.getScoredSortedSet(id);
zset.expire(10, TimeUnit.SECONDS);
zset.add(nanoTime, nanoTime);
zset.removeRangeByScore(0, true,
nanoTime - 10 * 1000 * 1000 * 1000, true);
int size = zset.size();
// 超过了10次，则进入黑名单
if(size > 10){
// 加入黑名单，30秒之后不能再访问
bucket.set(true, 30, TimeUnit.SECONDS);
// 或者返回最近的一个请求的缓存
return;
}
// 放行
}
如果没有redis，只是单机，用java可以大概这么写：
public class TestLimitFlow {
private Lock lock = new ReentrantLock();
private Map<String, Entity> map = new ConcurrentHashMap<>();
public TestLimitFlow() {
new Thread(() -> {
while (true) {
System.out.println(map.size());
map.forEach((key, value) -> {
long now = System.nanoTime();
if (value.expireTime < now) {
map.remove(key);
}
});
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
class Entity<T> {
public long expireTime;
T o;
}
public void interceptor(String ip) throws InterruptedException {
lock.lock();
Entity e = map.get(ip);
long nanoTime = System.nanoTime();
if (e == null) {
TreeSet<Long> treeSet = new TreeSet<>();
e = new Entity<TreeSet<Long>>();
e.expireTime = nanoTime + 10L * 1000 * 1000 * 1000;
e.o = treeSet;
map.put(ip, e);
}
lock.unlock();
synchronized (e) {
e.expireTime = nanoTime + 10L * 1000 * 1000 * 1000;
TreeSet<Long> zset = (TreeSet<Long>) e.o;
zset.add(nanoTime);
List<Long> deleteKey = Lists.newArrayList();
for(Long item : zset){
if(item < nanoTime - 10L * 1000 * 1000 * 1000){
System.out.println("true");
// zset.remove(item);
deleteKey.add(item);
}else {
break;
}
}
deleteKey.forEach(item -> {
zset.remove(item);
});
System.out.println(zset.size());
if (zset.size() > 10) {
// 加入黑名单
System.out.println("黑名单：" + ip + ":" + zset.size());
}
}
}
public static void main(String[] args) throws InterruptedException {
TestLimitFlow testXL = new TestLimitFlow();
for(int j = 0; j < 10;j++){
int finalJ = j;
new Thread(){
@Override
public void run() {
for (int i = 0; i < 12; i++) {
// TimeUnit.SECONDS.sleep(1);
try {
testXL.interceptor("localhost" + finalJ);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}.start();
}
}
}
网友评论