增加窗口统计功能，增加 scala 部分实现

java-simple-dspool/JdbcConnection.java

@@ -0,0 +1,24 @@
+package org.apache.kyuubi.engine.jdbc.session;
+
+import java.sql.Connection;
+
+/**
+ * <pre>
+ *
+ * Created by zhenqin.
+ * User: zhenqin
+ * Date: 2026/2/11
+ * Time: 13:32
+ * Vendor: yiidata.com
+ *
+ * </pre>
+ *
+ * @author zhenqin
+ */
+public interface JdbcConnection extends Connection {
+
+    /**
+     * 直接关闭
+     */
+    public void directClose();
+}

java-simple-dspool/JdbcConnectionWrapper.java

@@ -0,0 +1,91 @@
+package org.apache.kyuubi.engine.jdbc.session;
+
+import org.apache.commons.dbcp2.DelegatingConnection;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Objects;
+
+/**
+ * JDBC 包装类
+ * <pre>
+ *
+ * Created by zhenqin.
+ * User: zhenqin
+ * Date: 2026/2/11
+ * Time: 10:18
+ * Vendor: yiidata.com
+ *
+ * </pre>
+ *
+ * @author zhenqin
+ */
+public class JdbcConnectionWrapper extends DelegatingConnection<Connection> implements JdbcConnection {
+
+    /**
+     * MySQL 等数据库支持的 connection Id
+     */
+    private String connectionId;
+
+    /**
+     * 强制关闭，并且废弃的连接
+     */
+    private boolean validFlag = true;
+
+    /**
+     * 直接的连接
+     */
+    final Connection realConnection;
+
+    /**
+     * 代理的连接和直接的连接
+     * @param connection
+     * @param realConnection
+     */
+    public JdbcConnectionWrapper(Connection connection, Connection realConnection) {
+        super(connection);
+        this.realConnection = realConnection;
+    }
+
+    public String getConnectionId() {
+        return connectionId;
+    }
+
+    public void setConnectionId(String connectionId) {
+        this.connectionId = connectionId;
+    }
+
+    public boolean isValidFlag() {
+        return validFlag;
+    }
+
+    public void setValidFlag(boolean validFlag) {
+        this.validFlag = validFlag;
+    }
+
+    @Override
+    public void directClose() {
+        try {
+            realConnection.close();
+        } catch (Exception ignore) {}
+    }
+
+    @Override
+    public void close() throws SQLException {
+        directClose();
+        super.close();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        JdbcConnectionWrapper that = (JdbcConnectionWrapper) o;
+        return Objects.equals(getDelegate(), that.getDelegate());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getDelegate());
+    }
+}

java-simple-dspool/JdbcStatementWrapper.java

@@ -0,0 +1,55 @@
+package org.apache.kyuubi.engine.jdbc.session;
+
+import org.apache.commons.dbcp2.DelegatingConnection;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.SQLException;
+import java.sql.Statement;
+
+/**
+ * JDBC JdbcStatementWrapper 包装类
+ * <pre>
+ *
+ * Created by zhenqin.
+ * User: zhenqin
+ * Date: 2026/2/11
+ * Time: 12:42
+ * Vendor: yiidata.com
+ *
+ * </pre>
+ *
+ * @author zhenqin
+ */
+public class JdbcStatementWrapper extends DelegatingStatement {
+
+    final Logger log = LoggerFactory.getLogger(SimpleDataSource.class);
+
+    /**
+     * 内置，主要返回连接使用
+     */
+    final DelegatingConnection<?> copyConnection;
+
+    public JdbcStatementWrapper(DelegatingConnection<?> connection, Statement statement) {
+        super(connection, statement);
+        this.copyConnection = connection;
+    }
+
+    /**
+     * 获取内部的连接，不检查
+     * @return
+     */
+    public DelegatingConnection getInternalConnection() {
+        return getConnectionInternal() == null ? copyConnection : getConnectionInternal();
+    }
+
+    @Override
+    public void setQueryTimeout(final int seconds) throws SQLException {
+        if(seconds <= 0) {
+            log.warn("invalid query timeout value: " + seconds);
+            return;
+        }
+        super.setQueryTimeout(seconds);
+    }
+}

