timi-io/src/main/java/com/imyeyu/io/IOSpeedService.java

package com.imyeyu.io;


import com.imyeyu.java.TimiJava;
import com.imyeyu.java.bean.CallbackArg;
import com.imyeyu.utils.Calc;
import lombok.Getter;
import lombok.Setter;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;

/**
 * 字节速度计算，单例服务
 * <pre>
 *     // 开始计算（默认周期 1 秒）
 *     IOSpeedService.getInstance().start();
 *
 *     // 监听全局字节量
 *     IOSpeedService.getInstance().addBufferListener(d -> System.out.println(d));
 *
 *     // 创建字节统计节点
 *     IOSpeedService.item item = IOSpeedService.getInstance().createItem();
 *
 *     // 监听节点字节量
 *     item.addBufferListener(d -> System.out.println(d));
 *
 *     // 推送字节量
 *     item.push(1024);
 * </pre>
 *
 * @author 夜雨
 * @version 2021-11-30 12:43
 */
public class IOSpeedService {

	/** 单例对象 */
	private static IOSpeedService service;

	/** 加盐，用于在数据对齐的 IO 流中看起来像非对齐传输。此操作会使计算数据变得非真实，仅作为 UI 显示时可使用 */
	@Setter
	@Getter
	private long salt;
	private Timer timer;
	private double totalBufferOld;

	/** 总缓冲量 */
	@Getter
	private double totalBuffer;

	/** true 为运行中 */
	@Getter
	private boolean isRunning;

	private final Map<String, Item> items;
	private final List<CallbackArg<Double>> bufferListeners;

	private IOSpeedService() {
		items = new HashMap<>();
		bufferListeners = new ArrayList<>();
	}

	/** 启动服务，默认每秒计算一次 */
	public void start() {
		start(1000);
	}

	/**
	 * 启动服务
	 *
	 * @param interval 频率，单位：毫秒
	 */
	public void start(int interval) {
		if (timer != null) {
			shutdown();
		}
		timer = new Timer(true);
		timer.scheduleAtFixedRate(new TimerTask() {

			@Override
			public void run() {
				// 单项速度
				Item value;
				for (Map.Entry<String, Item> item : items.entrySet()) {
					value = item.getValue();
					synchronized (value.bufferListeners) {
						long _salt = 0;
						if (value.buffer != value.bufferOld) {
							_salt = Calc.random(-salt, +salt);
						}
						for (int i = 0; i < value.bufferListeners.size(); i++) {
							value.bufferListeners.get(i).handler(value.buffer - value.bufferOld + _salt);
						}
					}
					value.bufferOld = value.buffer;
				}
				// 全局速度
				synchronized (bufferListeners) {
					long _salt = 0;
					if (totalBuffer != totalBufferOld) {
						_salt = Calc.random(-salt, +salt);
					}
					for (int i = 0; i < bufferListeners.size(); i++) {
						bufferListeners.get(i).handler(totalBuffer - totalBufferOld + _salt);
					}
				}
				totalBufferOld = totalBuffer;
			}
		}, 0, interval);
		isRunning = true;
	}

	/**
	 * 创建统计节点
	 *
	 * @return 统计节点
	 */
	public Item createItem() {
		return createItem(null);
	}

	/**
	 * 创建统计节点
	 *
	 * @param customID 自定义 ID
	 * @return 统计节点
	 */
	public Item createItem(String customID) {
		Item item = new Item(customID);
		item.bufferListeners.add(d -> totalBuffer += d);
		items.put(item.id, item);
		return item;
	}

	/**
	 * 获取统计节点
	 *
	 * @param id 节点 ID
	 * @return 统计节点
	 */
	public Item item(String id) {
		return items.get(id);
	}

	/** 重置计算 */
	public void reset() {
		totalBufferOld = totalBuffer = 0;
	}

	/** 终止服务 */
	public void shutdown() {
		if (timer != null) {
			timer.cancel();
			timer.purge();
			timer = null;
		}
		isRunning = false;
	}

	/**
	 * 获取单例对象
	 *
	 * @return 单例对象
	 */
	public static synchronized IOSpeedService getInstance() {
		if (service == null) {
			service = new IOSpeedService();
		}
		return service;
	}

	/**
	 * 添加全局字节量监听
	 *
	 * @param bufferListener 全局字节量监听
	 */
	public void addBufferListener(CallbackArg<Double> bufferListener) {
		synchronized (bufferListeners) {
			bufferListeners.add(bufferListener);
		}
	}

	/**
	 * 移除全局字节量监听
	 *
	 * @param bufferListener 全局字节量监听
	 */
	public void removeBufferListener(CallbackArg<Double> bufferListener) {
		synchronized (bufferListeners) {
			bufferListeners.remove(bufferListener);
		}
	}

	/**
	 * 统计节点，节点之间互相不受影响
	 *
	 * @author 夜雨
	 * @version 2023-05-09 10:34
	 */
	public static final class Item {

		final List<CallbackArg<Double>> bufferListeners;

		String id;
		double buffer;
		double bufferOld;

		Item(String customID) {
			if (TimiJava.isEmpty(customID)) {
				id = UUID.randomUUID().toString();
			} else {
				id = customID;
			}
			buffer = 0;
			bufferListeners = new ArrayList<>();
		}

		/**
		 * 推送已处理字节量
		 *
		 * @param buffer 字节量
		 */
		public void push(double buffer) {
			this.buffer += buffer;
		}

		/** 重置计算 */
		public void reset() {
			bufferOld = buffer = 0;
		}

		/**
		 * 移除字节量监听
		 *
		 * @param bufferListener 全局字节量监听
		 */
		public void addBufferListener(CallbackArg<Double> bufferListener) {
			synchronized (bufferListeners) {
				bufferListeners.add(bufferListener);
			}
		}

		/**
		 * 移除字节量监听
		 *
		 * @param bufferListener 全局字节量监听
		 */
		public void removeBufferListener(CallbackArg<Double> bufferListener) {
			synchronized (bufferListeners) {
				bufferListeners.remove(bufferListener);
			}
		}
	}
}