滑动窗口算法

Author: beston123

pom.xml

@@ -124,6 +124,18 @@
             <version>1.16.22</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.5</version>
+        </dependency>
+
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>2.6</version>
+        </dependency>
+
     </dependencies>
 
     <build>

src/main/java/code/algorithm/slidingwindow/SlidingNode.java

@@ -0,0 +1,84 @@
+package code.algorithm.slidingwindow;
+
+import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * 〈滑动节点〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author zixiao
+ * @date 2019/5/24
+ */
+public class SlidingNode implements Serializable{
+
+    /**
+     * 周期秒数
+     */
+    private int period;
+
+    /**
+     * 总数量
+     */
+    private AtomicInteger totalCount = new AtomicInteger();
+
+    /**
+     * 总RT
+     */
+    private LongAdder totalRt = new LongAdder();
+
+    public SlidingNode(int period) {
+        this.period = period;
+    }
+
+    public int getPeriod(){
+        return period;
+    }
+
+    public void addCount(int count){
+        totalCount.addAndGet(count);
+    }
+
+    public void addRt(long rt){
+        totalRt.add(rt);
+    }
+
+    public int totalCount(){
+        return totalCount.get();
+    }
+
+    public long totalRt(){
+        return totalRt.longValue();
+    }
+
+    public long avgRt(){
+        return totalCount() == 0 ? 0 : totalRt() / totalCount();
+    }
+
+    public int tps(){
+        return totalCount() / getPeriod();
+    }
+
+    /**
+     * 重置计数
+     */
+    public void reset(){
+        totalCount.set(0);
+        totalRt.reset();
+    }
+
+    @Override
+    public String toString() {
+        return "SlidingNode{" +
+                "period=" + period +
+                ", totalCount=" + totalCount() +
+                ", totalRt=" + totalRt() +
+                '}';
+    }
+
+    public String prettyPrint(){
+        return String.format("TotalSec=%s, totalCount=%s, TPS=%s, totalRT=%s, avgRT=%s",
+                getPeriod(), totalCount(), tps(), totalRt(), avgRt());
+    }
+}

src/main/java/code/algorithm/slidingwindow/SlidingWindow.java

@@ -0,0 +1,120 @@
+package code.algorithm.slidingwindow;
+
+import java.io.Closeable;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 〈滑动窗口〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author zixiao
+ * @date 2019/5/24
+ */
+public class SlidingWindow implements Closeable{
+
+    /**
+     * 节点数据
+     */
+    private SlidingNode[] nodes;
+
+    /**
+     * slot数
+     */
+    private int slotNum;
+
+    /**
+     * 当前slot位置
+     */
+    private int currentSlot = 0;
+
+    /**
+     * 窗口大小
+     */
+    private int windowSize;
+
+    /**
+     * 单节点的周期（秒）
+     */
+    private int nodePeriod;
+
+    private ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(1);
+
+    public SlidingWindow(int slotNum, int windowSize){
+        this(slotNum, windowSize, 1);
+    }
+
+    public SlidingWindow(int slotNum, int windowSize, int nodePeriod){
+        this.slotNum = slotNum;
+        this.windowSize = windowSize;
+        this.nodePeriod = nodePeriod;
+
+        this.nodes = new SlidingNode[slotNum];
+        this.scheduler = new ScheduledThreadPoolExecutor(1);
+
+        this.init();
+    }
+
+    private void init(){
+        for(int i=0; i<slotNum; i++){
+            nodes[i] = new SlidingNode(nodePeriod);
+        }
+
+        scheduler.scheduleAtFixedRate(new Runnable() {
+            @Override
+            public void run() {
+                slidingToNext();
+            }
+        }, nodePeriod, nodePeriod, TimeUnit.SECONDS);
+    }
+
+    /**
+     * 滑动到下一个窗口
+     */
+    private void slidingToNext(){
+        int nextSlot = currentSlot;
+        if (nextSlot >= (slotNum-1)) {
+            nextSlot = 0;
+        }else{
+            nextSlot++;
+        }
+        //清除数据
+        nodes[nextSlot].reset();
+        currentSlot = nextSlot;
+    }
+
+    /**
+     * 上一个窗口
+     * @return
+     */
+    public SlidingNode lastWindow(){
+        int slotIndex = currentSlot;
+        SlidingNode total = new SlidingNode(windowSize * nodePeriod);
+        for (int i=0; i< windowSize; i++){
+            slotIndex = lastNodeIndex(slotIndex);
+            total.addCount(nodes[slotIndex].totalCount());
+            total.addRt(nodes[slotIndex].totalRt());
+        }
+        return total;
+    }
+
+    /**
+     * 上一个节点位置
+     * @param slotIndex
+     * @return
+     */
+    private int lastNodeIndex(int slotIndex){
+        return (slotIndex == 0) ? (slotNum-1) : (slotIndex - 1);
+    }
+
+    @Override
+    public void close(){
+        scheduler.shutdownNow();
+    }
+
+    public SlidingNode current(){
+        return nodes[currentSlot];
+    }
+
+}

src/main/java/code/algorithm/slidingwindow/SlidingWindowTest.java

@@ -0,0 +1,67 @@
+package code.algorithm.slidingwindow;
+
+import java.util.Random;
+import java.util.concurrent.*;
+
+/**
+ * 〈一句话功能简述〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author zixiao
+ * @date 2019/5/24
+ */
+public class SlidingWindowTest {
+
+    public static void main(String[] args) throws InterruptedException {
+        int windowSize = 5;
+        SlidingWindow slidingWindow = new SlidingWindow(60, windowSize);
+        ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(1);
+        scheduler.scheduleAtFixedRate(new Runnable() {
+            @Override
+            public void run() {
+                SlidingNode node = slidingWindow.lastWindow();
+                System.out.println(node.prettyPrint());
+            }
+        }, windowSize, 3, TimeUnit.SECONDS);
+
+        Thread t = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                for(int i=0; i<2000; i++){
+                    int rt = new Random().nextInt(21);
+                    slidingWindow.current().addCount(1);
+                    slidingWindow.current().addRt(rt);
+                    try {
+                        Thread.sleep(rt);
+                    } catch (InterruptedException e) {
+                        e.printStackTrace();
+                    }
+                }
+            }
+        });
+        t.setDaemon(true);
+        t.start();
+
+        Thread t2 = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                for(int i=0; i<2000; i++){
+                    int rt = new Random().nextInt(21);
+                    slidingWindow.current().addCount(1);
+                    slidingWindow.current().addRt(rt);
+                    try {
+                        Thread.sleep(rt);
+                    } catch (InterruptedException e) {
+                        e.printStackTrace();
+                    }
+                }
+            }
+        });
+        t2.setDaemon(true);
+        t2.start();
+
+        Thread.sleep(30000);
+        scheduler.shutdownNow();
+        slidingWindow.close();
+    }
+}