Java简单实现滑动窗口

由于最近有一个统计单位时间内某key的访问次数的需求,譬如每5秒访问了redis的某key超过100次,就取出该key单独处理。

这样的单位时间统计,很明显我们都知道有个边界问题,譬如5秒内100次的限制。刚好前4.99秒访问都是0,最后0.01秒来了100次,5.01秒又来了100次。也就是访问有明显的毛刺情况出现,为了弱化这个毛刺情况,我们可以采用滑动窗口。

滑动窗口

滑动窗口的主要原理比较简单,就是将这个单位时间进行拆分,譬如5秒的统计范围,我们将它划分成5个1秒。

当请求进来时,先判断当前请求属于这5个1秒的时间片中的哪个,然后将对应的时间片对应的统计值加1,再判断当前加上前4个时间片的次数总和是否已经超过了设置的阈值。

当时间已经到达第6个时间片时,就把第一个时间片给干掉,因为无论第一片是多少个统计值,它都不会再参与后续的计算了。

就这样,随着时间的推移,统计值就随着各个时间片的滚动,不断地进行统计。

具体要将单位时间拆分为多少片,要根据实际情况来决定。当然,毫无疑问的是切分的越小,毛刺现象也越少。系统统计也越准确,随之就是内存占用会越大,因为你的这个窗口的数组会更大。

代码实现思路就是定义好分片数量,每个分片都有一个独立的计数器,所有的分片合计为一个数组。当请求来时,按照分片规则,判断请求应该划分到哪个分片中去。要判断是否超过阈值,就将前N个统计值相加,对比定义的阈值即可。

代码我直接引用别人写好的了,源代码在https://www.iteye.com/blog/go12345-1744728

