時間輪演算法
時間輪是一種高效、低延遲的調度資料結構，其在Linux內核中廣泛使用，是Linux內核定時器的實作方法和基礎之一，按使用場景，大致可以分為兩種時間輪：原始時間輪和分層時間輪，分層時間輪是原始時間輪的升級版本，來應對時間“槽”數量比較大的情況，對記憶體和精度都有很高要求的情況，延遲任務的場景一般只需要用到原始時間輪就可以了，
代碼案例
推薦使用Netty提供的HashedWheelTimer工具類來實作延遲任務，
引入依賴：
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-common</artifactId>
<version>4.1.23.Final</version>
</dependency>
紅包過期佇列資訊：
/**
* 紅包過期佇列資訊
*/
public class RedPacketTimerTask implements TimerTask {
private static final DateTimeFormatter F = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
/**
* 紅包 ID
*/
private final long redPacketId;
/**
* 創建時間戳
*/
private final long timestamp;
public RedPacketTimerTask(long redPacketId) {
this.redPacketId = redPacketId;
this.timestamp = System.currentTimeMillis();
}
@Override
public void run(Timeout timeout) {
//異步處理任務
System.out.println(String.format("任務執行時間:%s,紅包創建時間:%s,紅包ID:%s",
LocalDateTime.now().format(F), LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()).format(F), redPacketId));
}
}
測驗用例：
/**
* 基于 netty 的時間輪演算法 HashedWheelTimer 實作的延遲任務
*/
public class RedPacketHashedWheelTimer {
private static final DateTimeFormatter F = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
public static void main(String[] args) throws Exception {
ThreadFactory factory = r -> {
Thread thread = new Thread(r);
thread.setDaemon(true);
thread.setName("RedPacketHashedWheelTimerWorker");
return thread;
};
/**
* @param tickDuration - 每tick一次的時間間隔
* @param unit - tickDuration 的時間單位
* @param ticksPerWheel - 時間輪中的槽數
* @param leakDetection - 檢查記憶體溢位
*/
Timer timer = new HashedWheelTimer(factory, 1,
TimeUnit.SECONDS, 100,true);
System.out.println(String.format("開始任務時間:%s",LocalDateTime.now().format(F)));
for(int i=1;i<10;i++){
TimerTask timerTask = new RedPacketTimerTask(i);
timer.newTimeout(timerTask, i, TimeUnit.SECONDS);
}
Thread.sleep(Integer.MAX_VALUE);
}
}
列印任務執行日志：
開始任務時間:2020-02-12 15:22:23.404
任務執行時間:2020-02-12 15:22:25.410,紅包創建時間:2020-02-12 15:22:23.409,紅包ID:1
任務執行時間:2020-02-12 15:22:26.411,紅包創建時間:2020-02-12 15:22:23.414,紅包ID:2
任務執行時間:2020-02-12 15:22:27.424,紅包創建時間:2020-02-12 15:22:23.414,紅包ID:3
任務執行時間:2020-02-12 15:22:28.410,紅包創建時間:2020-02-12 15:22:23.414,紅包ID:4
任務執行時間:2020-02-12 15:22:29.411,紅包創建時間:2020-02-12 15:22:23.414,紅包ID:5
任務執行時間:2020-02-12 15:22:30.409,紅包創建時間:2020-02-12 15:22:23.414,紅包ID:6
任務執行時間:2020-02-12 15:22:31.411,紅包創建時間:2020-02-12 15:22:23.414,紅包ID:7
任務執行時間:2020-02-12 15:22:32.409,紅包創建時間:2020-02-12 15:22:23.414,紅包ID:8
任務執行時間:2020-02-12 15:22:33.411,紅包創建時間:2020-02-12 15:22:23.414,紅包ID:9
原始碼相關
其核心是workerThread執行緒，主要負責每過tickDuration時間就累加一次tick，同時也負責執行到期的timeout任務以及添加timeout任務到指定的wheel中，
構造方法：
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.
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);
leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
this.maxPendingTimeouts = maxPendingTimeouts;
if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
reportTooManyInstances();
}
}
新增任務，創建即啟動：
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;
}
執行緒啟動：
/**
* 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.
}
}
}
執行相關操作：
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();
}
小結
以上方案并沒有實作持久化和分布式，生產環境可根據實際業務需求選擇使用，
原始碼
https://gitee.com/52itstyle/spring-boot-seckill
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/16290.html
標籤：架構設計
上一篇：分布式系統的CAP定理
下一篇：springcloud 專案原始碼 微服務 分布式 Activiti6 作業流 vue.js html 跨域 前后分離