registry ..
diff --git a/README.md b/README.md index cc8612e..03110cd 100644 --- a/README.md +++ b/README.md
@@ -81,7 +81,7 @@ When the user is managing changes in the background, - Soul dynamically updates the cache by listening to the zookeeper node, websocket push,http longPull. + Soul dynamically updates the cache by listening to the zookeeper ttlCacheTimerTask, websocket push,http longPull. 
diff --git a/soul-common/src/main/java/org/dromara/soul/common/cache/HashedWheelTimer.java b/soul-common/src/main/java/org/dromara/soul/common/cache/HashedWheelTimer.java deleted file mode 100644 index 20a23bb..0000000 --- a/soul-common/src/main/java/org/dromara/soul/common/cache/HashedWheelTimer.java +++ /dev/null
@@ -1,657 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements.See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License.You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.dromara.soul.common.cache; - -import java.util.Collections; -import java.util.HashSet; -import java.util.Queue; -import java.util.Set; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; -import org.dromara.soul.common.utils.OsUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Loop time implementation. - * io.netty.util.HashedWheelTimer - * - * @author sixh - */ -public class HashedWheelTimer implements Timer { - - private static final Logger LOGGER = LoggerFactory.getLogger(HashedWheelTimer.class); - - /** - * 状态更新. - */ - private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER; - - - /** - * 任务执行状态. - */ - private static final int WORKER_STATE_INIT = 0; - - /** - * 任务已启动. - */ - private static final int WORKER_STATE_STARTED = 1; - - /** - * 任务已结束. - */ - private static final int WORKER_STATE_SHUTDOWN = 2; - - /** - * 时间分片槽，默认512. - */ - private final HashedWheelTimer.HashedWheelBucket[] wheel; - - /** - * 执行时间分片槽的次数； {@link #wheel}长度 -1. - */ - private final int mask; - - /** - * 时间分片执行时间. - */ - private final long tickDuration; - - /** - * 执行线程. - */ - private final Thread workerThread; - - /** - * 任务执行实现. - */ - private final HashedWheelTimer.Worker worker = new HashedWheelTimer.Worker(); - - /** - * 开始时间. - */ - private volatile long startTime; - - /** - * 开始时间初始化. - */ - private final CountDownLatch startTimeInitialized = new CountDownLatch(1); - - - /** - * 超时器. - */ - private final Queue<HashedWheelTimeout> timeouts = new ConcurrentLinkedQueue<>(); - - /** - * 取消超时器. - */ - private final Queue<HashedWheelTimeout> cancelledTimeouts = new ConcurrentLinkedQueue<>(); - - static { - WORKER_STATE_UPDATER = AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState"); - } - - /** - * 0 - init, 1 - started, 2 - shut down. - */ - @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization"}) - private volatile int workerState = WORKER_STATE_INIT; - - /** - * 初始化一个环形时间执行器. - * - * @param threadFactory 线程初始化工厂； - */ - HashedWheelTimer(ThreadFactory threadFactory) { - this(threadFactory, 100, TimeUnit.MILLISECONDS, 512); - } - - /** - * 初始化一个环形时间执行器. - * - * @param threadFactory 线程初始化工厂； - * @param tickDuration 持续执行时间； - * @param unit 时间单位； - * @param ticksPerWheel 时间分片数； - */ - private HashedWheelTimer(ThreadFactory threadFactory, - long tickDuration, TimeUnit unit, int ticksPerWheel) { - /* - * 创建时间分片区； - */ - this.wheel = createWheel(ticksPerWheel); - //执行时间分片槽的次数； - mask = wheel.length - 1; - //把所有时间转为纳秒进方便操作； - this.tickDuration = unit.toNanos(tickDuration); - // Prevent overflow. - if (this.tickDuration >= Long.MAX_VALUE / wheel.length) { - throw new IllegalArgumentException(String.format( - "tickDuration: %d (expected: 0 < tickDuration in nanos < %d", - tickDuration, Long.MAX_VALUE / wheel.length)); - } - workerThread = threadFactory.newThread(worker); - } - - /** - * 启动任务. - */ - private void start() { - switch (WORKER_STATE_UPDATER.get(this)) { - //更改线程状态；如果当前的状态是初始，则改成启动中； - case WORKER_STATE_INIT: - if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) { - //启动线程； - workerThread.start(); - } - break; - case WORKER_STATE_STARTED: - break; - case WORKER_STATE_SHUTDOWN: - throw new IllegalStateException("cannot be started once stopped"); - default: - throw new Error("Invalid WorkerState"); - } - - // 等待初始化的Worker线程的startTime实始化； - while (startTime == 0) { - try { - startTimeInitialized.await(); - } catch (InterruptedException ignore) { - // Ignore - it will be ready very soon. - } - } - } - - private static HashedWheelTimer.HashedWheelBucket[] createWheel(int ticksPerWheel) { - int newTicksPerWheel = ticksPerWheel; - if (newTicksPerWheel <= 0) { - throw new IllegalArgumentException( - "ticksPerWheel must be greater than 0: " + newTicksPerWheel); - } - if (newTicksPerWheel > 1073741824) { - throw new IllegalArgumentException( - "ticksPerWheel may not be greater than 2^30: " + newTicksPerWheel); - } - - newTicksPerWheel = normalizeTicksPerWheel(newTicksPerWheel); - HashedWheelTimer.HashedWheelBucket[] wheel = new HashedWheelTimer.HashedWheelBucket[newTicksPerWheel]; - for (int i = 0; i < wheel.length; i++) { - wheel[i] = new HashedWheelTimer.HashedWheelBucket(); - } - return wheel; - } - - private static int normalizeTicksPerWheel(int ticksPerWheel) { - int normalizedTicksPerWheel = 1; - while (normalizedTicksPerWheel < ticksPerWheel) { - normalizedTicksPerWheel <<= 1; - } - return normalizedTicksPerWheel; - } - - @Override - public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) { - if (task == null) { - throw new NullPointerException("task"); - } - if (unit == null) { - throw new NullPointerException("unit"); - } - start(); - /* - * 设置一下超时作务到列队中；计算出下一次的超时时间； - */ - long deadline = System.nanoTime() + unit.toNanos(delay) - startTime; - HashedWheelTimer.HashedWheelTimeout timeout = new HashedWheelTimer.HashedWheelTimeout(this, task, deadline); - timeouts.add(timeout); - return timeout; - } - - @Override - public Set<Timeout> stop() { - if (Thread.currentThread() == workerThread) { - throw new IllegalStateException( - HashedWheelTimer.class.getSimpleName() + - ".stop() cannot be called from " + - TimerTask.class.getSimpleName()); - } - - if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) { - // workerState can be 0 or 2 at this moment - let it always be 2. - WORKER_STATE_UPDATER.set(this, WORKER_STATE_SHUTDOWN); - return Collections.emptySet(); - } - - boolean interrupted = false; - while (workerThread.isAlive()) { - workerThread.interrupt(); - try { - workerThread.join(100); - } catch (InterruptedException ignored) { - interrupted = true; - } - } - - if (interrupted) { - Thread.currentThread().interrupt(); - } - - return worker.unprocessedTimeouts(); - } - - private final class Worker implements Runnable { - private final Set<Timeout> unprocessedTimeouts = new HashSet<>(); - - /** - * 到达一个槽就是++. - */ - private long tick; - - @Override - public void run() { - //初始化启动开始时间； - startTime = System.nanoTime(); - if (startTime == 0) { - // 因为startTime使用0来判断是否进行初始化，所以要保证他的初始化不为0; - startTime = 1; - } - //结束执行start()方法； - startTimeInitialized.countDown(); - - do { - //计算下一个到达的时间片；如果没有到达则需要等待； - final long deadline = waitForNextTick(); - if (deadline > 0) { - int idx = (int) (tick & mask); - processCancelledTasks(); - //取出一个时间分片槽 - HashedWheelTimer.HashedWheelBucket bucket = - wheel[idx]; - //转换时间分片槽。并将数据放入到时间分片槽上； - transferTimeoutsToBuckets(); - bucket.expireTimeouts(deadline); - tick++; - } - } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED); - - // 如果调用Stop方法，就会填充未超时的Task将他返回给用户； - for (HashedWheelTimer.HashedWheelBucket bucket : wheel) { - bucket.clearTimeouts(unprocessedTimeouts); - } - for (; ; ) { - HashedWheelTimer.HashedWheelTimeout timeout = timeouts.poll(); - if (timeout == null) { - break; - } - if (!timeout.isCancelled()) { - unprocessedTimeouts.add(timeout); - } - } - processCancelledTasks(); - } - - private void transferTimeoutsToBuckets() { - // transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just - // adds new timeouts in a loop. - for (int i = 0; i < 100000; i++) { - //获取一个作务； - HashedWheelTimer.HashedWheelTimeout timeout = timeouts.poll(); - if (timeout == null) { - // all processed - break; - } - if (timeout.state() == HashedWheelTimer.HashedWheelTimeout.ST_CANCELLED) { - // Was cancelled in the meantime. - continue; - } - //需要轮转多少次，就可以到达时间分片上，（超时时间/任务分片轮转时间） - long calculated = timeout.deadline / tickDuration; - //计算还有剩余多少轮没有执行； - timeout.remainingRounds = (calculated - tick) / wheel.length; - // Ensure we don't schedule for past. - final long ticks = Math.max(calculated, tick); - //算出任务应该插入的时间分片槽位置； - int stopIndex = (int) (ticks & mask); - HashedWheelTimer.HashedWheelBucket bucket = wheel[stopIndex]; - bucket.addTimeout(timeout); - } - } - - /** - * 流程任务取消；从取消队列中获取任务。并删除. - */ - private void processCancelledTasks() { - for (; ; ) { - HashedWheelTimer.HashedWheelTimeout timeout = cancelledTimeouts.poll(); - if (timeout == null) { - // all processed - break; - } - try { - timeout.remove(); - } catch (Throwable t) { - if (LOGGER.isWarnEnabled()) { - LOGGER.warn("An exception was thrown while process a cancellation task", t); - } - } - } - } - - /** - * calculate goal nanoTime from startTime and current tick number, - * then wait until that goal has been reached. - * - * @return Long.MIN_VALUE if received a shutdown request, - * current time otherwise (with Long.MIN_VALUE changed by +1). - */ - private long waitForNextTick() { - //计算时间片；例：1000000*(1+1) - long deadline = tickDuration * (tick + 1); - - for (; ; ) { - //得到当前已经耗时； - final long currentTime = System.nanoTime() - startTime; - //得到休眠的时间毫秒数； - long sleepTimeMs = (deadline - currentTime + 999999) / 1000000; - - if (sleepTimeMs <= 0) { - if (currentTime == Long.MIN_VALUE) { - return -Long.MAX_VALUE; - } else { - return currentTime; - } - } - - // Check if we run on windows, as if thats the case we will need - // to round the sleepTime as workaround for a bug that only affect - // the JVM if it runs on windows. - // - // See https://github.com/netty/netty/issues/356 - if (OsUtils.isWindows()) { - sleepTimeMs = sleepTimeMs / 10 * 10; - } - - try { - TimeUnit.MILLISECONDS.sleep(sleepTimeMs); - } catch (InterruptedException ignored) { - if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) { - return Long.MIN_VALUE; - } - } - } - } - - Set<Timeout> unprocessedTimeouts() { - return Collections.unmodifiableSet(unprocessedTimeouts); - } - } - - /** - * 超时任务对象. - */ - private static final class HashedWheelTimeout implements Timeout { - - private static final int ST_INIT = 0; - - private static final int ST_CANCELLED = 1; - - private static final int ST_EXPIRED = 2; - - private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER; - - static { - AtomicIntegerFieldUpdater<HashedWheelTimeout> updater = - AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.HashedWheelTimeout.class, "state"); - STATE_UPDATER = updater; - } - - private final HashedWheelTimer timer; - - private final TimerTask task; - - private final long deadline; - - @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization"}) - private volatile int state = ST_INIT; - - /** - * remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the - * HashedWheelTimeout will be added to the correct HashedWheelBucket. - */ - private long remainingRounds; - - /** - * This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list. - * As only the workerThread will act on it there is no need for synchronization / volatile. - */ - private HashedWheelTimer.HashedWheelTimeout next; - - private HashedWheelTimer.HashedWheelTimeout prev; - - /** - * The bucket to which the timeout was added - */ - private HashedWheelTimer.HashedWheelBucket bucket; - - HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) { - this.timer = timer; - this.task = task; - this.deadline = deadline; - } - - @Override - public Timer timer() { - return timer; - } - - @Override - public TimerTask task() { - return task; - } - - @Override - public long deadline() { - return deadline; - } - - @Override - public boolean cancel() { - // only update the state it will be removed from HashedWheelBucket on next tick. - if (!compareAndSetState(ST_INIT, ST_CANCELLED)) { - return false; - } - // If a task should be canceled we put this to another queue which will be processed on each tick. - // So this means that we will have a GC latency of max. 1 tick duration which is good enough. This way - // we can make again use of our MpscLinkedQueue and so minimize the locking / overhead as much as possible. - timer.cancelledTimeouts.add(this); - return true; - } - - void remove() { - HashedWheelTimer.HashedWheelBucket bucket = this.bucket; - if (bucket != null) { - bucket.remove(this); - } - } - - boolean compareAndSetState(int expected, int state) { - return STATE_UPDATER.compareAndSet(this, expected, state); - } - - int state() { - return state; - } - - @Override - public boolean isCancelled() { - return state() == ST_CANCELLED; - } - - @Override - public boolean isExpired() { - return state() == ST_EXPIRED; - } - - void expire() { - if (!compareAndSetState(ST_INIT, ST_EXPIRED)) { - return; - } - - try { - task.run(this); - } catch (Throwable t) { - if (LOGGER.isWarnEnabled()) { - LOGGER.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t); - } - } - } - - @Override - public boolean isDefault() { - return false; - } - } - - /** - * 每个分片槽封装对象， - * 这里使用的是双向队列. - */ - private static final class HashedWheelBucket { - private HashedWheelTimeout head; - - private HashedWheelTimeout tail; - - /** - * Add {@link HashedWheelTimeout} to this bucket. - */ - void addTimeout(HashedWheelTimeout timeout) { - assert timeout.bucket == null; - timeout.bucket = this; - if (head == null) { - head = tail = timeout; - } else { - tail.next = timeout; - timeout.prev = tail; - tail = timeout; - } - } - - /** - * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}. - */ - void expireTimeouts(long deadline) { - HashedWheelTimeout timeout = head; - // process all timeouts - while (timeout != null) { - boolean remove = false; - if (timeout.remainingRounds <= 0) { - if (timeout.deadline <= deadline) { - timeout.expire(); - } else { - // The timeout was placed into a wrong slot. This should never happen. - throw new IllegalStateException(String.format( - "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline)); - } - remove = true; - } else if (timeout.isCancelled()) { - remove = true; - } else { - timeout.remainingRounds--; - } - // store reference to next as we may null out timeout.next in the remove block. - HashedWheelTimeout next = timeout.next; - if (remove) { - remove(timeout); - } - timeout = next; - } - } - - void remove(HashedWheelTimeout timeout) { - HashedWheelTimeout next = timeout.next; - // remove timeout that was either processed or cancelled by updating the linked-list - if (timeout.prev != null) { - timeout.prev.next = next; - } - if (timeout.next != null) { - timeout.next.prev = timeout.prev; - } - - if (timeout == head) { - // if timeout is also the tail we need to adjust the entry too - if (timeout == tail) { - tail = null; - head = null; - } else { - head = next; - } - } else if (timeout == tail) { - // if the timeout is the tail modify the tail to be the prev node. - tail = timeout.prev; - } - // null out prev, next and bucket to allow for GC. - timeout.prev = null; - timeout.next = null; - timeout.bucket = null; - } - - /** - * Clear this bucket and return all not expired / cancelled {@link Timeout}s. - */ - void clearTimeouts(Set<Timeout> set) { - for (; ; ) { - HashedWheelTimeout timeout = pollTimeout(); - if (timeout == null) { - return; - } - if (timeout.isExpired() || timeout.isCancelled()) { - continue; - } - set.add(timeout); - } - } - - private HashedWheelTimeout pollTimeout() { - HashedWheelTimeout head = this.head; - if (head == null) { - return null; - } - HashedWheelTimeout next = head.next; - if (next == null) { - tail = this.head = null; - } else { - this.head = next; - next.prev = null; - } - - // null out prev and next to allow for GC. - head.next = null; - head.prev = null; - head.bucket = null; - return head; - } - } - -}
diff --git a/soul-common/src/main/java/org/dromara/soul/common/cache/Timeout.java b/soul-common/src/main/java/org/dromara/soul/common/cache/Timeout.java deleted file mode 100644 index 52fbbaf..0000000 --- a/soul-common/src/main/java/org/dromara/soul/common/cache/Timeout.java +++ /dev/null
@@ -1,75 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements.See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License.You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.dromara.soul.common.cache; - -/** - * timeout. - * - * @author chenbinl - * @see HashedWheelTimer; - */ -public interface Timeout { - - /** - * Returns the {@link Timer} that created this handle. - */ - Timer timer(); - - /** - * Returns the {@link TimerTask} which is associated with this handle. - */ - TimerTask task(); - - /** - * Returns {@code true} if and only if the {@link TimerTask} associated - * with this handle has been expired. - */ - boolean isExpired(); - - /** - * Returns {@code true} if and only if the {@link TimerTask} associated - * with this handle has been cancelled. - */ - boolean isCancelled(); - - /** - * Attempts to cancel the {@link TimerTask} associated with this handle. - * If the task has been executed or cancelled already, it will return with - * no side effect. - * - * @return True if the cancellation completed successfully, otherwise false - */ - boolean cancel(); - - /** - * 是否已经最后执行期限. - * - * @return true or false. - */ - long deadline(); - - /** - * 是否为自定义的一个timeout. - * - * @return true or false. - */ - default boolean isDefault() { - return true; - } -} \ No newline at end of file
diff --git a/soul-common/src/main/java/org/dromara/soul/common/cache/Timer.java b/soul-common/src/main/java/org/dromara/soul/common/cache/Timer.java deleted file mode 100644 index fbe8eda..0000000 --- a/soul-common/src/main/java/org/dromara/soul/common/cache/Timer.java +++ /dev/null
@@ -1,51 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements.See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License.You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.dromara.soul.common.cache; - -import java.util.Set; -import java.util.concurrent.TimeUnit; - -/** - * Timer. - * @author chenbin - */ -public interface Timer { - - /** - * Schedules the specified {@link TimerTask} for one-time execution after - * the specified delay. - * @param task timeTask - * @param delay delay time; - * @param unit time unit; - * @return a handle which is associated with the specified task - * - * @throws IllegalStateException if this timer has been - * {@linkplain #stop() stopped} already - */ - Timeout newTimeout(TimerTask task, long delay, TimeUnit unit); - - /** - * Releases all resources acquired by this {@link Timer} and cancels all - * tasks which were scheduled but not executed yet. - * - * @return the handles associated with the tasks which were canceled by - * this method - */ - Set<Timeout> stop(); -}
diff --git a/soul-common/src/main/java/org/dromara/soul/common/cache/TimerTask.java b/soul-common/src/main/java/org/dromara/soul/common/cache/TimerTask.java deleted file mode 100644 index 7060f23..0000000 --- a/soul-common/src/main/java/org/dromara/soul/common/cache/TimerTask.java +++ /dev/null
@@ -1,38 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements.See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License.You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.dromara.soul.common.cache; - -import java.util.concurrent.TimeUnit; - -/** - * A task which is executed after the delay specified with - * {@link Timer#newTimeout(TimerTask, long, TimeUnit)}. - * - * @author chenbin - */ -public interface TimerTask { - - /** - * Executed after the delay specified with - * {@link Timer#newTimeout(TimerTask, long, TimeUnit)}. - * - * @param timeout a handle which is associated with this task - */ - void run(Timeout timeout); -}
diff --git a/soul-common/src/main/java/org/dromara/soul/common/cache/TtlCache.java b/soul-common/src/main/java/org/dromara/soul/common/cache/TtlCache.java index b3ebac7..fb62939 100644 --- a/soul-common/src/main/java/org/dromara/soul/common/cache/TtlCache.java +++ b/soul-common/src/main/java/org/dromara/soul/common/cache/TtlCache.java
@@ -18,8 +18,12 @@ package org.dromara.soul.common.cache; -import java.util.concurrent.TimeUnit; import org.dromara.soul.common.concurrent.SoulThreadFactory; +import org.dromara.soul.common.timer.HashedWheelTimer; +import org.dromara.soul.common.timer.Timeout; +import org.dromara.soul.common.timer.TimerTask; + +import java.util.concurrent.TimeUnit; /** * 实现一个关于时间的过期的缓存；当缓存的对象到期后， @@ -94,14 +98,14 @@ * @return Timeout */ public Timeout put(K k, V v, long expire, TimeUnit unit) { - Node node = new Node(k, v, expire, unit); - return timer.newTimeout(node, expire, unit); + TtlCacheTimerTask ttlCacheTimerTask = new TtlCacheTimerTask(k, v, expire, unit); + return timer.newTimeout(ttlCacheTimerTask, expire, unit); } /** * 保存. */ - public class Node implements TimerTask { + public class TtlCacheTimerTask implements TimerTask { /** * 缓存的KEY. */ @@ -135,7 +139,7 @@ * @param expire 默认等待时间; * @param unit 默认等时间的单位; */ - public Node(K key, V value, long expire, TimeUnit unit) { + TtlCacheTimerTask(K key, V value, long expire, TimeUnit unit) { this.key = key; this.value = value; time = System.nanoTime();
diff --git a/soul-common/src/main/java/org/dromara/soul/common/timer/HashedWheelTimer.java b/soul-common/src/main/java/org/dromara/soul/common/timer/HashedWheelTimer.java new file mode 100644 index 0000000..4e43b72 --- /dev/null +++ b/soul-common/src/main/java/org/dromara/soul/common/timer/HashedWheelTimer.java
@@ -0,0 +1,829 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements.See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License.You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.dromara.soul.common.timer; + + +import org.dromara.soul.common.utils.OsUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import java.util.concurrent.atomic.AtomicLong; + +/** + * A {@link Timer} optimized for approximated I/O timeout scheduling. + * + * <h3>Tick Duration</h3> + * <p> + * As described with 'approximated', this timer does not execute the scheduled + * {@link TimerTask} on time. {@link HashedWheelTimer}, on every tick, will + * check if there are any {@link TimerTask}s behind the schedule and execute + * them. + * <p> + * You can increase or decrease the accuracy of the execution timing by + * specifying smaller or larger tick duration in the constructor. In most + * network applications, I/O timeout does not need to be accurate. Therefore, + * the default tick duration is 100 milliseconds and you will not need to try + * different configurations in most cases. + * + * <h3>Ticks per Wheel (Wheel Size)</h3> + * <p> + * {@link HashedWheelTimer} maintains a data structure called 'wheel'. + * To put simply, a wheel is a hash table of {@link TimerTask}s whose hash + * function is 'dead line of the task'. The default number of ticks per wheel + * (i.e. the size of the wheel) is 512. You could specify a larger value + * if you are going to schedule a lot of timeouts. + * + * <h3>Do not create many instances.</h3> + * <p> + * {@link HashedWheelTimer} creates a new thread whenever it is instantiated and + * started. Therefore, you should make sure to create only one instance and + * share it across your application. One of the common mistakes, that makes + * your application unresponsive, is to create a new instance for every connection. + * + * <h3>Implementation Details</h3> + * <p> + * {@link HashedWheelTimer} is based on + * <a href="http://cseweb.ucsd.edu/users/varghese/">George Varghese</a> and + * Tony Lauck's paper, + * <a href="http://cseweb.ucsd.edu/users/varghese/PAPERS/twheel.ps.Z">'Hashed + * and Hierarchical Timing Wheels: data structures to efficiently implement a + * timer facility'</a>. More comprehensive slides are located + * <a href="http://www.cse.wustl.edu/~cdgill/courses/cs6874/TimingWheels.ppt">here</a>. + */ +public class HashedWheelTimer implements Timer { + + static final Logger logger = + LoggerFactory.getLogger(HashedWheelTimer.class); + + private static final AtomicInteger INSTANCE_COUNTER = new AtomicInteger(); + private static final AtomicBoolean WARNED_TOO_MANY_INSTANCES = new AtomicBoolean(); + private static final int INSTANCE_COUNT_LIMIT = 64; + private static final long MILLISECOND_NANOS = TimeUnit.MILLISECONDS.toNanos(1); + + + private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER = + AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState"); + + private final Worker worker = new Worker(); + private final Thread workerThread; + + public static final int WORKER_STATE_INIT = 0; + public static final int WORKER_STATE_STARTED = 1; + public static final int WORKER_STATE_SHUTDOWN = 2; + @SuppressWarnings({"unused", "FieldMayBeFinal"}) + private volatile int workerState; // 0 - init, 1 - started, 2 - shut down + + private final long tickDuration; + private final HashedWheelBucket[] wheel; + private final int mask; + private final CountDownLatch startTimeInitialized = new CountDownLatch(1); + private final Queue<HashedWheelTimeout> timeouts = new ConcurrentLinkedQueue<>(); + private final Queue<HashedWheelTimeout> cancelledTimeouts = new ConcurrentLinkedDeque<>(); + private final AtomicLong pendingTimeouts = new AtomicLong(0); + private final long maxPendingTimeouts; + + private volatile long startTime; + + /** + * Creates a new timer with the default thread factory + * ({@link Executors#defaultThreadFactory()}), default tick duration, and + * default number of ticks per wheel. + */ + public HashedWheelTimer() { + this(Executors.defaultThreadFactory()); + } + + /** + * Creates a new timer with the default thread factory + * ({@link Executors#defaultThreadFactory()}) and default number of ticks + * per wheel. + * + * @param tickDuration the duration between tick + * @param unit the time unit of the {@code tickDuration} + * @throws NullPointerException if {@code unit} is {@code null} + * @throws IllegalArgumentException if {@code tickDuration} is <= 0 + */ + public HashedWheelTimer(long tickDuration, TimeUnit unit) { + this(Executors.defaultThreadFactory(), tickDuration, unit); + } + + /** + * Creates a new timer with the default thread factory + * ({@link Executors#defaultThreadFactory()}). + * + * @param tickDuration the duration between tick + * @param unit the time unit of the {@code tickDuration} + * @param ticksPerWheel the size of the wheel + * @throws NullPointerException if {@code unit} is {@code null} + * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0 + */ + public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) { + this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel); + } + + /** + * Creates a new timer with the default tick duration and default number of + * ticks per wheel. + * + * @param threadFactory a {@link ThreadFactory} that creates a + * background {@link Thread} which is dedicated to + * {@link TimerTask} execution. + * @throws NullPointerException if {@code threadFactory} is {@code null} + */ + public HashedWheelTimer(ThreadFactory threadFactory) { + this(threadFactory, 100, TimeUnit.MILLISECONDS); + } + + /** + * Creates a new timer with the default number of ticks per wheel. + * + * @param threadFactory a {@link ThreadFactory} that creates a + * background {@link Thread} which is dedicated to + * {@link TimerTask} execution. + * @param tickDuration the duration between tick + * @param unit the time unit of the {@code tickDuration} + * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null} + * @throws IllegalArgumentException if {@code tickDuration} is <= 0 + */ + public HashedWheelTimer( + ThreadFactory threadFactory, long tickDuration, TimeUnit unit) { + this(threadFactory, tickDuration, unit, 512); + } + + /** + * Creates a new timer. + * + * @param threadFactory a {@link ThreadFactory} that creates a + * background {@link Thread} which is dedicated to + * {@link TimerTask} execution. + * @param tickDuration the duration between tick + * @param unit the time unit of the {@code tickDuration} + * @param ticksPerWheel the size of the wheel + * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null} + * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0 + */ + public HashedWheelTimer( + ThreadFactory threadFactory, + long tickDuration, TimeUnit unit, int ticksPerWheel) { + this(threadFactory, tickDuration, unit, ticksPerWheel, true); + } + + /** + * Creates a new timer. + * + * @param threadFactory a {@link ThreadFactory} that creates a + * background {@link Thread} which is dedicated to + * {@link TimerTask} execution. + * @param tickDuration the duration between tick + * @param unit the time unit of the {@code tickDuration} + * @param ticksPerWheel the size of the wheel + * @param leakDetection {@code true} if leak detection should be enabled always, + * if false it will only be enabled if the worker thread is not + * a daemon thread. + * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null} + * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0 + */ + public HashedWheelTimer( + ThreadFactory threadFactory, + long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection) { + this(threadFactory, tickDuration, unit, ticksPerWheel, leakDetection, -1); + } + + /** + * Creates a new timer. + * + * @param threadFactory a {@link ThreadFactory} that creates a + * background {@link Thread} which is dedicated to + * {@link TimerTask} execution. + * @param tickDuration the duration between tick + * @param unit the time unit of the {@code tickDuration} + * @param ticksPerWheel the size of the wheel + * @param leakDetection {@code true} if leak detection should be enabled always, + * if false it will only be enabled if the worker thread is not + * a daemon thread. + * @param maxPendingTimeouts The maximum number of pending timeouts after which call to + * {@code newTimeout} will result in + * {@link java.util.concurrent.RejectedExecutionException} + * being thrown. No maximum pending timeouts limit is assumed if + * this value is 0 or negative. + * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null} + * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0 + */ + public HashedWheelTimer( + ThreadFactory threadFactory, + long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection, + long maxPendingTimeouts) { + + if (threadFactory == null) { + throw new NullPointerException("threadFactory"); + } + if (unit == null) { + throw new NullPointerException("unit"); + } + if (tickDuration <= 0) { + throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration); + } + if (ticksPerWheel <= 0) { + throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel); + } + + // Normalize ticksPerWheel to power of two and initialize the wheel. + wheel = createWheel(ticksPerWheel); + mask = wheel.length - 1; + + // Convert tickDuration to nanos. + long duration = unit.toNanos(tickDuration); + + // Prevent overflow. + if (duration >= Long.MAX_VALUE / wheel.length) { + throw new IllegalArgumentException(String.format( + "tickDuration: %d (expected: 0 < tickDuration in nanos < %d", + tickDuration, Long.MAX_VALUE / wheel.length)); + } + + if (duration < MILLISECOND_NANOS) { + logger.warn("Configured tickDuration {} smaller then {}, using 1ms.", + tickDuration, MILLISECOND_NANOS); + this.tickDuration = MILLISECOND_NANOS; + } else { + this.tickDuration = duration; + } + + workerThread = threadFactory.newThread(worker); + + this.maxPendingTimeouts = maxPendingTimeouts; + + if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT && + WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) { + reportTooManyInstances(); + } + } + + @Override + protected void finalize() throws Throwable { + try { + super.finalize(); + } finally { + // This object is going to be GCed and it is assumed the ship has sailed to do a proper shutdown. If + // we have not yet shutdown then we want to make sure we decrement the active instance count. + if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) { + INSTANCE_COUNTER.decrementAndGet(); + } + } + } + + private static HashedWheelBucket[] createWheel(int ticksPerWheel) { + if (ticksPerWheel <= 0) { + throw new IllegalArgumentException( + "ticksPerWheel must be greater than 0: " + ticksPerWheel); + } + if (ticksPerWheel > 1073741824) { + throw new IllegalArgumentException( + "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel); + } + + ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel); + HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel]; + for (int i = 0; i < wheel.length; i++) { + wheel[i] = new HashedWheelBucket(); + } + return wheel; + } + + private static int normalizeTicksPerWheel(int ticksPerWheel) { + int normalizedTicksPerWheel = 1; + while (normalizedTicksPerWheel < ticksPerWheel) { + normalizedTicksPerWheel <<= 1; + } + return normalizedTicksPerWheel; + } + + /** + * Starts the background thread explicitly. The background thread will + * start automatically on demand even if you did not call this method. + * + * @throws IllegalStateException if this timer has been + * {@linkplain #stop() stopped} already + */ + public void start() { + switch (WORKER_STATE_UPDATER.get(this)) { + case WORKER_STATE_INIT: + if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) { + workerThread.start(); + } + break; + case WORKER_STATE_STARTED: + break; + case WORKER_STATE_SHUTDOWN: + throw new IllegalStateException("cannot be started once stopped"); + default: + throw new Error("Invalid WorkerState"); + } + + // Wait until the startTime is initialized by the worker. + while (startTime == 0) { + try { + startTimeInitialized.await(); + } catch (InterruptedException ignore) { + // Ignore - it will be ready very soon. + } + } + } + + @Override + public Set<Timeout> stop() { + if (Thread.currentThread() == workerThread) { + throw new IllegalStateException( + HashedWheelTimer.class.getSimpleName() + + ".stop() cannot be called from " + + TimerTask.class.getSimpleName()); + } + + if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) { + // workerState can be 0 or 2 at this moment - let it always be 2. + if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) { + INSTANCE_COUNTER.decrementAndGet(); + } + + return Collections.emptySet(); + } + + try { + boolean interrupted = false; + while (workerThread.isAlive()) { + workerThread.interrupt(); + try { + workerThread.join(100); + } catch (InterruptedException ignored) { + interrupted = true; + } + } + + if (interrupted) { + Thread.currentThread().interrupt(); + } + } finally { + INSTANCE_COUNTER.decrementAndGet(); + } + return worker.unprocessedTimeouts(); + } + + @Override + public boolean isStop() { + return WORKER_STATE_SHUTDOWN == WORKER_STATE_UPDATER.get(this); + } + + @Override + public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) { + if (task == null) { + throw new NullPointerException("task"); + } + if (unit == null) { + throw new NullPointerException("unit"); + } + + long pendingTimeoutsCount = pendingTimeouts.incrementAndGet(); + + if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) { + pendingTimeouts.decrementAndGet(); + throw new RejectedExecutionException("Number of pending timeouts (" + + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending " + + "timeouts (" + maxPendingTimeouts + ")"); + } + + start(); + + // Add the timeout to the timeout queue which will be processed on the next tick. + // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket. + long deadline = System.nanoTime() + unit.toNanos(delay) - startTime; + + // Guard against overflow. + if (delay > 0 && deadline < 0) { + deadline = Long.MAX_VALUE; + } + HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline); + timeouts.add(timeout); + return timeout; + } + + /** + * Returns the number of pending timeouts of this {@link Timer}. + */ + public long pendingTimeouts() { + return pendingTimeouts.get(); + } + + private static void reportTooManyInstances() { + if (logger.isErrorEnabled()) { + String resourceType = HashedWheelTimer.class.getSimpleName(); + logger.error("You are creating too many " + resourceType + " instances. " + + resourceType + " is a shared resource that must be reused across the JVM," + + "so that only a few instances are created."); + } + } + + private final class Worker implements Runnable { + private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>(); + + private long tick; + + @Override + public void run() { + // Initialize the startTime. + startTime = System.nanoTime(); + if (startTime == 0) { + // We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized. + startTime = 1; + } + + // Notify the other threads waiting for the initialization at start(). + startTimeInitialized.countDown(); + + do { + final long deadline = waitForNextTick(); + if (deadline > 0) { + int idx = (int) (tick & mask); + processCancelledTasks(); + HashedWheelBucket bucket = + wheel[idx]; + transferTimeoutsToBuckets(); + bucket.expireTimeouts(deadline); + tick++; + } + } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED); + + // Fill the unprocessedTimeouts so we can return them from stop() method. + for (HashedWheelBucket bucket : wheel) { + bucket.clearTimeouts(unprocessedTimeouts); + } + for (; ; ) { + HashedWheelTimeout timeout = timeouts.poll(); + if (timeout == null) { + break; + } + if (!timeout.isCancelled()) { + unprocessedTimeouts.add(timeout); + } + } + processCancelledTasks(); + } + + private void transferTimeoutsToBuckets() { + // transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just + // adds new timeouts in a loop. + for (int i = 0; i < 100000; i++) { + HashedWheelTimeout timeout = timeouts.poll(); + if (timeout == null) { + // all processed + break; + } + if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) { + // Was cancelled in the meantime. + continue; + } + + long calculated = timeout.deadline / tickDuration; + timeout.remainingRounds = (calculated - tick) / wheel.length; + + final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past. + int stopIndex = (int) (ticks & mask); + + HashedWheelBucket bucket = wheel[stopIndex]; + bucket.addTimeout(timeout); + } + } + + private void processCancelledTasks() { + for (; ; ) { + HashedWheelTimeout timeout = cancelledTimeouts.poll(); + if (timeout == null) { + // all processed + break; + } + try { + timeout.remove(); + } catch (Throwable t) { + if (logger.isWarnEnabled()) { + logger.warn("An exception was thrown while process a cancellation task", t); + } + } + } + } + + /** + * calculate goal nanoTime from startTime and current tick number, + * then wait until that goal has been reached. + * + * @return Long.MIN_VALUE if received a shutdown request, + * current time otherwise (with Long.MIN_VALUE changed by +1) + */ + private long waitForNextTick() { + long deadline = tickDuration * (tick + 1); + + for (; ; ) { + final long currentTime = System.nanoTime() - startTime; + long sleepTimeMs = (deadline - currentTime + 999999) / 1000000; + + if (sleepTimeMs <= 0) { + if (currentTime == Long.MIN_VALUE) { + return -Long.MAX_VALUE; + } else { + return currentTime; + } + } + + // Check if we run on windows, as if thats the case we will need + // to round the sleepTime as workaround for a bug that only affect + // the JVM if it runs on windows. + // + // See https://github.com/netty/netty/issues/356 + if (OsUtils.isWindows()) { + sleepTimeMs = sleepTimeMs / 10 * 10; + } + + try { + Thread.sleep(sleepTimeMs); + } catch (InterruptedException ignored) { + if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) { + return Long.MIN_VALUE; + } + } + } + } + + public Set<Timeout> unprocessedTimeouts() { + return Collections.unmodifiableSet(unprocessedTimeouts); + } + } + + private static final class HashedWheelTimeout implements Timeout { + + private static final int ST_INIT = 0; + private static final int ST_CANCELLED = 1; + private static final int ST_EXPIRED = 2; + private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER = + AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state"); + + private final HashedWheelTimer timer; + private final TimerTask task; + private final long deadline; + + @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization"}) + private volatile int state = ST_INIT; + + // remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the + // HashedWheelTimeout will be added to the correct HashedWheelBucket. + long remainingRounds; + + // This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list. + // As only the workerThread will act on it there is no need for synchronization / volatile. + HashedWheelTimeout next; + HashedWheelTimeout prev; + + // The bucket to which the timeout was added + HashedWheelBucket bucket; + + HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) { + this.timer = timer; + this.task = task; + this.deadline = deadline; + } + + @Override + public Timer timer() { + return timer; + } + + @Override + public TimerTask task() { + return task; + } + + @Override + public boolean cancel() { + // only update the state it will be removed from HashedWheelBucket on next tick. + if (!compareAndSetState(ST_INIT, ST_CANCELLED)) { + return false; + } + // If a task should be canceled we put this to another queue which will be processed on each tick. + // So this means that we will have a GC latency of max. 1 tick duration which is good enough. This way + // we can make again use of our MpscLinkedQueue and so minimize the locking / overhead as much as possible. + timer.cancelledTimeouts.add(this); + return true; + } + + void remove() { + HashedWheelBucket bucket = this.bucket; + if (bucket != null) { + bucket.remove(this); + } else { + timer.pendingTimeouts.decrementAndGet(); + } + } + + public boolean compareAndSetState(int expected, int state) { + return STATE_UPDATER.compareAndSet(this, expected, state); + } + + public int state() { + return state; + } + + @Override + public boolean isCancelled() { + return state() == ST_CANCELLED; + } + + @Override + public boolean isExpired() { + return state() == ST_EXPIRED; + } + + public void expire() { + if (!compareAndSetState(ST_INIT, ST_EXPIRED)) { + return; + } + + try { + task.run(this); + } catch (Throwable t) { + if (logger.isWarnEnabled()) { + logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t); + } + } + } + + @Override + public String toString() { + final long currentTime = System.nanoTime(); + long remaining = deadline - currentTime + timer.startTime; + + StringBuilder buf = new StringBuilder(192) + .append(this.getClass().getSimpleName()) + .append('(') + .append("deadline: "); + if (remaining > 0) { + buf.append(remaining) + .append(" ns later"); + } else if (remaining < 0) { + buf.append(-remaining) + .append(" ns ago"); + } else { + buf.append("now"); + } + + if (isCancelled()) { + buf.append(", cancelled"); + } + + return buf.append(", task: ") + .append(task()) + .append(')') + .toString(); + } + + } + + /** + * Bucket that stores HashedWheelTimeouts. These are stored in a linked-list like datastructure to allow easy + * removal of HashedWheelTimeouts in the middle. Also the HashedWheelTimeout act as nodes themself and so no + * extra object creation is needed. + */ + private static final class HashedWheelBucket { + // Used for the linked-list datastructure + private HashedWheelTimeout head; + private HashedWheelTimeout tail; + + /** + * Add {@link HashedWheelTimeout} to this bucket. + */ + public void addTimeout(HashedWheelTimeout timeout) { + assert timeout.bucket == null; + timeout.bucket = this; + if (head == null) { + head = tail = timeout; + } else { + tail.next = timeout; + timeout.prev = tail; + tail = timeout; + } + } + + /** + * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}. + */ + public void expireTimeouts(long deadline) { + HashedWheelTimeout timeout = head; + + // process all timeouts + while (timeout != null) { + HashedWheelTimeout next = timeout.next; + if (timeout.remainingRounds <= 0) { + next = remove(timeout); + if (timeout.deadline <= deadline) { + timeout.expire(); + } else { + // The timeout was placed into a wrong slot. This should never happen. + throw new IllegalStateException(String.format( + "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline)); + } + } else if (timeout.isCancelled()) { + next = remove(timeout); + } else { + timeout.remainingRounds--; + } + timeout = next; + } + } + + public HashedWheelTimeout remove(HashedWheelTimeout timeout) { + HashedWheelTimeout next = timeout.next; + // remove timeout that was either processed or cancelled by updating the linked-list + if (timeout.prev != null) { + timeout.prev.next = next; + } + if (timeout.next != null) { + timeout.next.prev = timeout.prev; + } + + if (timeout == head) { + // if timeout is also the tail we need to adjust the entry too + if (timeout == tail) { + tail = null; + head = null; + } else { + head = next; + } + } else if (timeout == tail) { + // if the timeout is the tail modify the tail to be the prev node. + tail = timeout.prev; + } + // null out prev, next and bucket to allow for GC. + timeout.prev = null; + timeout.next = null; + timeout.bucket = null; + timeout.timer.pendingTimeouts.decrementAndGet(); + return next; + } + + /** + * Clear this bucket and return all not expired / cancelled {@link Timeout}s. + */ + public void clearTimeouts(Set<Timeout> set) { + for (; ; ) { + HashedWheelTimeout timeout = pollTimeout(); + if (timeout == null) { + return; + } + if (timeout.isExpired() || timeout.isCancelled()) { + continue; + } + set.add(timeout); + } + } + + private HashedWheelTimeout pollTimeout() { + HashedWheelTimeout head = this.head; + if (head == null) { + return null; + } + HashedWheelTimeout next = head.next; + if (next == null) { + tail = this.head = null; + } else { + this.head = next; + next.prev = null; + } + + // null out prev and next to allow for GC. + head.next = null; + head.prev = null; + head.bucket = null; + return head; + } + } +}
diff --git a/soul-common/src/main/java/org/dromara/soul/common/timer/Timeout.java b/soul-common/src/main/java/org/dromara/soul/common/timer/Timeout.java new file mode 100644 index 0000000..02ef6e4 --- /dev/null +++ b/soul-common/src/main/java/org/dromara/soul/common/timer/Timeout.java
@@ -0,0 +1,60 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements.See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License.You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.dromara.soul.common.timer; + +/** + * A handle associated with a {@link TimerTask} that is returned by a + * {@link Timer}. + * + * @author netty. + */ +public interface Timeout { + + /** + * Returns the {@link Timer} that created this handle. + */ + Timer timer(); + + /** + * Returns the {@link TimerTask} which is associated with this handle. + */ + TimerTask task(); + + /** + * Returns {@code true} if and only if the {@link TimerTask} associated + * with this handle has been expired. + */ + boolean isExpired(); + + /** + * Returns {@code true} if and only if the {@link TimerTask} associated + * with this handle has been cancelled. + */ + boolean isCancelled(); + + /** + * Attempts to cancel the {@link TimerTask} associated with this handle. + * If the task has been executed or cancelled already, it will return with + * no side effect. + * + * @return True if the cancellation completed successfully, otherwise false + */ + boolean cancel(); +} \ No newline at end of file
diff --git a/soul-common/src/main/java/org/dromara/soul/common/timer/Timer.java b/soul-common/src/main/java/org/dromara/soul/common/timer/Timer.java new file mode 100644 index 0000000..a26cc6a --- /dev/null +++ b/soul-common/src/main/java/org/dromara/soul/common/timer/Timer.java
@@ -0,0 +1,55 @@ +/* + * Copyright 2012 The Netty Project + * + * The Netty Project licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package org.dromara.soul.common.timer; + +import java.util.Set; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; + +/** + * Schedules {@link TimerTask}s for one-time future execution in a background + * thread. + * + * @author netty. + */ +public interface Timer { + + /** + * Schedules the specified {@link TimerTask} for one-time execution after + * the specified delay. + * + * @return a handle which is associated with the specified task + * @throws IllegalStateException if this timer has been {@linkplain #stop() stopped} already + * @throws RejectedExecutionException if the pending timeouts are too many and creating new timeout + * can cause instability in the system. + */ + Timeout newTimeout(TimerTask task, long delay, TimeUnit unit); + + /** + * Releases all resources acquired by this {@link Timer} and cancels all + * tasks which were scheduled but not executed yet. + * + * @return the handles associated with the tasks which were canceled by + * this method + */ + Set<Timeout> stop(); + + /** + * stop method status. + * @return bool. + */ + boolean isStop(); +}
diff --git a/soul-common/src/main/java/org/dromara/soul/common/timer/TimerTask.java b/soul-common/src/main/java/org/dromara/soul/common/timer/TimerTask.java new file mode 100644 index 0000000..c26526a --- /dev/null +++ b/soul-common/src/main/java/org/dromara/soul/common/timer/TimerTask.java
@@ -0,0 +1,40 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements.See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License.You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.dromara.soul.common.timer; + +import java.util.concurrent.TimeUnit; + +/** + * A task which is executed after the delay specified with + * {@link Timer#newTimeout(TimerTask, long, TimeUnit)}. + * io.netty.util.TimerTask. + * + * @author netty. + */ +public interface TimerTask { + + /** + * Executed after the delay specified with + * {@link Timer#newTimeout(TimerTask, long, TimeUnit)}. + * + * @param timeout a handle which is associated with this task + */ + void run(Timeout timeout); +}
diff --git a/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/AbstractRegistry.java b/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/AbstractRegistry.java index 09e3aef..57d2d3f 100644 --- a/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/AbstractRegistry.java +++ b/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/AbstractRegistry.java
@@ -17,11 +17,16 @@ package org.dromara.soul.register.api; +import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; + +import com.google.common.collect.Sets; import org.dromara.soul.common.exception.SoulException; import org.dromara.soul.common.http.URL; +import org.dromara.soul.register.api.exception.RegisterException; /** * AbstractRegistry. @@ -41,12 +46,12 @@ /** * The notifyUrls. */ - protected final Map<URL, RegisterNotifyListener> subscribed = new ConcurrentHashMap<>(); + private final Map<URL, RegisterNotifyListener> subscribed = new ConcurrentHashMap<>(); /** * The Registered. */ - protected final Map<URL, Object> registered = new ConcurrentHashMap<>(); + private final Set<URL> registered = Sets.newConcurrentHashSet(); /** * Instantiates a new Abstract registry. @@ -69,10 +74,21 @@ } /** - * Retry. + * get subscribed. + * + * @return map. */ - public void retry() { - //todo:重试. + Map<URL, RegisterNotifyListener> getSubscribed() { + return Collections.unmodifiableMap(subscribed); + } + + /** + * get registered. + * + * @return set. + */ + Set<URL> getRegistered() { + return Collections.unmodifiableSet(registered); } /** @@ -89,7 +105,7 @@ * * @return the remote url */ - public URL getRemoteUrl() { + protected URL getRemoteUrl() { return remoteUrl; } @@ -103,22 +119,38 @@ @Override public void register(URL url) { - + if (url == null) { + throw new RegisterException("url is null || url == null"); + } + registered.add(url); } + @Override public void unregister(URL url) { - + if (url == null) { + throw new RegisterException("url is null || url == null"); + } + registered.remove(url); } @Override public void subscribe(URL url, RegisterNotifyListener listener) { - + if (url == null) { + throw new RegisterException("url is null || url == null"); + } + if (listener == null) { + throw new RegisterException("listener is null || listener == null"); + } + subscribed.putIfAbsent(url, listener); } @Override public void unsubscribe(URL url) { - + if (url == null) { + throw new RegisterException("url is null || url == null"); + } + subscribed.remove(url); } @Override
diff --git a/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/FailbackRegistry.java b/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/FailbackRegistry.java new file mode 100644 index 0000000..d861987 --- /dev/null +++ b/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/FailbackRegistry.java
@@ -0,0 +1,280 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements.See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License.You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.dromara.soul.register.api; + +import com.google.common.collect.Maps; +import org.dromara.soul.common.concurrent.SoulThreadFactory; +import org.dromara.soul.common.http.URL; +import org.dromara.soul.common.timer.HashedWheelTimer; +import org.dromara.soul.common.timer.Timeout; +import org.dromara.soul.common.timer.Timer; +import org.dromara.soul.common.timer.TimerTask; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** + * FailbackRegistry . + * + * @author sixh + */ +public abstract class FailbackRegistry extends AbstractRegistry { + + private Logger logger = LoggerFactory.getLogger(FailbackRegistry.class); + + private final Map<URL, FailbackTask> failbackRegister = Maps.newConcurrentMap(); + + private final Map<URL, FailbackTask> failbackSubscribe = Maps.newConcurrentMap(); + + private static final HashedWheelTimer TIMER = new HashedWheelTimer(SoulThreadFactory.create("soul.failback.register")); + /** + * retry exec time. + */ + private final Long retryTime = TimeUnit.SECONDS.toMillis(5); + + /** + * Instantiates a new Abstract registry. + * + * @param url the url. + */ + public FailbackRegistry(URL url) { + super(url); + } + + @Override + public void register(URL url) { + removeFailbackRegister(url); + try { + super.register(url); + doRegister(url); + } catch (Throwable t) { + addFailbackRegister(url, new FailbackTask(url, url1 -> { + doRegister(url1); + failbackRegister.remove(url1); + }, retryTime)); + } + } + + @Override + public void unregister(URL url) { + removeFailbackRegister(url); + try { + super.unregister(url); + doUnRegister(url); + } catch (Throwable t) { + addFailbackRegister(url, new FailbackTask(url, url1 -> { + doUnRegister(url1); + failbackRegister.remove(url1); + }, retryTime)); + } + } + + @Override + public void subscribe(URL url, RegisterNotifyListener listener) { + removeFailbackSubscribe(url); + try { + super.subscribe(url, listener); + doSubscribe(url); + } catch (Throwable t) { + addFailbackSubscribe(url, new FailbackTask(url, url1 -> { + doSubscribe(url1); + failbackSubscribe.remove(url1); + }, retryTime)); + } + } + + @Override + public void unsubscribe(URL url) { + removeFailbackSubscribe(url); + try { + super.unsubscribe(url); + doUnSubscribe(url); + } catch (Throwable t) { + addFailbackSubscribe(url, new FailbackTask(url, url1 -> { + doUnSubscribe(url1); + failbackSubscribe.remove(url1); + }, retryTime)); + } + } + + private void addFailbackRegister(URL url, FailbackTask task) { + FailbackTask failbackTask = failbackRegister.get(url); + if (failbackTask != null) { + return; + } + failbackRegister.putIfAbsent(url, task); + TIMER.newTimeout(task, retryTime, TimeUnit.MILLISECONDS); + } + + private void removeFailbackRegister(URL url) { + FailbackTask remove = failbackRegister.remove(url); + if (remove != null) { + remove.cancel(); + } + } + + private void addFailbackSubscribe(URL url, FailbackTask task) { + FailbackTask failbackTask = failbackRegister.get(url); + if (failbackTask != null) { + return; + } + failbackSubscribe.putIfAbsent(url, task); + TIMER.newTimeout(task, retryTime, TimeUnit.MILLISECONDS); + } + + private void removeFailbackSubscribe(URL url) { + FailbackTask remove = failbackSubscribe.remove(url); + if (remove != null) { + remove.cancel(); + } + } + + protected void recover() { + Set<URL> registered = getRegistered(); + if (!registered.isEmpty()) { + if (logger.isInfoEnabled()) { + logger.info("registered : {}", registered); + } + registered.forEach(e -> addFailbackSubscribe(e, new FailbackTask(e, + url -> { + doRegister(url); + failbackRegister.remove(url); + }, retryTime))); + } + Map<URL, RegisterNotifyListener> subscribed = getSubscribed(); + if (!subscribed.isEmpty()) { + if (logger.isInfoEnabled()) { + logger.info("subscribed: {}", subscribed); + } + subscribed.forEach((k, v) -> addFailbackSubscribe(k, new FailbackTask(k, url -> { + doSubscribe(url); + failbackSubscribe.remove(url); + }, retryTime))); + } + } + + /** + * register template method. + * + * @param url url. + */ + protected abstract void doRegister(URL url); + + /** + * Do un register. + * + * @param url the url + */ + protected abstract void doUnRegister(URL url); + + /** + * Do subscribe. + * + * @param url the url + */ + protected abstract void doSubscribe(URL url); + + /** + * Do un subscribe. + * + * @param url the url + */ + protected abstract void doUnSubscribe(URL url); + + /** + * The type Failback task. + */ + static class FailbackTask implements TimerTask { + + private FailbackRunner runner; + + private URL url; + + private Long tick; + + private volatile boolean cancel = false; + + /** + * Instantiates a new Failback task. + * + * @param url the url + * @param runner the runner + * @param tick the tick + */ + FailbackTask(URL url, FailbackRunner runner, Long tick) { + this.runner = runner; + this.url = url; + this.tick = tick; + } + + /** + * Cancel. + */ + void cancel() { + this.cancel = true; + } + + /** + * Is cancel boolean. + * + * @return the boolean + */ + private boolean isCancel() { + return cancel; + } + + @Override + public void run(Timeout timeout) { + if (timeout.isCancelled() || timeout.timer().isStop() || isCancel()) { + // other thread cancel this timeout or stop the timer. + return; + } + try { + runner.apply(url); + } catch (Throwable e) { + replay(timeout, tick); + } + } + + private void replay(Timeout timeout, long tick) { + Timer timer = timeout.timer(); + if (timer.isStop() || timeout.isCancelled() || isCancel()) { + return; + } + timer.newTimeout(this, tick, TimeUnit.MILLISECONDS); + } + } + + /** + * The interface Failback runner. + */ + @FunctionalInterface + interface FailbackRunner { + /** + * Apply. + * + * @param url the url + */ + void apply(URL url); + } +}
diff --git a/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/exception/RegisterException.java b/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/exception/RegisterException.java new file mode 100644 index 0000000..d6e1fe2 --- /dev/null +++ b/soul-register/soul-register-api/src/main/java/org/dromara/soul/register/api/exception/RegisterException.java
@@ -0,0 +1,41 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements.See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License.You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.dromara.soul.register.api.exception; + +import org.dromara.soul.common.exception.SoulException; + +/** + * RegisterException . + * + * @author sixh + */ +public class RegisterException extends SoulException { + public RegisterException(Throwable e) { + super(e); + } + + public RegisterException(String message) { + super(message); + } + + public RegisterException(String message, Throwable throwable) { + super(message, throwable); + } +}
diff --git a/soul-register/soul-register-zookeeper/src/main/java/ZookeeperRegistry.java b/soul-register/soul-register-zookeeper/src/main/java/ZookeeperRegistry.java index 7f55846..f977a59 100644 --- a/soul-register/soul-register-zookeeper/src/main/java/ZookeeperRegistry.java +++ b/soul-register/soul-register-zookeeper/src/main/java/ZookeeperRegistry.java
@@ -15,16 +15,17 @@ * limitations under the License. */ -import java.util.Optional; import org.dromara.soul.common.exception.SoulException; import org.dromara.soul.common.extension.Join; import org.dromara.soul.common.http.URL; -import org.dromara.soul.register.api.AbstractRegistry; +import org.dromara.soul.register.api.FailbackRegistry; import org.dromara.soul.register.api.RegisterConst; import org.dromara.soul.remoting.zookeeper.ZookeeperClient; import org.dromara.soul.remoting.zookeeper.ZookeeperClientCache; import org.dromara.soul.remoting.zookeeper.ZookeeperStatusCallback; +import java.util.Optional; + import static org.dromara.soul.register.api.RegisterConst.BASE_URL_PATH_KEY; /** @@ -34,7 +35,7 @@ * @author sixh */ @Join -public class ZookeeperRegistry extends AbstractRegistry { +public class ZookeeperRegistry extends FailbackRegistry { private final ZookeeperClient client; @@ -43,7 +44,7 @@ client = ZookeeperClientCache.getClient(url); client.statusChange(connectionState -> { if (connectionState.equals(ZookeeperStatusCallback.RECONNECTED)) { - retry(); + recover(); } }); } @@ -64,15 +65,6 @@ return path + url.encode(); } - @Override - public void register(URL url) { - try { - String path = toPath(url); - client.create(path, isEphemeral(url)); - } catch (Throwable e) { - throw new SoulException("register failed " + url + "to zookeeper " + getRemoteUrl()); - } - } private boolean isEphemeral(URL url) { String parameter = url.getParameter(RegisterConst.EPHEMERAL_KEY); @@ -82,11 +74,31 @@ } @Override - public void unregister(URL url) { + protected void doRegister(URL url) { + try { + String path = toPath(url); + client.create(path, isEphemeral(url)); + } catch (Throwable e) { + throw new SoulException("register failed " + url + "to zookeeper " + getRemoteUrl()); + } + } + + @Override + protected void doUnRegister(URL url) { try { client.delete(toPath(url)); } catch (Throwable e) { throw new SoulException("unregister failed " + url + "to zookeeper " + getRemoteUrl()); } } + + @Override + protected void doSubscribe(URL url) { + + } + + @Override + protected void doUnSubscribe(URL url) { + + } }
diff --git a/soul-remoting/soul-remoting-api/src/main/java/org/dromara/soul/remoting/api/ChannelCache.java b/soul-remoting/soul-remoting-api/src/main/java/org/dromara/soul/remoting/api/ChannelCache.java index 138a9800f..8434ae3 100644 --- a/soul-remoting/soul-remoting-api/src/main/java/org/dromara/soul/remoting/api/ChannelCache.java +++ b/soul-remoting/soul-remoting-api/src/main/java/org/dromara/soul/remoting/api/ChannelCache.java
@@ -107,10 +107,10 @@ Channel get(String key) { Timeout timeout = CACHES.get(key); if (timeout != null && !timeout.isDefault()) { - Node node = (Node) timeout.task(); - String sKey = node.getKey(); + TtlCacheTimerTask ttlCacheTimerTask = (TtlCacheTimerTask) timeout.task(); + String sKey = ttlCacheTimerTask.getKey(); if (sKey.equals(key)) { - return node.getValue(); + return ttlCacheTimerTask.getValue(); } } return null; @@ -154,6 +154,6 @@ @SuppressWarnings("unchecked") public Collection<Channel> getAll() { - return CACHES.values().stream().map(e -> ((Node) e.task()).getValue()).collect(Collectors.toList()); + return CACHES.values().stream().map(e -> ((TtlCacheTimerTask) e.task()).getValue()).collect(Collectors.toList()); } }