计数器的基本使用

    /**
     * redis计数器
     * @param key 键
     * @return
     */
    public Long test(String key){
        //key为key，delta为递增数目，返回值为增长后的数量。
        Long increment = redisTemplate.opsForValue().increment(key, 1);
        return increment;
    }

 

 短信发送请求次数限制

    /**
     *  短信验证码请求次数限制
     * @param mobile
     * @return
     */
    public String sendVerifyCode(String mobile){
        Long increment = redisTemplate.opsForValue().increment(mobile, 1);
        if (increment == 1) {
            //设置有效期一分钟
            redisTemplate.expire(mobile, 60, TimeUnit.SECONDS);
        }
        if (increment > 1) {
            return "每分钟只能发送一次短信";
        }

        //发送验证码


        return "验证码已发送";
    }

IP请求次数限制

定义消息发送类

/**
 * 拦截器消息发送
 * @author 向振华
 * @date 2018/11/28 10:52
 */
public class BaseInterceptor extends HandlerInterceptorAdapter {

    /**
     * 发送异常信息
     * @param request
     * @param response
     * @param code
     * @param msg
     * @param data
     */
    public void sendMsg(HttpServletRequest request, HttpServletResponse response, int code, String msg, String data) {
        response.setCharacterEncoding("UTF-8");
        response.setContentType("application/json");
        //封装成返回给前端的固定格式
        ResponseMessage responseMessage = new ResponseMessage(code, msg, data);
        String jsonObject = JSONObject.toJSONString(responseMessage);
        PrintWriter out = null;
        try {
            out = response.getWriter();
            out.append(jsonObject);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (out != null) {
                out.close();
            }
        }
    }
}


@Data
public class ResponseMessage<T> implements Serializable {
    private static final long serialVersionUID = -2988364810346364595L;

    /**
     * 1:成功 -1:失败
     */
    private int code;
    /**
     * 错误信息
     */
    private String errorMsg;
    /**
     * 返回数据
     */
    private T data;

    public ResponseMessage(int code, String errorMsg, T data) {
        this.code = code;
        this.errorMsg = errorMsg;
        this.data = data;
    }

    public ResponseMessage(int code, String errorMsg) {
        this.code = code;
        this.errorMsg = errorMsg;
    }

    public ResponseMessage() {
    }
}

拦截器编写

/**
 * IP请求次数限制
 * @author 向振华
 * @date 2018/11/28 10:55
 */
public class CounterInterceptor extends BaseInterceptor {

    @Resource
    private RedisTemplate redisTemplate;

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        String ip = getRemoteHost(request);
        String key = ip + request.getRequestURI() + request.getSession().getId();
        Long increment = redisTemplate.opsForValue().increment(key, 1);
        if (increment == 1) {
            //设置有效期一分钟
            redisTemplate.expire(key, 60, TimeUnit.SECONDS);
        }
        //次数限制为5次
        if (increment > 5) {
            sendMsg(request, response, -1, "次数限制", request.getRequestURI());
            return false;
        }
        return true;
    }

    /**
     * 获取请求IP地址
     * @param request
     * @return
     */
    public String getRemoteHost(HttpServletRequest request) {
        String ip = request.getHeader("x-forwarded-for");
        if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
            ip = request.getHeader("Proxy-Client-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();
        }
        return ip.equals("0:0:0:0:0:0:0:1") ? "127.0.0.1" : ip;
    }
}

拦截器配置

/**
 *  拦截器配置
 * @author 向振华
 * @date 2018/11/28 11:30
 */
@Configuration
@EnableTransactionManagement
public class MyWebAppConfig extends WebMvcConfigurerAdapter {

    @Bean
    CounterInterceptor counterInterceptor() {
        return new CounterInterceptor();
    }

    /**
     * 拦截器配置
     * @param registry
     */
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        //配置IP请求次数限制
        registry.addInterceptor(counterInterceptor());

        super.addInterceptors(registry);
    }
}

 

海量数据实时计算系统在高并发互联网应用中的原理和实践

占超群(阿里巴巴)

2012-12-05 16:50

永泰大宴会厅C

演讲幻灯片：海量数据实时计算系统在高并发互联网应用中的原理和实践[PDF]




