基于redis的分布式服务限流方案 -凯发k8国际

由于api接口无法控制调用方的行为，因此当遇到瞬时请求量激增时，会导致接口占用过多服务器资源，使得其他请求响应速度降低或是超时，更有甚者可能导致服务器宕机。 

限流指对应用服务接口的请求调用次数进行限制，对超过限制次数的请求则进行快速失败或丢弃。

限流可以应对：

1、热点业务带来的高并发请求；

2、客户端异常重试导致的并发请求；

3、恶意攻击请求；

限流算法多种多样，比如常见的：固定窗口计数器、滑动窗口计数器、漏桶、令牌桶等。本章通过redis 的lua来实现滑动窗口的计数器算法。

1、redis lua脚本如下：

local ratelimit_info = redis.pcall('hmget',keys[1],'last_time','current_token') 
local last_time = ratelimit_info[1] 
local current_token = tonumber(ratelimit_info[2]) 
local max_token = tonumber(argv[1]) 
local token_rate = tonumber(argv[2]) 
local current_time = tonumber(argv[3]) 
local reverse_time = token_rate*1000/max_token 
if current_token == nil then 
  current_token = max_token 
  last_time = current_time 
else 
  local past_time = current_time-last_time 
  local reverse_token = math.floor(past_time/reverse_time)
  current_token = current_token reverse_token 
  last_time = reverse_time*reverse_token last_time 
  if current_token>max_token then 
    current_token = max_token 
  end 
end 
local result = '0' 
if(current_token>0) then 
  result = '1' 
  current_token = current_token-1 
end 
redis.call('hmset',keys[1],'last_time',last_time,'current_token',current_token) 
redis.call('pexpire',keys[1],math.ceil(reverse_time*(max_token-current_token) (current_time-last_time))) 
return result

2、项目中引入spring-data-redis和commons-codec，相关配置请自行google。

3、redisratelimitscript类

package com.huatech.support.limit;
import org.apache.commons.codec.digest.digestutils;
import org.springframework.data.redis.core.script.redisscript;
public class redisratelimitscript implements redisscript {
   private static final string script = 
      "local ratelimit_info = redis.pcall('hmget',keys[1],'last_time','current_token') local last_time = ratelimit_info[1] local current_token = tonumber(ratelimit_info[2]) local max_token = tonumber(argv[1]) local token_rate = tonumber(argv[2]) local current_time = tonumber(argv[3]) local reverse_time = token_rate*1000/max_token if current_token == nil then current_token = max_token last_time = current_time else local past_time = current_time-last_time local reverse_token = math.floor(past_time/reverse_time) current_token = current_token reverse_token last_time = reverse_time*reverse_token last_time if current_token>max_token then current_token = max_token end end local result = '0' if(current_token>0) then result = '1' current_token = current_token-1 end redis.call('hmset',keys[1],'last_time',last_time,'current_token',current_token) redis.call('pexpire',keys[1],math.ceil(reverse_time*(max_token-current_token) (current_time-last_time))) return result"; 
  @override   
  public string getsha1() { 
    return digestutils.sha1hex(script); 
  } 
  @override   
  public class getresulttype() {     
	  return string.class; 
  } 
  @override   
  public string getscriptasstring() {     
	  return script; 
  } 
} 

4、添加ratelimit注解

package com.huatech.support.limit;
import java.lang.annotation.documented;
import java.lang.annotation.elementtype;
import java.lang.annotation.retention;
import java.lang.annotation.retentionpolicy;
import java.lang.annotation.target;
@target({ elementtype.type, elementtype.method})
@retention(retentionpolicy.runtime)
@documented
public @interface ratelimit {
	
	/**
	 * 接口标识
	 * @return
	 */
	string value() default "";
	
	/**
	 * 周期：多久为一个周期，单位s
	 * @return
	 */
	int period() default 1;
	
	/**
	 * 周期速率
	 * @return
	 */
	int rate() default 100;
	
	/**
	 * 限制类型，默认按接口限制
	 * @return
	 */
	limittype limittype() default limittype.global;
	
	/**
	 * 超限后处理方式，默认拒绝访问
	 * @return
	 */
	limitedmethod method() default limitedmethod.access_denied;
}

基于redis的分布式服务限流有两种落地方案：

一种是基于aop的切面实现，另一种是基于interceptor的拦截器实现，下面分别做介绍。

方案一：基于aspject的aop实现方案

1、添加limitaspect类

package com.huatech.common.aop;
import java.lang.reflect.method;
import java.util.arraylist;
import java.util.hashmap;
import java.util.list;
import java.util.map;
import javax.servlet.http.httpservletrequest;
import javax.servlet.http.httpservletresponse;
import org.apache.commons.lang3.stringutils;
import org.aspectj.lang.proceedingjoinpoint;
import org.aspectj.lang.annotation.around;
import org.aspectj.lang.annotation.aspect;
import org.aspectj.lang.reflect.methodsignature;
import org.slf4j.logger;
import org.slf4j.loggerfactory;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.data.redis.core.stringredistemplate;
import org.springframework.stereotype.component;
import org.springframework.web.context.request.requestcontextholder;
import org.springframework.web.context.request.servletrequestattributes;
import org.springframework.web.util.webutils;
import com.alibaba.fastjson.jsonobject;
import com.huatech.common.constant.constants;
import com.huatech.common.util.iputil;
import com.huatech.support.limit.ratelimit;
import com.huatech.support.limit.redisratelimitscript;
@aspect
@component
public class limitaspect {
	
	private static final logger logger = loggerfactory.getlogger(limitaspect.class);
	@autowired
	private stringredistemplate redistemplate;
	
	@around("execution(* com.huatech.core.controller..*(..) ) && @annotation(com.huatech.support.limit.ratelimit)")
	public object interceptor(proceedingjoinpoint joinpoint) throws throwable{
		
		methodsignature signature = (methodsignature) joinpoint.getsignature();
		method method = signature.getmethod();
		ratelimit ratelimit = method.getannotation(ratelimit.class);
		if(ratelimit !=	null) {
			servletrequestattributes requestattributes = (servletrequestattributes) requestcontextholder.getrequestattributes();
			httpservletrequest request = requestattributes.getrequest();
			httpservletresponse response = requestattributes.getresponse();
			
			class targetclass = method.getdeclaringclass();
			list keylist = new arraylist<>(1);
		    string key = ratelimit.value();
		    if(stringutils.isblank(key)){
		    	key = targetclass.getname()   "-"   method.getname();
		    }
		    switch (ratelimit.limittype()) {
			case ip:
				string ip = iputil.getremortip(request);
				key = ip   "-"   key;
				break;
			case user:
				string userid = webutils.getsessionattribute(request, constants.session_user_id).tostring();
				key = userid   "-"   key;
			default:
				break;
			}
		    keylist.add(key);
		    
		    long timer = system.currenttimemillis();
		    boolean pass = "1".equals(redistemplate.execute(new redisratelimitscript(), keylist, 
		    		integer.tostring(ratelimit.rate()), integer.tostring(ratelimit.period()), 
		    		long.tostring(timer)));
		    if(pass){
		    	return joinpoint.proceed();
		    }else{				
		    	logger.warn("接口key：{}, 周期：{}， 频率：{}", key, ratelimit.period(), ratelimit.rate());
		    	map result = new hashmap<>();
				result.put("code", "400");
				result.put("msg", "访问超过次数限制！");
				response.setcontenttype("application/json");
				response.setcharacterencoding("utf-8");
				response.getwriter().print(jsonobject.tojson(result));
		    	return null;
		    }
		    
		}else{
			return joinpoint.proceed();
		}
	}
	
}

2、在spring-mvc配置文件中开启自定义注解

3、开启limitaspect类的自动扫描操作，或者在spring配置文件中配置bean

方式二：基于interceptor的拦截器实现方案

1、添加ratelimitinterceptor类

public class ratelimitinterceptor extends handlerinterceptoradapter {
	
