java简单手写版本实现时间轮算法

时间轮

关于时间轮的介绍,网上有很多,这里就不重复了

核心思想

• 一个环形数组存储时间轮的所有槽(看你的手表),每个槽对应当前时间轮的最小精度
• 超过当前时间轮最大表示范围的会被丢到上层时间轮,上层时间轮的最小精度即为下层时间轮能表达的最大时间(时分秒概念)
• 每个槽对应一个环形链表存储该时间应该被执行的任务
• 需要一个线程去驱动指针运转,获取到期任务
  

以下给出java 简单手写版本实现

代码实现

时间轮主数据结构

/**
 * @author apdoer
 * @version 1.0
 * @date 2021/3/22 19:31
 */
@slf4j
public class timewheel {
  /**
   * 一个槽的时间间隔(时间轮最小刻度)
   */
  private long tickms;

  /**
   * 时间轮大小(槽的个数)
   */
  private int wheelsize;

  /**
   * 一轮的时间跨度
   */
  private long interval;

  private long currenttime;

  /**
   * 槽
   */
  private timertasklist[] buckets;

  /**
   * 上层时间轮
   */
  private volatile timewheel overflowwheel;

  /**
   * 一个timer只有一个delayqueue
   */
  private delayqueue<timertasklist> delayqueue;

  public timewheel(long tickms, int wheelsize, long currenttime, delayqueue<timertasklist> delayqueue) {
    this.currenttime = currenttime;
    this.tickms = tickms;
    this.wheelsize = wheelsize;
    this.interval = tickms * wheelsize;
    this.buckets = new timertasklist[wheelsize];
    this.currenttime = currenttime - (currenttime % tickms);
    this.delayqueue = delayqueue;
    for (int i = 0; i < wheelsize; i++) {
      buckets[i] = new timertasklist();
    }
  }

  public boolean add(timertaskentry entry) {
    long expiration = entry.getexpirems();
    if (expiration < tickms + currenttime) {
      //到期了
      return false;
    } else if (expiration < currenttime + interval) {
      //扔进当前时间轮的某个槽里,只有时间大于某个槽,才会放进去
      long virtualid = (expiration / tickms);
      int index = (int) (virtualid % wheelsize);
      timertasklist bucket = buckets[index];
      bucket.addtask(entry);
      //设置bucket 过期时间
      if (bucket.setexpiration(virtualid * tickms)) {
        //设好过期时间的bucket需要入队
        delayqueue.offer(bucket);
        return true;
      }
    } else {
      //当前轮不能满足,需要扔到上一轮
      timewheel timewheel = getoverflowwheel();
      return timewheel.add(entry);
    }
    return false;
  }


  private timewheel getoverflowwheel() {
    if (overflowwheel == null) {
      synchronized (this) {
        if (overflowwheel == null) {
          overflowwheel = new timewheel(interval, wheelsize, currenttime, delayqueue);
        }
      }
    }
    return overflowwheel;
  }

  /**
   * 推进指针
   *
   * @param timestamp
   */
  public void advancelock(long timestamp) {
    if (timestamp > currenttime + tickms) {
      currenttime = timestamp - (timestamp % tickms);
      if (overflowwheel != null) {
        this.getoverflowwheel().advancelock(timestamp);
      }
    }
  }
}

定时器接口

/**
 * 定时器
 * @author apdoer
 * @version 1.0
 * @date 2021/3/22 20:30
 */
public interface timer {

  /**
   * 添加一个新任务
   *
   * @param timertask
   */
  void add(timertask timertask);


  /**
   * 推动指针
   *
   * @param timeout
   */
  void advanceclock(long timeout);

  /**
   * 等待执行的任务
   *
   * @return
   */
  int size();

  /**
   * 关闭服务,剩下的无法被执行
   */
  void shutdown();
}

定时器实现

/**
 * @author apdoer
 * @version 1.0
 * @date 2021/3/22 20:33
 */
@slf4j
public class systemtimer implements timer {
  /**
   * 底层时间轮
   */
  private timewheel timewheel;
  /**
   * 一个timer只有一个延时队列
   */
  private delayqueue<timertasklist> delayqueue = new delayqueue<>();
  /**
   * 过期任务执行线程
   */
  private executorservice workerthreadpool;
  /**
   * 轮询delayqueue获取过期任务线程
   */
  private executorservice bossthreadpool;


  public systemtimer() {
    this.timewheel = new timewheel(1, 20, system.currenttimemillis(), delayqueue);
    this.workerthreadpool = executors.newfixedthreadpool(100);
    this.bossthreadpool = executors.newfixedthreadpool(1);
    //20ms推动一次时间轮运转
    this.bossthreadpool.submit(() -> {
      for (; ; ) {
        this.advanceclock(20);
      }
    });
  }


  public void addtimertaskentry(timertaskentry entry) {
    if (!timewheel.add(entry)) {
      //已经过期了
      timertask timertask = entry.gettimertask();
      log.info("=====任务:{} 已到期,准备执行============",timertask.getdesc());
      workerthreadpool.submit(timertask);
    }
  }