1import java.util.concurrent.atomic.AtomicInteger; 2 3/** 4 * 滑动窗口。该窗口同样的key,都是单线程计算。 5 * 6 * @author wuweifeng wrote on 2019-12-04. 7 */ 8public class SlidingWindow { 9 /** 10 * 循环队列,就是装多个窗口用,该数量是windowSize的2倍 11 */ 12 private AtomicInteger[] timeSlices; 13 /** 14 * 队列的总长度 15 */ 16 private int timeSliceSize; 17 /** 18 * 每个时间片的时长,以毫秒为单位 19 */ 20 private int timeMillisPerSlice; 21 /** 22 * 共有多少个时间片(即窗口长度) 23 */ 24 private int windowSize; 25 /** 26 * 在一个完整窗口期内允许通过的最大阈值 27 */ 28 private int threshold; 29 /** 30 * 该滑窗的起始创建时间,也就是第一个数据 31 */ 32 private long beginTimestamp; 33 /** 34 * 最后一个数据的时间戳 35 */ 36 private long lastAddTimestamp; 37 38 public static void main(String[] args) { 39 //1秒一个时间片,窗口共5个 40 SlidingWindow window = new SlidingWindow(100, 4, 8); 41 for (int i = 0; i < 100; i++) { 42 System.out.println(window.addCount(2)); 43 44 window.print(); 45 System.out.println("--------------------------"); 46 try { 47 Thread.sleep(102); 48 } catch (InterruptedException e) { 49 e.printStackTrace(); 50 } 51 } 52 } 53 54 public SlidingWindow(int duration, int threshold) { 55 //超过10分钟的按10分钟 56 if (duration > 600) { 57 duration = 600; 58 } 59 //要求5秒内探测出来的, 60 if (duration <= 5) { 61 this.windowSize = 5; 62 this.timeMillisPerSlice = duration * 200; 63 } else { 64 this.windowSize = 10; 65 this.timeMillisPerSlice = duration * 100; 66 } 67 this.threshold = threshold; 68 // 保证存储在至少两个window 69 this.timeSliceSize = windowSize * 2; 70 71 reset(); 72 } 73 74 public SlidingWindow(int timeMillisPerSlice, int windowSize, int threshold) { 75 this.timeMillisPerSlice = timeMillisPerSlice; 76 this.windowSize = windowSize; 77 this.threshold = threshold; 78 // 保证存储在至少两个window 79 this.timeSliceSize = windowSize * 2; 80 81 reset(); 82 } 83 84 /** 85 * 初始化 86 */ 87 private void reset() { 88 beginTimestamp = SystemClock.now(); 89 //窗口个数 90 AtomicInteger[] localTimeSlices = new AtomicInteger[timeSliceSize]; 91 for (int i = 0; i < timeSliceSize; i++) { 92 localTimeSlices[i] = new AtomicInteger(0); 93 } 94 timeSlices = localTimeSlices; 95 } 96 97 private void print() { 98 for (AtomicInteger integer : timeSlices) { 99 System.out.print(integer + "-"); 100 } 101 } 102 103 /** 104 * 计算当前所在的时间片的位置 105 */ 106 private int locationIndex() { 107 long now = SystemClock.now(); 108 //如果当前的key已经超出一整个时间片了,那么就直接初始化就行了,不用去计算了 109 if (now - lastAddTimestamp > timeMillisPerSlice * windowSize) { 110 reset(); 111 } 112 113 return (int) (((now - beginTimestamp) / timeMillisPerSlice) % timeSliceSize); 114 } 115 116 /** 117 * 增加count个数量 118 */ 119 public boolean addCount(int count) { 120 //当前自己所在的位置,是哪个小时间窗 121 int index = locationIndex(); 122// System.out.println("index:" + index); 123 //然后清空自己前面windowSize到2*windowSize之间的数据格的数据 124 //譬如1秒分4个窗口,那么数组共计8个窗口 125 //当前index为5时,就清空6、7、8、1。然后把2、3、4、5的加起来就是该窗口内的总和 126 clearFromIndex(index); 127 128 int sum = 0; 129 // 在当前时间片里继续+1 130 sum += timeSlices[index].addAndGet(count); 131 //加上前面几个时间片 132 for (int i = 1; i < windowSize; i++) { 133 sum += timeSlices[(index - i + timeSliceSize) % timeSliceSize].get(); 134 } 135 System.out.println(sum + "---" + threshold); 136 137 lastAddTimestamp = SystemClock.now(); 138 139 return sum >= threshold; 140 } 141 142 private void clearFromIndex(int index) { 143 for (int i = 1; i <= windowSize; i++) { 144 int j = index + i; 145 if (j >= windowSize * 2) { 146 j -= windowSize * 2; 147 } 148 timeSlices[j].set(0); 149 } 150 } 151 152} 153 154 155import java.util.concurrent.Executors; 156import java.util.concurrent.ScheduledExecutorService; 157import java.util.concurrent.ThreadFactory; 158import java.util.concurrent.TimeUnit; 159import java.util.concurrent.atomic.AtomicLong; 160 161/** 162 * 用于解决高并发下System.currentTimeMillis卡顿 163 * @author lry 164 */ 165public class SystemClock { 166 167 private final int period; 168 169 private final AtomicLong now; 170 171 private static class InstanceHolder { 172 private static final SystemClock INSTANCE = new SystemClock(1); 173 } 174 175 private SystemClock(int period) { 176 this.period = period; 177 this.now = new AtomicLong(System.currentTimeMillis()); 178 scheduleClockUpdating(); 179 } 180 181 private static SystemClock instance() { 182 return InstanceHolder.INSTANCE; 183 } 184 185 private void scheduleClockUpdating() { 186 ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { 187 @Override 188 public Thread newThread(Runnable runnable) { 189 Thread thread = new Thread(runnable, "System Clock"); 190 thread.setDaemon(true); 191 return thread; 192 } 193 }); 194 scheduler.scheduleAtFixedRate(() -> now.set(System.currentTimeMillis()), period, period, TimeUnit.MILLISECONDS); 195 } 196 197 private long currentTimeMillis() { 198 return now.get(); 199 } 200 201 /** 202 * 用来替换原来的System.currentTimeMillis() 203 */ 204 public static long now() { 205 return instance().currentTimeMillis(); 206 } 207}

参照代码main方法,通过修改每个时间片的时间,窗口数量,阈值,来进行测试。

这就是简单实现了。

点赞
收藏

评论区

加载中...

相关推荐

MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1

文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s

Oracle 分组与拼接字符串同时使用

SELECTT.,ROWNUMIDFROM(SELECTT.EMPLID,T.NAME,T.BU,T.REALDEPART,T.FORMATDATE,SUM(T.S0)S0,MAX(UPDATETIME)CREATETIME,LISTAGG(TOCHAR(

MySQL部分从库上面因为大量的临时表tmp_table造成慢查询

背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_

手写Java HashMap源码

HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22

Java日期时间API系列31

  时间戳是指格林威治时间1970年01月01日00时00分00秒起至现在的总毫秒数,是所有时间的基础,其他时间可以通过时间戳转换得到。Java中本来已经有相关获取时间戳的方法,Java8后增加新的类Instant等专用于处理时间戳问题。 1获取时间戳的方法和性能对比1.1获取时间戳方法Java8以前

Twitter的分布式自增ID算法snowflake (Java版)

概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移