流量统计是监控系统中非常重要的一部分，用来计算单位时间内通过的平均流量。本例简单用代码模拟了流量请求的模式，采用单独一个检测线程，里面实现了一个方法watch(int byteSize)，用来统计通过的流量，内部采用一个原子类计数相加，另外在阶段定时统计完成后进行清零操作，重新开始累计。此例子不仅限于流量统计，其他类似场景也可以使用，比如访问量，活跃数等，是一个简单的模型。实现一般，有改进空间。以下是代码：

package com.igoso.learning;

import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by igoso on 18-3-6.
 */
public class SimpleFlowTest {

    public static void main(String[] args) {
        final Watcher watcher = new Watcher("watcher", 5);
        watcher.start();

        int maxByteSize = 10000;
        final Thread[] producers = new Thread[5];
        for (int i = 0; i < 5; i++) {
            producers[i] = new Producer(watcher, "producer-" + i, maxByteSize);
            producers[i].start();
        }
    }
}

class Producer extends Thread {
    private Watcher watcher = null;
    private int maxByteSize = 10000;
    private long count = 0;

    public Producer(final Watcher watcher, String name, int maxByteSize) {
        this.watcher = watcher;
        this.setName(name);
        this.maxByteSize = maxByteSize;
    }

    @Override
    public void run() {
        for (; ; ) {
            try {
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            int byteSize = produce();
            watcher.watch(byteSize);
        }
    }

    /**
     * 模拟每个数据包的大小
     *
     * @return
     */
    private int produce() {
        int size = new Random().nextInt(maxByteSize) + 1;
        if (count + size >= Long.MAX_VALUE) {
            count = 0;
        } else {
            count += size;
        }
        return size;
    }
}

class Watcher extends Thread {
    final private AtomicInteger increment = new AtomicInteger(0);
    private int period = 5;

    public Watcher(String name, int period) {
        this.setName(name);
        this.period = period;
    }

    public void watch(int byteSize) {
        this.increment.getAndAdd(byteSize);
    }

    @Override
    public void run() {
        for (; ; ) {
            try {
                TimeUnit.SECONDS.sleep(period);
                System.out.printf("%.2f kb/s\n", increment.doubleValue() / period / 1000);
                //reset
                increment.set(0);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}