  @override
  public void add(timertask timertask) {
    log.info("=======添加任务开始====task:{}", timertask.getdesc());
    timertaskentry entry = new timertaskentry(timertask, timertask.getdelayms() + system.currenttimemillis());
    timertask.settimertaskentry(entry);
    addtimertaskentry(entry);
  }

  /**
   * 推动指针运转获取过期任务
   *
   * @param timeout 时间间隔
   * @return
   */
  @override
  public synchronized void advanceclock(long timeout) {
    try {
      timertasklist bucket = delayqueue.poll(timeout, timeunit.milliseconds);
      if (bucket != null) {
        //推进时间
        timewheel.advancelock(bucket.getexpiration());
        //执行过期任务(包含降级)
        bucket.clear(this::addtimertaskentry);
      }
    } catch (interruptedexception e) {
      log.error("advanceclock error");
    }
  }

  @override
  public int size() {
    //todo
    return 0;
  }

  @override
  public void shutdown() {
    this.bossthreadpool.shutdown();
    this.workerthreadpool.shutdown();
    this.timewheel = null;
  }
}

存储任务的环形链表

/**
 * @author apdoer
 * @version 1.0
 * @date 2021/3/22 19:26
 */
@data
@slf4j
class timertasklist implements delayed {
  /**
   * timertasklist 环形链表使用一个虚拟根节点root
   */
  private timertaskentry root = new timertaskentry(null, -1);

  {
    root.next = root;
    root.prev = root;
  }

  /**
   * bucket的过期时间
   */
  private atomiclong expiration = new atomiclong(-1l);

  public long getexpiration() {
    return expiration.get();
  }

  /**
   * 设置bucket的过期时间,设置成功返回true
   *
   * @param expirationms
   * @return
   */
  boolean setexpiration(long expirationms) {
    return expiration.getandset(expirationms) != expirationms;
  }

  public boolean addtask(timertaskentry entry) {
    boolean done = false;
    while (!done) {
      //如果timertaskentry已经在别的list中就先移除,同步代码块外面移除,避免死锁,一直到成功为止
      entry.remove();
      synchronized (this) {
        if (entry.timedtasklist == null) {
          //加到链表的末尾
          entry.timedtasklist = this;
          timertaskentry tail = root.prev;
          entry.prev = tail;
          entry.next = root;
          tail.next = entry;
          root.prev = entry;
          done = true;
        }
      }
    }
    return true;
  }

  /**
   * 从 timedtasklist 移除指定的 timertaskentry
   *
   * @param entry
   */
  public void remove(timertaskentry entry) {
    synchronized (this) {
      if (entry.gettimedtasklist().equals(this)) {
        entry.next.prev = entry.prev;
        entry.prev.next = entry.next;
        entry.next = null;
        entry.prev = null;
        entry.timedtasklist = null;
      }
    }
  }

  /**
   * 移除所有
   */
  public synchronized void clear(consumer<timertaskentry> entry) {
    timertaskentry head = root.next;
    while (!head.equals(root)) {
      remove(head);
      entry.accept(head);
      head = root.next;
    }
    expiration.set(-1l);
  }

  @override
  public long getdelay(timeunit unit) {
    return math.max(0, unit.convert(expiration.get() - system.currenttimemillis(), timeunit.milliseconds));
  }

  @override
  public int compareto(delayed o) {
    if (o instanceof timertasklist) {
      return long.compare(expiration.get(), ((timertasklist) o).expiration.get());
    }
    return 0;
  }
}

存储任务的容器entry

/**
 * @author apdoer
 * @version 1.0
 * @date 2021/3/22 19:26
 */
@data
class timertaskentry implements comparable<timertaskentry> {
  private timertask timertask;
  private long expirems;
  volatile timertasklist timedtasklist;
  timertaskentry next;
  timertaskentry prev;

  public timertaskentry(timertask timedtask, long expirems) {
    this.timertask = timedtask;
    this.expirems = expirems;
    this.next = null;
    this.prev = null;
  }

  void remove() {
    timertasklist currentlist = timedtasklist;
    while (currentlist != null) {
      currentlist.remove(this);
      currentlist = timedtasklist;
    }
  }

  @override
  public int compareto(timertaskentry o) {
    return ((int) (this.expirems - o.expirems));
  }
}

任务包装类(这里也可以将工作任务以线程变量的方式去传入)

@data
@slf4j
class timertask implements runnable {
  /**
   * 延时时间
   */
  private long delayms;
  /**
   * 任务所在的entry
   */
  private timertaskentry timertaskentry;

  private string desc;

  public timertask(string desc, long delayms) {
    this.desc = desc;
    this.delayms = delayms;
    this.timertaskentry = null;
  }

  public synchronized void settimertaskentry(timertaskentry entry) {
    // 如果这个timetask已经被一个已存在的timertaskentry持有,先移除一个
    if (timertaskentry != null && timertaskentry != entry) {
      timertaskentry.remove();
    }
    timertaskentry = entry;
  }

  public timertaskentry gettimertaskentry() {
    return timertaskentry;
  }

  @override
  public void run() {
    log.info("============={}任务执行", desc);
  }
}

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。