随着互联网个性化时代的来临，很多需要通过多角度多属性进行实时分析的需求应运而生，如想即时统计最近任意几天、指定省份、指定地市、指定性别、指定年龄、指定频道的用户情况；如想结合用户的省份、地市、年龄、性别、偏好、最近购买情况给用户即时推荐最热门的商品；如想统计出多个关键词组合在省份、地市、年龄、性别、偏好等条件下的分布情况；而这些多维度多属性统计会存在无法穷举和无法通过离线计算进行预算的问题，如果固化业务逻辑则无法应对互联网业务的快速变化，在高并发互联网应用中解决这类问题则面临更大的挑战。

Garuda 是一个针对海量数据进行任意维度的密集计算与检索的支持高并发低延时的在线数据应用系统。目前在淘宝主站、SNS、数据魔方、淘宝指数等产品中得到了广泛应用，给这些产品提供高并发的实时计算服务；当前集群数据量已超过300亿条，平均每个请求涉及到300万条以上记录的实时多维度计算，请求响应时间小于50ms，QPS 超过300。本次分享将就Garuda设计原理、应用场景、历史变迁进行详细介绍，也将会介绍结合Redis、HBase的一些过渡方案的快速实现原理。

1.3.4 设计并发友好的应用程序
 
使用传统的命令风格编写多线程应用程序时，我们要面对两个问题：
■很难让现有的顺序代码变成并行的代码，因为要显式使用线程，必须修改大部分的基本代码。
■使用共享状态和锁定是困难的。必须仔细考虑如何使用锁来避免争用条件和死锁，留下足够的空间并行执行。
函数式编程给了我们答案：
■使用声明编式程风格时，可以在现有的代码中引入并行。替换很少几个基元，就能把命令与并行方式执行的命令组合起来。
■由于具有不可变性，不可能引入争用条件，可以编写无需锁定的代码；只要看到哪部分程序是独立的，就可以立刻把它修改成并行运行的任务。
这两个方面能够影响我们如何设计应用程序，以及更易于编写并行运行的代码，充分利用多核的机器。但是，写不可变代码，并不开然获得并化行，还有棘手的工作要做，只是函数式编程能够将花在并行化上的额外努力降至最低。
还没有完，当开始用函数方式进行思考，还会发现在设计中有其他改变……

同样还是工作中用到的ExecutorService这个线程池。
多线程这块一直在我印象里就是一个难缠的家伙。自己本来也是新手
还用了ExecutorService这个类，所以不用想，给我带来了很大的麻烦。ExecutorService这个从API查得好像有三种模式。具体的
我也就不多少有什么区别了，因为我也搞不清。最终我选择了Executors.newFixedThreadPool(10)； 这个是生成一个固定线程数
的线程池，在这里我就定义了一个只能容纳10个线程的线程池。
那么定义好了怎么用呢，看看代码吧：
int pageDownloaderPoolNumber =10;
public static ExecutorService pageDownloaderPool = Executors.newFixedThreadPool(pageDownloaderPoolNumber);
首先定义好pageDownloaderPool，最大线程数为10。
crawlerFactory cf = new crawlerFactory(crawler); // 此处为new一个线程对象，因为在线程池中放的任务必须是一个线程类的对象
即继承了Runnable接口或其他某某的类的对象。
pageFutureList.add(pageDownloaderPool.submit(cf));//此处是将线程任务放进线程池中执行。注意，因为该线程池可以返回
一个任务队列。所以我们要定义一个FutureList来获得这些任务队列：
public static List<Future> pageFutureList = new ArrayList<Future>(pageDownloaderPoolNumber);
这样线程池里的任务就启动了，但要记住，如果你有30个线程同时启动，那么最多有10个在工作，同时还得对任务队列进行管理
特别是pageFutureList.add(pageDownloaderPool.submit(cf));这个在循环中的话一定要加上控制：
while(){
----------
hasFreeThread(pageFutureList);
if(pageFutureList.size() < pageDownloaderPoolNumber){
     pageFutureList.add(pageDownloaderPool.submit(cf));
｝
 
}
 
同时有这么一个方法去检测队列是否有空闲的：
public static boolean hasFreeThread(List<Future> list){

        synchronized (list) {

            if(list.size() == 0){

                return true;

            }

            Future finishedFuture = null;

            try{

                for (Future temp : list) {

                    if(temp.isDone()/* || temp.get() == null*/){

                        finishedFuture = temp;

                        return true;

                    }

                }

            }catch(Exception e){

                e.printStackTrace();

                return false;

            }finally{

                if(finishedFuture != null){

                    list.remove(finishedFuture);

                }

            }

            return false;

        }

    }
这样的话任务队列中就保持着永远最大数是10，当完成一个任务变成9后，下次就再加入一个任何到队列，就避免了线程的阻塞