	private static final logger logger = loggerfactory.getlogger(ratelimitinterceptor.class);
	@autowired stringredistemplate redistemplate;
	@override
	public boolean prehandle(httpservletrequest request, httpservletresponse response, object handler) throws exception {
		if (handler instanceof handlermethod) {
			handlermethod method = (handlermethod) handler;
			final ratelimit ratelimit = method.getmethodannotation(ratelimit.class);
			if (ratelimit != null) {
				// 令牌名称
				list keylist = new arraylist<>(1);
			    string key = ratelimit.value();
			    if(stringutils.isblank(key)){
			    	key = method.getclass().getname()   "-"   method.getmethod().getname();
			    }
			    switch (ratelimit.limittype()) {
					case ip:
						string ip = iputil.getremortip(request);
						key = ip   "-"   key;
						break;
					case user:
						string userid = webutils.getsessionattribute(request, constants.session_user_id).tostring();
						key = "uid:"   userid   "-"   key;
					default:
						break;
				}
			    keylist.add(key);
			    
			    long timer = system.currenttimemillis();
			    boolean pass = "1".equals(redistemplate.execute(new redisratelimitscript(), keylist, 
			    		integer.tostring(ratelimit.rate()), integer.tostring(ratelimit.period()), 
			    		long.tostring(timer)));
			    if(pass){
			    	return true;
			    }else{				
			    	logger.warn("接口key：{}, 周期：{}， 频率：{}", key, ratelimit.period(), ratelimit.rate());
			    	map result = new hashmap<>();
					result.put("code", "400");
					result.put("msg", "访问超过次数限制！");
					response.setcontenttype("application/json");
					response.setcharacterencoding("utf-8");
					response.getwriter().print(jsonobject.tojson(result));
			    	return false;
			    }
				
			}
		}
		return true;
	}
}

2、在spring-mvc配置文件中配置拦截器

 	
 		
		****
		
		
			
			
		
	 

使用@ratelimit

  在controller类的方法头上添加ratelimit注解

 /**
     * 服务端ping地址
     * @param request
     * @param response
     * @throws exception
     */
    @requestmapping(value = "/api/app/open/ping.htm")
    @ratelimit(value="ping", period=5, rate=5)
    public void ping(httpservletrequest request, httpservletresponse response) throws exception {
    	map data = new hashmap();
    	data.put("time", system.currenttimemillis());
    	servletutils.successdata(response,data);
    }

 另外两个枚举类

package com.huatech.support.limit;
/**
 * 超限处理方式
 * @author lh@erongdu.com
 * @since 2019年8月28日
 * @version 1.0
 *
 */
public enum limitedmethod {
	
	/**
	 * 拒绝访问（直接拒绝访问，不预警）
	 */
	access_denied,
	/**
	 * 预警短信（发送预警短信，但不拒绝访问）
	 */
	warn_sms,
	/**
	 * 拒绝访问并预警
	 */
	denied_and_sms
	;
}

package com.huatech.support.limit;
/**
 * 接口限制类型
 * @author lh@erongdu.com
 * @since 2019年8月29日
 * @version 1.0
 *
 */
public enum limittype {
	
	/**
	 * 整个接口限制
	 */
	global("接口"), 
	/**
	 * ip层面限制
	 */
	ip("ip"), 
	/**
	 * 用户层面限制
	 */
	user("用户");
	
	public string value;
	private limittype(string value) {
		this.value = value;
	}
	
	
}

 iputil工具类

package com.huatech.common.util;
import java.net.inetaddress;
import java.net.unknownhostexception;
import javax.servlet.http.httpservletrequest;
import org.slf4j.logger;
import org.slf4j.loggerfactory;
/**
 * 
 * @author lh@erongdu.com
 * @since 2019年8月29日
 * @version 1.0
 *
 */
public class iputil {
	
	public static final logger logger = loggerfactory.getlogger(iputil.class);
    
	/**
	 * 获取请求ip
	 * @param request
	 * @return
	 */
	public static string getremortip(httpservletrequest request) {
		string ip = request.getheader("x-forwarded-for");
		if (ip == null || ip.length() == 0 || "unknown".equalsignorecase(ip)) {
			ip = request.getheader("x-real-ip");
		}
		if (ip == null || ip.length() == 0 || "unknown".equalsignorecase(ip)) {
			ip = request.getheader("wl-proxy-client-ip");
		}
		if (ip == null || ip.length() == 0 || "unknown".equalsignorecase(ip)) {
			ip = request.getremoteaddr();
		}
		
		 //这里主要是获取本机的ip,可有可无  
	    if ("127.0.0.1".equals(ip) || ip.endswith("0:0:0:0:0:0:1")) {  
	        // 根据网卡取本机配置的ip  
	        inetaddress inet = null;
	        try {  
	            inet = inetaddress.getlocalhost();  
	        } catch (unknownhostexception e) {  
	            logger.error(e.getmessage(), e);
	        }
	        if(inet != null){
	        	ip = inet.gethostaddress();
	        }
	        return ip;
	    } 
		if(ip.length() > 0){
			string[] iparray = ip.split(",");
			if (iparray != null && iparray.length > 1) {
				return iparray[0];
			}
			return ip;
		}
		
		return "";
	}
}