PowerMsg3 定时任务调度器：高效、可靠的 HashedWheelTimer 实现

该类使用 HashedWheelTimer 进行定时任务调度，提供可靠的定时任务调度功能，支持一次性、固定延迟、固定频率等多种调度方式。

代码分析：

package com.taobao.powermsg3.sdk.localserver.scheduler;

import com.taobao.hsf.NamedThreadFactory;
import com.taobao.powermsg3.sdk.localserver.PmLocalServerSDKConfig;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import io.vertx.core.Context;
import lombok.Data;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.Duration;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;


@Component
@Slf4j
public final class Scheduler {
    public static final int MAX_TICKS_PER_WHEEL = 2048;
    public static final int MINIMAL_TICK_IN_MILLIS = 1;
    public static final String LOCALSERVER_SCHEDULER_THREAD_NAME = "localserver-scheduler";
    private final HashedWheelTimer hashedWheelTimer;

    private final AtomicLong timerIdGen = new AtomicLong();

    private final ConcurrentHashMap<Object, Timeout> timerKey2Timeouts = new ConcurrentHashMap<>();

    @Autowired
    public Scheduler(final PmLocalServerSDKConfig pmLocalServerSDKConfig) {
        final PmLocalServerSDKConfig.SchedulerConfig schedulerConfig = pmLocalServerSDKConfig.getScheduler();
        final int tickInMillis = Math.max(MINIMAL_TICK_IN_MILLIS, schedulerConfig.getTickDurationInMillis());
        final int ticksPerWheel = Math.max(MAX_TICKS_PER_WHEEL, schedulerConfig.getTicksPerWheel());
        this.hashedWheelTimer = new HashedWheelTimer(
            new NamedThreadFactory(LOCALSERVER_SCHEDULER_THREAD_NAME),
            tickInMillis, TimeUnit.MILLISECONDS,
            ticksPerWheel
        );
    }

    public boolean cancel(final Object timerKey) {
        final Timeout timeout = timerKey2Timeouts.get(timerKey);
        tryCancel(timeout);
        return true;
    }

    public Long scheduleOnce(@NonNull final Duration delay,
                             @NonNull final Runnable runnable,
                             @NonNull final Context context) {
        final Long id = timerIdGen.incrementAndGet();
        scheduleOnce(id, delay, runnable, context);
        return id;
    }

    public void scheduleOnce(@NonNull final Object timerKey,
                             @NonNull final Duration delay,
                             @NonNull final Runnable runnable,
                             @NonNull final Context context) {
        scheduleWithTimerKey(delay, new ScheduledTimerTask(timerKey, delay, false, runnable, context));
    }

    public Long scheduleWithFixDelay(@NonNull final Duration initialDelay,
                                     @NonNull final Duration delay,
                                     @NonNull final Runnable runnable,
                                     @NonNull final Context context) {
        final Long id = timerIdGen.incrementAndGet();
        scheduleWithFixDelay(id, initialDelay, delay, runnable, context);
        return id;
    }

    public void scheduleWithFixDelay(@NonNull final Object timerKey,
                                     @NonNull final Duration initialDelay,
                                     @NonNull final Duration delay,
                                     @NonNull final Runnable runnable,
                                     @NonNull final Context context) {

        timerKey2Timeouts.compute(timerKey, (key, timeout) -> {
            tryCancel(timeout);
            return scheduleWithTimerKey(initialDelay,
                new ScheduledTimerTask(timerKey, delay, true, runnable, context));
        });
    }

    @Data
    private class ScheduledTimerTask implements TimerTask {
        private final Object timerKey;
        private final Duration delay;
        private final boolean reschedule;
        private final Runnable runnable;
        private final Context context;

        @Override
        public void run(final Timeout timeout) throws Exception {
            try {
                context.runOnContext(e -> runnable.run());
            } catch (Exception e) {
                log.error("Error when trigger task of runnable:[{}]", runnable, e);
            } finally {
                if (reschedule) {
                    timerKey2Timeouts.computeIfPresent(
                        timerKey,
                        (key, value) -> {
                            tryCancel(timeout);
                            return scheduleWithTimerKey(delay, ScheduledTimerTask.this);
                        });
                }
            }
        }
    }

    private boolean tryCancel(final Timeout timeout) {
        if (timeout == null || timeout.isCancelled() || timeout.isExpired()) {
            return true;
        }
        return timeout.cancel();
    }

    private Timeout scheduleWithTimerKey(final Duration delay,
                                         final ScheduledTimerTask scheduledTimerTask) {
        return hashedWheelTimer.newTimeout(
            scheduledTimerTask,
            delay.toNanos(),
            TimeUnit.NANOSECONDS);
    }


    @PostConstruct
    protected void init() {
        hashedWheelTimer.start();
    }

    @PreDestroy
    protected void stop() {
        final Set<Timeout> timeouts = hashedWheelTimer.stop();
        if (CollectionUtils.isNotEmpty(timeouts)) {
            timeouts.stream()
                .filter(timeout -> !timeout.isCancelled() && !timeout.isExpired())
                .forEach(timeout -> {
                    final TimerTask task = timeout.task();
                    try {
                        task.run(timeout);
                    }
                    catch (Exception e) {
                        //Ignore
                    }
                });
        }
    }
}

优化建议：

TimerKey 的数据结构选择：
- 目前使用 ConcurrentHashMap 进行 TimerKey 与 Timeout 的映射，可以考虑使用 ConcurrentSkipListMap 或 TreeMap 代替，这样可以保证 TimerKey 的有序性，方便进行一些范围查询等操作。
TimerIdGen 的线程安全性：
- 目前 TimerIdGen 是一个 AtomicLong，可以考虑将其封装在一个单独的类中，确保线程安全性，同时可以提供更多的定制化功能，例如定制化 id 生成策略。
异常处理：
- 在定时任务执行时，如果出现异常，只是简单地记录了错误日志，没有进行其他处理。可以考虑对异常进行捕获和处理，例如定制化异常处理策略、自动重试等。
配置项动态调整：
- 目前 Scheduler 的配置项在初始化时进行了一次性设置，无法动态调整。可以考虑将配置项设置为动态可调整的，例如通过 JMX 或配置中心等方式进行动态调整。
定时任务执行性能优化：
- 在定时任务执行时，使用了 Vert.x 的 Context 进行任务执行，这样可以确保任务执行在正确的线程上，避免了多线程竞争问题。不过 Context 的创建和销毁也需要一定的时间和性能开销，可以尝试使用其他更轻量级的方式进行任务执行，例如使用 CompletableFuture 等。

总结：

该类代码简洁、易读，但还可以进一步优化。通过优化 TimerKey 的数据结构、增强 TimerIdGen 的线程安全性、完善异常处理机制、实现配置项动态调整以及提升定时任务执行性能，可以使该调度器更加灵活、高效和可靠。