java-simple-dspool/PooledDataSource.java

@@ -1,4 +1,4 @@
-package com.yiidata.dataops.common.bigdata;
+package org.apache.kyuubi.engine.jdbc.session;
 
 
 import java.io.Closeable;
@@ -8,7 +8,7 @@ import javax.sql.DataSource;
 
 /**
  *
- * Hive DAtaSource 实现
+ * JDBC DataSource 实现
  *
  *  @author zhaopx
  */

java-simple-dspool/SimpleDataSource.java

@@ -1,6 +1,9 @@
-package com.yiidata.dataops.common.bigdata;
+package org.apache.kyuubi.engine.jdbc.session;
 
-import lombok.extern.slf4j.Slf4j;
+import org.apache.kyuubi.config.KyuubiConf;
+import org.apache.kyuubi.engine.jdbc.connection.JdbcConnectionProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.PrintWriter;
@@ -12,64 +15,77 @@ import java.sql.SQLException;
 import java.sql.SQLFeatureNotSupportedException;
 import java.util.Date;
 import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
 /**
- *  
+ *  自实现的基于 JdbcConnectionProvider 的数据源 连接池
  * @author zhaopx
  */
 
-@Slf4j
-public class SimpleDataSource implements PooledDataSource {	 
+public class SimpleDataSource implements PooledDataSource {
 
-	private ConcurrentMap<Connection, Date> pool = new ConcurrentHashMap<Connection, Date>();
-	
+	final Logger log = LoggerFactory.getLogger(SimpleDataSource.class);
+
+	/**
+	 * 缓存的池子
+	 */
+	private final ConcurrentMap<JdbcConnectionWrapper, Date> pool = new ConcurrentHashMap<JdbcConnectionWrapper, Date>();
+
+	/**
+	 * 5 个连接，最大
+	 */
+	private int maxSize = 10;
+
+	/**
+	 * 最小连接数
+	 */
+	private int minSize = 1;
 
-	private int maxSize;
-	private int minSize;
-	private int waitTime;
+	/**
+	 * 连接等待时间
+	 */
+	private int waitTime = 30000;
 	
 	private Semaphore semaphore;
 
-	private final Properties connProps=new Properties();
+	private final Properties connProps = new Properties();
 
 
-	private HiveConnectionService connectionService;
+	private JdbcConnectionProvider jdbcConnectionProvider;
+	private KyuubiConf kyuubiConf;
 	
-	public SimpleDataSource(HiveConnectionService connectionService, Properties poolProperties) {
-		this.connectionService = connectionService;
-		connProps.putAll(poolProperties);
-
-		maxSize  = Integer.parseInt(poolProperties.getProperty("pool.max", "15"));
-		minSize  = Integer.parseInt(poolProperties.getProperty("pool.min", "3"));
-		waitTime = Integer.parseInt(poolProperties.getProperty("pool.waitTime", "5"));
-		initConnections(poolProperties);
+	public SimpleDataSource(JdbcConnectionProvider jdbcConnectionProvider, KyuubiConf kyuubiConf) {
+		this.jdbcConnectionProvider = jdbcConnectionProvider;
+		this.kyuubiConf = kyuubiConf;
+		this.maxSize = Math.max(Integer.parseInt(kyuubiConf.getOption("kyuubi.engine.jdbc.pool.maxSize").getOrElse(()-> "10")), 10);
+		this.minSize = Math.max(Integer.parseInt(kyuubiConf.getOption("kyuubi.engine.jdbc.pool.minSize").getOrElse(()-> "1")), 1);
+		this.waitTime = Math.max(Integer.parseInt(kyuubiConf.getOption("kyuubi.engine.jdbc.pool.maxWait").getOrElse(()-> "30000")), 1000);
+		initConnections();
 	}
 
-	private void initConnections(Properties poolProperties) {
+	private void initConnections() {
 		log.info("Initializing simple data source{ pool.max = " + maxSize + ", pool.min = " + minSize + "}");
 		semaphore = new Semaphore(maxSize, false);
 		if (minSize > 0 && minSize < maxSize) {
 			try {
 				// 尝试获得连接
-				Connection conn = getConnection();
-				if(conn!=null){
-					conn.close();
-				}
+				Connection conn = getRealConnection(null, null);
+				conn.close();
 			} catch (SQLException e) {
-				throw new IllegalArgumentException(e);
+				throw new RuntimeException(e);
 			}
 		}
 	}
 
 	public void close() throws IOException {
 		Exception ex = null;
-		for (Connection conn : pool.keySet()) {
+		for (JdbcConnectionWrapper conn : pool.keySet()) {
 			try {
-				conn.close();
+				conn.directClose();
 			} catch (Exception e) { ex = e; }
 		}
 		pool.clear();
@@ -78,15 +94,23 @@ public class SimpleDataSource implements PooledDataSource {
 		}
 		log.info("closed data source{ pool.max = " + maxSize + ", pool.min = " + minSize + "}");
 	}
-	
-	private void closeConnection(Connection realConnection) throws SQLException {
+
+	/**
+	 * 关闭连接，这里是软关闭
+	 * @param realConnection
+	 * @throws SQLException
+	 */
+	private void closeConnection(Connection realConnection, Connection proxyConnection) throws SQLException {
 		synchronized (pool) {
-			if (pool.size() <= maxSize) {
-				pool.put(realConnection, new Date());
+			if (pool.size() <= maxSize && realConnection instanceof JdbcConnectionWrapper && ((JdbcConnectionWrapper)realConnection).isValidFlag()) {
+				// 正常的连接不关闭，放到池中
+				pool.put((JdbcConnectionWrapper)realConnection, new Date());
+				return;
+			} else if(pool.size() <= maxSize && !(realConnection instanceof JdbcConnectionWrapper)) {
+				pool.put(new JdbcConnectionWrapper(proxyConnection, realConnection), new Date());
 				return;
 			}
 		}
-			
 		try {
 			realConnection.close();
 		} finally {
@@ -94,6 +118,35 @@ public class SimpleDataSource implements PooledDataSource {
 		}
 	}
 
+	/**
+	 * 关闭连接，这里是软关闭
+	 * @param realConnection
+	 * @throws SQLException
+	 */
+	public void closeConnectionAndRemove(Connection realConnection) throws SQLException {
+		if(realConnection == null) {
+			return;
+		}
+		synchronized (pool) {
+			// 从缓存移除
+			if(realConnection instanceof JdbcConnectionWrapper) {
+				pool.remove((JdbcConnectionWrapper) realConnection);
+			} else {
+				pool.remove(realConnection);
+			}
+		}
+		try {
+			if(realConnection instanceof JdbcConnectionWrapper) {
+				((JdbcConnectionWrapper)realConnection).directClose();
+			} else {
+				realConnection.close();
+			}
+		} catch (Exception ignore) {
+		} finally {
+			semaphore.release();
+		}
+	}
+
 	public Connection getConnection() throws SQLException {
 		return getConnection(null, null);
 	}
@@ -101,54 +154,60 @@ public class SimpleDataSource implements PooledDataSource {
 	public Connection getConnection(String username, String password) throws SQLException {		 
 		synchronized (pool) {
 			if (!pool.isEmpty()) {
-				Connection realConn = pool.keySet().iterator().next();
+				JdbcConnectionWrapper realConn = pool.keySet().iterator().next();
 				pool.remove(realConn);
+				if(realConn.isValidFlag()) {
+					return realConn;
+				}
 
 				// hive jdbc 不支持设置  AutoCommit
 				//realConn.setAutoCommit(true);
-
 				return getProxyConnection(realConn);
 			}
 		}
-		 	
+
 		try {
-			if(semaphore.tryAcquire(waitTime,TimeUnit.SECONDS)) {
+			if (semaphore.tryAcquire(waitTime, TimeUnit.MILLISECONDS)) {
 				return getProxyConnection(getRealConnection(username, password));
-			}else {
-				throw new IllegalArgumentException("Connection pool is full: "+maxSize);
+			} else {
+				throw new RuntimeException("Connection pool is full: " + maxSize);
 			}
-		}catch(SQLException e) {
+		} catch (SQLException e) {
 			semaphore.release();
 			throw e;
 		} catch (InterruptedException e) {
-			log.error(e.getMessage(),e);
-			Thread.currentThread().interrupt();
-			return null;
+			throw new RuntimeException(e);
 		}
 	}
 	
-	private Connection getProxyConnection(final Connection realConnection) {		
+	private Connection getProxyConnection(final Connection realConnection) {
 		InvocationHandler handler = new InvocationHandler() {
 			public Object invoke(Object proxy, Method method, Object[] params) throws Exception {
 				Object ret = null;
 				if ("close".equals(method.getName())) {
-					closeConnection(realConnection);
-				}else if ("unwrap".equals(method.getName())) {
-					ret=realConnection;
+					closeConnection(realConnection, (Connection)proxy);
+				} else if ("directClose".equals(method.getName())) {
+					// 实际的关闭
+					try {
+						realConnection.close();
+					} catch (Exception ignore) {}
+					ret = Void.TYPE.newInstance();
+				} else if ("unwrap".equals(method.getName())) {
+					ret = realConnection;
 				} else {
 					ret = method.invoke(realConnection, params);
 				}
 				return ret;
 			}
 		};
-		return (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), new Class[] { Connection.class }, handler);
+		return new JdbcConnectionWrapper((JdbcConnection) Proxy.newProxyInstance(JdbcConnection.class.getClassLoader(), new Class[] { JdbcConnection.class }, handler), realConnection);
 	}
 
-	
 
-	protected Connection getRealConnection(String username, String password) throws SQLException {
+
+	public Connection getRealConnection(String username, String password) throws SQLException {
 		try {
-			return connectionService.getConnection();
+			return jdbcConnectionProvider.getConnection(kyuubiConf);
 		} catch (Exception e) {
 			throw new SQLException(e);
 		}
@@ -163,11 +222,10 @@ public class SimpleDataSource implements PooledDataSource {
 	}
 
 	public void setLogWriter(PrintWriter out) throws SQLException {
-		throw new UnsupportedOperationException();
 	}
 
 	public void setLoginTimeout(int seconds) throws SQLException {
-		throw new UnsupportedOperationException();
+
 	}
 
 	public int getLoginTimeout() throws SQLException {
@@ -206,4 +264,12 @@ public class SimpleDataSource implements PooledDataSource {
 	public Properties getConnProps() {
 		return connProps;
 	}
+
+	/**
+	 * 返回内置的绑定
+	 * @return
+	 */
+	public Set<JdbcConnectionWrapper> getPoolSet() {
+		return pool.keySet();
+	}
 }

scala-classic/Redised.scala

@@ -0,0 +1,160 @@
+package org.apache.kyuubi.events.handler
+
+import org.apache.commons.lang3.StringUtils
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig
+import org.apache.kyuubi.Logging
+import redis.clients.jedis.{Jedis, JedisPool, JedisPubSub}
+
+import java.io.{Closeable, IOException}
+import java.time.Duration
+import java.util
+import java.util.Properties
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+/**
+ * 易点天下
+ *
+ * <pre>
+ *
+ * Created by zhenqin.
+ * User: zhenqin
+ * Date: 2026/1/7
+ * Time: 12:53
+ * Vendor: yiidata.com
+ *
+ * </pre>
+ *
+ * @author zhenqin
+ */
+class Redised(props: Properties) extends Closeable with Logging {
+
+  /**
+   * Redis 的 Java Client
+   */
+  private val redisPool: JedisPool = initJedis(props)
+
+  /**
+   * Redis DB
+   */
+  val db = props.getProperty("db", "0").toInt;
+
+  private def initJedis(props: Properties): JedisPool = {
+    val poolConfig = new GenericObjectPoolConfig[Jedis]();
+    // 设置连接池的最大空闲连接数等其他属性...// 设置连接池的最大空闲连接数等其他属性...
+    poolConfig.setMaxTotal(props.getProperty("maxTotal", "100").toInt)
+    poolConfig.setMaxIdle(Math.max(poolConfig.getMaxTotal, 5))
+    poolConfig.setMinIdle(3)
+    poolConfig.setTestOnBorrow(false)
+    poolConfig.setTestWhileIdle(true)
+    poolConfig.setTestOnCreate(false)
+    poolConfig.setMinEvictableIdleTime(Duration.ofSeconds(95))
+    logger.info("connect redis: " + props.getProperty("host", "localhost") + ":" + props.getProperty("port", "6379"))
+    // 判断密码，如果有密码
+    if(StringUtils.isBlank(props.getProperty("password"))) {
+      return new JedisPool(poolConfig, props.getProperty("host", "localhost"), props.getProperty("port", "6379").toInt,
+      props.getProperty("timeout", "3000").toInt)
+    }
+    new JedisPool(poolConfig, props.getProperty("host", "localhost"), props.getProperty("port", "6379").toInt,
+      props.getProperty("timeout", "3000").toInt, props.getProperty("password"))
+  }
+
+  @throws[IOException]
+  def zsetadd(sessionId: String, score: Double, statementId: String): Unit = {
+    val jedis = redisPool.getResource
+    try jedis.zadd("z:" + sessionId, score, statementId)
+    finally if (jedis != null) jedis.close()
+  }
+
+  @throws[IOException]
+  def zcount(sessionId: String): Long = {
+    val jedis = redisPool.getResource
+    try jedis.zcard("z:" + sessionId)
+    finally if (jedis != null) jedis.close()
+  }
+
+  @throws[IOException]
+  def zsettopn(sessionId: String, limit: Int): mutable.Set[String] = {
+    val jedis = redisPool.getResource
+    try {
+      jedis.zrevrange("z:" + sessionId, 0, limit).asScala;
+    }
+    finally if (jedis != null) jedis.close()
+  }
+
+  @throws[IOException]
+  def mput(sessionId: String, statementId: String, json: String): Unit = {
+    val jedis = redisPool.getResource
+    try jedis.hset(sessionId, statementId, json)
+    finally if (jedis != null) jedis.close()
+  }
+
+  /**
+   * 获取单个信息
+   * @param sessionId
+   * @param statementId
+   * @throws
+   * @return
+   */
+  @throws[IOException]
+  def mget(sessionId: String, statementId: String): String = {
+    val jedis = redisPool.getResource
+    try {
+      jedis.hget(sessionId, statementId);
+    }
+    finally if (jedis != null) jedis.close()
+  }
+
+  @throws[IOException]
+  def mget(sessionId: String, statementIds: Array[String]): Map[String, String] = {
+    val jedis = redisPool.getResource
+    val r = scala.collection.mutable.Map[String, String]()
+    try {
+      for(id <- statementIds) {
+        r += (id -> jedis.hget(sessionId, id))
+      }
+      return r.toMap[String, String]
+    }
+    finally if (jedis != null) jedis.close()
+  }
+
+  @throws[IOException]
+  def mgetAll(sessionId: String): util.Map[String, String] = {
+    val jedis = redisPool.getResource
+    try jedis.hgetAll(sessionId)
+    finally if (jedis != null) jedis.close()
+  }
+
+  @throws[IOException]
+  def publish(topic: String, json: String): Unit = {
+    val jedis = redisPool.getResource
+    try jedis.publish(topic, json)
+    finally if (jedis != null) jedis.close()
+  }
+
+  @throws[IOException]
+  def subscribe(topic: String, jedisPubSub: JedisPubSub): Unit = {
+    val jedis = redisPool.getResource
+    try jedis.subscribe(jedisPubSub, topic)
+    finally if (jedis != null) jedis.close()
+  }
+
+  def clear(): Unit = {
+    val jedis = redisPool.getResource
+    try {
+      jedis.select(db)
+      jedis.flushDB
+    } finally if (jedis != null) jedis.close()
+  }
+
+  override def close(): Unit = {
+    redisPool.close()
+  }
+
+  def shutdown(): Unit = {
+    try close()
+    catch {
+      case e: IOException => logger.error("close redis connection error!", e)
+    }
+  }
+}

slidingwindow/TimestampSlidingWindowCounter.java

@@ -0,0 +1,407 @@
+package org.apache.kyuubi.util;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * <pre>
+ *
+ * Created by zhenqin.
+ * User: zhenqin
+ * Date: 2026/3/13
+ * Time: 13:57
+ * Vendor: yiidata.com
+ *
+ * </pre>
+ *
+ * @author zhenqin
+ */
+public class TimestampSlidingWindowCounter {
+
+    // 存储每个key的事件时间戳队列
+    private final ConcurrentHashMap<String, Window> windows = new ConcurrentHashMap<>();
+
+    // 窗口大小（毫秒）
+    private final long windowSizeMs;
+
+    // 是否自动清理过期数据
+    private final boolean autoCleanup;
+
+    // 清理线程
+    private Timer cleanupTimer;
+
+    public TimestampSlidingWindowCounter(long windowSizeMs, boolean autoCleanup) {
+        this.windowSizeMs = windowSizeMs;
+        this.autoCleanup = autoCleanup;
+
+        if (autoCleanup) {
+            startCleanupTask();
+        }
+    }
+
+    /**
+     * 创建1分钟窗口计数器
+     */
+    public static TimestampSlidingWindowCounter createMinuteWindow() {
+        return new TimestampSlidingWindowCounter(Duration.ofMinutes(1).toMillis(), true);
+    }
+
+
+    /**
+     * 创建5分钟窗口计数器
+     */
+    @Deprecated
+    public static TimestampSlidingWindowCounter create5MinuteWindow() {
+        return new TimestampSlidingWindowCounter(Duration.ofMinutes(5).toMillis(), true);
+    }
+
+    /**
+     * 创建1小时窗口计数器
+     */
+    public static TimestampSlidingWindowCounter createHourWindow() {
+        return new TimestampSlidingWindowCounter(Duration.ofMinutes(60).toMillis(), true);
+    }
+
+    /**
+     * 记录一个事件
+     */
+    public long addEvent(String key) {
+        return addEvent(key, System.currentTimeMillis());
+    }
+
+    /**
+     * 记录一个指定时间的事件
+     */
+    public long addEvent(String key, long timestamp) {
+        Window window = windows.computeIfAbsent(key, k -> new Window(windowSizeMs));
+        return window.add(timestamp);
+    }
+
+    /**
+     * 记录一个指定时间的事件, 并添加 count 值
+     */
+    public long addEventCount(String key, long count) {
+        return addEventCount(key, System.currentTimeMillis(), count);
+    }
+
+    /**
+     * 记录一个指定时间的事件, 并添加 count 值
+     */
+    public long addEventCount(String key, long timestamp, long count) {
+        Window window = windows.computeIfAbsent(key, k -> new Window(windowSizeMs));
+        return window.add(timestamp, count);
+    }
+
+    /**
+     * 批量添加事件
+     */
+    public Map<String, Long> addEvents(Map<String, List<Long>> events) {
+        Map<String, Long> results = new HashMap<>();
+        events.forEach((key, timestamps) -> {
+            long count = 0;
+            for (long ts : timestamps) {
+                count = addEvent(key, ts);
+            }
+            results.put(key, count);
+        });
+        return results;
+    }
+
+    /**
+     * 获取最后放入的时间
+     * @param key
+     * @return
+     */
+    public long getLastTime(String key) {
+        Window window = windows.get(key);
+        return window != null ? window.getLastTime() : 0;
+    }
+
+    /**
+     * 获取当前窗口内的事件数量
+     */
+    public long getCount(String key) {
+        return getCount(key, System.currentTimeMillis());
+    }
+
+    /**
+     * 获取指定时间点窗口内的事件数量
+     */
+    public long getCount(String key, long timestamp) {
+        Window window = windows.get(key);
+        return window != null ? window.getCount(timestamp) : 0;
+    }
+
+    /**
+     * 获取指定时间点窗口内的计数器数量
+     */
+    public long getTotal(String key) {
+        Window window = windows.get(key);
+        return window != null ? window.getTotal(System.currentTimeMillis()) : 0;
+    }
+
+    /**
+     * 获取指定时间点窗口内的计数器数量
+     */
+    public long getTotal(String key, long timestamp) {
+        Window window = windows.get(key);
+        return window != null ? window.getTotal(timestamp) : 0;
+    }
+
+    /**
+     * 获取所有key的计数
+     */
+    public Map<String, Long> getAllCounts() {
+        Map<String, Long> counts = new HashMap<>();
+        long now = System.currentTimeMillis();
+        windows.forEach((key, window) -> {
+            counts.put(key, window.getCount(now));
+        });
+        return counts;
+    }
+
+    /**
+     * 判断是否超过阈值
+     */
+    public boolean isExceedLimit(String key, long threshold) {
+        return getCount(key) > threshold;
+    }
+
+    /**
+     * 判断是否超过阈值（带自定义时间）
+     */
+    public boolean isExceedLimit(String key, long threshold, long timestamp) {
+        return getCount(key, timestamp) > threshold;
+    }
+
+
+    /**
+     * 获取窗口内的事件时间戳列表
+     */
+    public List<Long> getEventTimestamps(String key) {
+        Window window = windows.get(key);
+        return window != null ? window.getTimestamps() : Collections.emptyList();
+    }
+
+    /**
+     * 清空指定key的数据
+     */
+    public void clear(String key) {
+        windows.remove(key);
+    }
+
+    /**
+     * 清空所有数据
+     */
+    public void clearAll() {
+        windows.clear();
+    }
+
+    /**
+     * 获取活跃的key数量
+     */
+    public int getActiveKeyCount() {
+        return windows.size();
+    }
+
+    /**
+     * 获取窗口大小
+     */
+    public long getWindowSizeMs() {
+        return windowSizeMs;
+    }
+
+    /**
+     * 启动清理任务
+     */
+    private void startCleanupTask() {
+        cleanupTimer = new Timer("SlidingWindow-Cleanup", true);
+        cleanupTimer.scheduleAtFixedRate(new TimerTask() {
+            @Override
+            public void run() {
+                try {
+                    cleanup();
+                } catch (Exception e) {
+                    // 忽略清理异常
+                }
+            }
+        }, windowSizeMs / 2, windowSizeMs / 2);
+    }
+
+    /**
+     * 手动清理过期数据
+     */
+    public void cleanup() {
+        long now = System.currentTimeMillis();
+        windows.entrySet().removeIf(entry -> {
+            Window window = entry.getValue();
+            return window.isEmpty(now);
+        });
+    }
+
+    /**
+     * 关闭清理任务
+     */
+    public void shutdown() {
+        if (cleanupTimer != null) {
+            cleanupTimer.cancel();
+        }
+    }
+
+
+    /**
+     * 内部窗口类
+     */
+    private static class Window {
+        // 使用双端队列存储时间戳
+        private final Deque<Long> timestamps = new LinkedBlockingDeque<>();
+
+        /** 每隔时间戳的统计值 */
+        private final Deque<Long> statistics = new LinkedBlockingDeque<>();
+
+        // 窗口大小
+        private final long windowSizeMs;
+
+        // 锁（保证线程安全）
+        private final ReentrantLock lock = new ReentrantLock();
+
+        // 缓存当前计数（优化性能）
+        private volatile long cachedCount = 0;
+        private volatile long lastCleanupTime = System.currentTimeMillis();
+
+        public Window(long windowSizeMs) {
+            this.windowSizeMs = windowSizeMs;
+        }
+
+        /**
+         * 添加事件
+         */
+        public long add(long timestamp) {
+            lock.lock();
+            try {
+                // 移除过期的数据
+                removeExpired(timestamp);
+
+                // 添加新事件
+                timestamps.addLast(timestamp);
+                statistics.addLast(0L);
+                cachedCount = timestamps.size();
+                lastCleanupTime = timestamp;
+
+                return cachedCount;
+            } finally {
+                lock.unlock();
+            }
+        }
+
+        /**
+         * 添加事件
+         */
+        public long add(long timestamp, long value) {
+            lock.lock();
+            try {
+                // 移除过期的数据
+                removeExpired(timestamp);
+
+                // 添加新事件
+                timestamps.addLast(timestamp);
+                statistics.addLast(value);
+                cachedCount = timestamps.size();
+                lastCleanupTime = timestamp;
+
+                return cachedCount;
+            } finally {
+                lock.unlock();
+            }
+        }
+
+
+        /**
+         * 获取当前计数
+         */
+        public long getCount(long timestamp) {
+            // 如果最后清理时间太旧，主动清理一次
+            if (timestamp - lastCleanupTime > windowSizeMs / 4) {
+                lock.lock();
+                try {
+                    removeExpired(timestamp);
+                } finally {
+                    lock.unlock();
+                }
+            }
+            return cachedCount;
+        }
+
+        /**
+         * 获取当前计数总和
+         */
+        public long getTotal(long timestamp) {
+            // 如果最后清理时间太旧，主动清理一次
+            if (timestamp - lastCleanupTime > windowSizeMs / 4) {
+                lock.lock();
+                try {
+                    removeExpired(timestamp);
+                } finally {
+                    lock.unlock();
+                }
+            }
+            return statistics.stream().mapToLong(it->it).sum();
+        }
+
+        /**
+         * 获取最后一次计数的拉入时间，可能返回 null
+         */
+        public long getLastTime() {
+            return Optional.ofNullable(timestamps.peekLast()).orElse(0L);
+        }
+
+        /**
+         * 获取所有时间戳
+         */
+        public List<Long> getTimestamps() {
+            lock.lock();
+            try {
+                return new ArrayList<>(timestamps);
+            } finally {
+                lock.unlock();
+            }
+        }
+
+        /**
+         * 是否为空
+         */
+        public boolean isEmpty(long timestamp) {
+            lock.lock();
+            try {
+                removeExpired(timestamp);
+                return timestamps.isEmpty();
+            } finally {
+                lock.unlock();
+            }
+        }
+
+        /**
+         * 移除过期数据
+         */
+        private void removeExpired(long now) {
+            long cutoff = now - windowSizeMs;
+            while (!timestamps.isEmpty() && timestamps.peekFirst() < cutoff) {
+                timestamps.removeFirst();
+                statistics.removeFirst();
+            }
+            cachedCount = timestamps.size();
+            lastCleanupTime = now;
+        }
+    }
+}