package com.bing.sh.timewheel;

import java.util.*;

public abstract class TimeWheel 

    // 每个槽位对应一个时间刻度
    protected int bufferSize = 60;


    public TimeWheel() 
    

    public TimeWheel(int size) 
        this.bufferSize = size;
    


    protected abstract void pushRingData(int ringId, Object obj);

    protected abstract Collection<?> removeRingData(int ringId);


    public void pusData(int ringId, Object obj) 
        int index = countIndex(ringId);
        this.pushRingData(index, obj);
    

    public Collection<?> removeData(int ringId) 
        int index = countIndex(ringId);
        return this.removeRingData(index);
    


    public int countIndex(int ringId) 
        if (ringId < 0) 
            ringId = -ringId;
        
        return ringId % bufferSize;

使用ringId作为key来计算index, 存取前都要计算一遍，pushData用来存任务，removeData用来从时间轮里取任务。

2. 用数组实现

创建EntryWheel实现类，继承TimeWheel，可使用initBuffer()方法给时间轮初始化, 可用ReentrantReadWriteLock 加读写锁，也可不加，因为轮询的时候是单线程的取。

package com.bing.sh.timewheel;


import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class EntryWheel extends TimeWheel 

    /**
     * 采用数组的形式, 通过index来作为存储的key, value为一个List或者Set列表
     */
    private Object[] ringBuffer;

    ReentrantReadWriteLock lock = new ReentrantReadWriteLock();


    public EntryWheel() 
        super();
        initBuffer();
    

    public void initBuffer() 
        ringBuffer = new Object[bufferSize];
        for (int i = 0; i < bufferSize; i++) 
            ringBuffer[i] = new HashSet<>();
        
    


    public EntryWheel(int size) 
        super(size);
        initBuffer();
    

    @Override
    protected void pushRingData(int index, Object obj) 
        // 1. 根据ringId计算index
        // 2. 根据index存放obj
        try 
            lock.writeLock().lock();
            Set<Object> objects = (Set<Object>) ringBuffer[index];
            objects.add(obj);
            ringBuffer[index] = objects;
         finally 
            lock.writeLock().unlock();
        

    

    @Override
    protected Collection<?> removeRingData(int index) 
        try 
            lock.readLock().lock();
            Set<Object> objects = (Set<Object>) ringBuffer[index];
            ringBuffer[index] = new HashSet<>();
            return objects;
         finally 
            lock.readLock().unlock();

3. 用Map实现

用Map替换为数组。

package com.bing.sh.timewheel;


import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class MapWheel extends TimeWheel 

    private Map<Integer/** id**/, Set<Object>> ringDataMap;

    ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    public MapWheel() 
        ringDataMap = new ConcurrentHashMap<>(bufferSize);
        initMap();
    

    private void initMap() 
        for (int i = 0; i < bufferSize; i++) 
            Set<Object> objectSet = new HashSet<>();
            ringDataMap.put(i, objectSet);
        
    

    public MapWheel(int size) 
        super(size);
        ringDataMap = new ConcurrentHashMap<>(bufferSize);
        initMap();
    

    @Override
    protected void pushRingData(int ringId, Object obj) 
        try 
            lock.writeLock().lock();
            Set<Object> objects = ringDataMap.get(ringId);
            objects.add(obj);
         finally 
            lock.writeLock().unlock();
        
    

    @Override
    protected Collection<?> removeRingData(int ringId) 
        try 
            lock.readLock().lock();
            Set<Object> objects = ringDataMap.get(ringId);
            ringDataMap.put(ringId, new HashSet<>());
            return objects;
         finally 
            lock.readLock().unlock();

4. 测试数组

package com.bing.sh.timewheel;

import org.junit.Before;
import org.junit.Test;

import java.util.Random;
import java.util.Set;
import java.util.concurrent.TimeUnit;

public class EntryWheelTests 

    Random random;


    @Before
    public void setup() 
        random = new Random();
    


    @Test
    public void testEntryWheel() 
        EntryWheel entryWheel = new EntryWheel();
        new Thread(() -> 
            while (true) 
                entryWheel.pusData(random.nextInt(60), random.nextInt(1000000));
                try 
                    Thread.sleep(10);
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            
        ).start();

        new Thread(() -> 
            while (true) 
                int nowTime = (int) (System.currentTimeMillis() / 1000 % 60);
                Set<Object> objectSet = (Set<Object>) entryWheel.removeData(nowTime);
                System.out.println(nowTime + ">>>" + objectSet);
                try 
                    Thread.sleep(1000);
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            
        ).start();

        try 
            TimeUnit.SECONDS.sleep(10);
         catch (InterruptedException e) 
            e.printStackTrace();

打印结果: