ID Generator

    关于ID Generator，想必大多数项目都有应用。跟按需生成ID相比，预生成一定数量的ID并加以缓存的方式更有助于提升性能。预生成ID的时机，通常是在发现缓存的ID用尽的时候。这种方式有个缺陷，即从调用者的角度来看每次取得ID所花费的时间可能并不均等。

    如果应用要求每次取得ID时都要尽可能的快且时间均等，那么ID Generator可以在发现缓存的ID用尽之前进行预生成，保持缓存中总是有可用的ID（例如每次预生成100个ID，在缓存中只剩下20个ID的时机再次进行预生成）。这种做法面临的主要问题就是并发控制。以及预生成时如果抛出异常， 那么该异常如何传播给请求ID的调用者（在预生成的时候，可能并没有没有线程在请求ID）。

    以下是一段笔者在项目中采用的代码片段, 首先是IdGenerator接口，以及几个跟ID生成相关的异常:

public interface IdGenerator<T> {
    T nextId();
}

public class GenerationException extends NestableRuntimeException {
    ...
}

public class GenerationInterruptedException extends GenerationException {
    ...
}

public class GenerationTimeoutException extends GenerationException {
    ...
}

 

    接下来是跟ID预生成相关的接口IdLoader

public interface IdLoader<T> {
    List<T> load() throws Exception;
}

 

    CachedIdGenerator是IdGenerator的一个实现，用于对ID缓存的管理，以及在适当的时机调用IdLoader进行ID的预生成。其最重要的一个属性是preLoadThreshold，通过调整这个属性的值，便可以控制ID预生成的时机：

• 负值：在发现缓存中的已经没有可用的ID进行分配时生成。生成ID的过程中，调用nextId()方法的线程会被阻塞。
• 0：在分配了缓存中的最后一个ID时生成，由于此时缓存中有最后一个可用的ID，因此调用nextId()方法的线程不会被阻塞。
• 正值：在缓存中的ID个数小于还有该阀值时生成，调用nextId()方法的线程不会被阻塞。

    关于CachedIdGenerator的并发控制：

• 如果多个线程同时调用nextId()方法，那么通过排他锁进行控制。
• CachedIdGenerator内部使用一个单独的线程（以下成load线程）进行ID预生成（即调用IdLoader的load()方法的线程）。
• 如果某次对nextId()方法的调用触发了ID预生成，那么在ID预生成结束之前该线程一直被阻塞。如果此时还有其它线程调用nextId()方法，那么这些线程也会被阻塞，即不会同时重复触发ID预生成。
• 如果IdLoader的load()方法抛出异常，那么CachedIdGenerator对该异常的处理区分以下两个场景：1 如果此时有调用nextId()方法的线程被阻塞，那么该异常会被传播给调用nextId()方法的线程；2 如果此时没有调用nextId()方法的线程被阻塞，那么CachedIdGenerator会将该异常记录到日志中，同时将 preLoadThreshold 自动调整为-1。在最终缓存的中的ID用尽时，才会再次触发ID预生成，此时调用nextId()方法的线程一定被阻塞，如果IdLoader的load 方法再次抛出异常，那么这个异常会被传播给调用nextId()方法的线程。

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CachedIdGenerator<T> implements IdGenerator<T> {
    //
    private static final Logger LOGGER = LoggerFactory.getLogger(CachedIdGenerator.class);
    
    //
    private final AsyncIdLoader loader;
    private final ReentrantLock lock = new ReentrantLock();
    private final LinkedList<T> cache = new LinkedList<T>();
    private final AtomicBoolean verbose = new AtomicBoolean(false);
    private final AtomicInteger preLoadThreshold = new AtomicInteger(0);
    
    /**
     * 
     */
    public CachedIdGenerator(IdLoader<T> loader) {
        this(loader, 0);
    }
    
    public CachedIdGenerator(IdLoader<T> loader, int preLoadThreshold) {
        this.loader = new AsyncIdLoader(loader);
        this.preLoadThreshold.set(preLoadThreshold);
    }
    
    /**
     * 
     */
    public final T nextId() {
        try {
            while(true) {
                final Future<T> f = next();
                final T r = f.get();
                if(r != null) {
                    return r;
                }
            }
        } catch (InterruptedException e) {
            throw new GenerationInterruptedException(e);
        } catch(ExecutionException e) {
            throw new GenerationException(((ExecutionException)e).getCause());
        } catch(Exception e) {
            throw new GenerationException(e);
        }
    }
    
    public final T nextId(long timeout, TimeUnit unit) {
        try {
            while(true) {
                //
                final long now = System.nanoTime();
                final Future<T> f = next();
                final T r = f.get(timeout, unit);
                if(r != null) {
                    return r;
                }
                
                //
                timeout -= unit.convert(System.nanoTime() - now, TimeUnit.NANOSECONDS);
                if(timeout < 0) {
                    throw new TimeoutException();
                }
            }
        } catch(TimeoutException e) {
            throw new GenerationTimeoutException(e);
        } catch (InterruptedException e) {
            throw new GenerationInterruptedException(e);
        } catch(ExecutionException e) {
            throw new GenerationException(((ExecutionException)e).getCause());
        } catch(Exception e) {
            throw new GenerationException(e);
        }
    }

    public boolean isVerbose() {
        return verbose.get();
    }
    
    public void setVerbose(boolean verbose) {
        this.verbose.set(verbose);
    }

    public void disablePreLoad() {
        this.preLoadThreshold.set(-1);
    }
    
    public int getPreLoadThreshold() {
        return preLoadThreshold.get();
    }
    
    public void setPreLoadThreshold(int threshold) {
        this.preLoadThreshold.set(threshold);
    }
    
    /**
     * 
     */
    protected Future<T> next() {
        Future<T> r = null;
        this.lock.lock();
        try {
            //
            if(!this.cache.isEmpty()) {
                final DummyFuture<T> df = new DummyFuture<T>();
                df.setResult(this.cache.removeFirst());
                r = df;
            }
            
            //
            if(r == null) {
                r = this.loader.load();
            } else if(cache.size() <= this.preLoadThreshold.get()) {
                this.loader.load();
            }
        } finally {
            this.lock.unlock();
        }
        return r;
    }
    
    /**
     * 
     */
    protected class AsyncIdLoader {
        //
        private final IdLoader<T> loader;
        private final ExecutorService executor;
        private final AtomicReference<Future<T>> result = new AtomicReference<Future<T>>();
        
        /**
         * 
         */
        public AsyncIdLoader(IdLoader<T> loader) {
            this.loader = loader;
            this.executor = Executors.newFixedThreadPool(1, new XThreadFactory(getClass().getSimpleName(), true));
        }
        
        /**
         * 
         */
        public Future<T> load() {
            //
            final Future<T> current = this.result.get();
            if(current != null) { // Loading is in progress
                return current;
            }
            
            //
            final Future<T> r = this.executor.submit(new Callable<T>() {

                public T call() throws Exception {
                    List<T> ids = null;
                    try {
                        //
                        if(isVerbose() && LOGGER.isInfoEnabled()) {
                            LOGGER.info("start to load ids, loader: {}", loader);
                        }
                        
                        //
                        ids = loader.load();
                        
                        //
                        if(isVerbose() && LOGGER.isInfoEnabled()) {
                            LOGGER.info("ids were successfully loaded, count: {}, loader: {}", (ids == null ? 0 : ids.size()), loader);
                        }
                        return null;
                    } catch (Exception e) {
                        disablePreLoad();
                        LOGGER.warn("unhandled exception in id loader: " + loader + ", pre-loading was disabled", e);
                        throw e;
                    } finally {
                        //
                        lock.lock();
                        try {
                            result.set(null);
                            if(ids != null && ids.size() > 0) {
                                cache.addAll(ids);
                            }
                        } finally {
                            lock.unlock();
                        }
                    }
                }
            });
            
            //
            this.result.set(r);
            return r;
        }
    } 
}

    最后是CachedIdGenerator用到的几个工具类：

 

public class XThreadFactory implements ThreadFactory {
    //
    private static final Logger LOGGER = LoggerFactory.getLogger(XThreadFactory.class);
    
    //
    private String name;
    private boolean daemon;
    private UncaughtExceptionHandler uncaughtExceptionHandler;
    private final ConcurrentHashMap<String, AtomicLong> sequences;
    
    
    /**
     * 
     */
    public XThreadFactory() {
        this(null, false, null);
    }
    
    public XThreadFactory(String name) {
        this(name, false, null);
    }
    
    public XThreadFactory(String name, boolean daemon) {
        this(name, daemon, null);
    }
    
    public XThreadFactory(String name, boolean daemon, UncaughtExceptionHandler handler) {
        this.name = name;
        this.daemon = daemon;
        this.uncaughtExceptionHandler = handler;
        this.sequences = new ConcurrentHashMap<String, AtomicLong>();
    }

    /**
     * 
     */
    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }
    
    public boolean isDaemon() {
        return daemon;
    }
    
    public void setDaemon(boolean daemon) {
        this.daemon = daemon;
    }

    public UncaughtExceptionHandler getUncaughtExceptionHandler() {
        return uncaughtExceptionHandler;
    }
    
    public void setUncaughtExceptionHandler(UncaughtExceptionHandler handler) {
        this.uncaughtExceptionHandler = handler;
    }
    
    /**
     * 
     */
    public Thread newThread(Runnable r) {
        //
        Thread t = new Thread(r);
        t.setDaemon(this.daemon);
        
        //
        String prefix = this.name;
        if(prefix == null || prefix.equals("")) {
            prefix = getInvoker(2);
        }
        t.setName(prefix + "-" + getSequence(prefix));
        
        //
        if(this.uncaughtExceptionHandler != null) {
            t.setUncaughtExceptionHandler(this.uncaughtExceptionHandler);
        } else {
            t.setUncaughtExceptionHandler(new UncaughtExceptionHandler() {
                public void uncaughtException(Thread t, Throwable e) {
                    LOGGER.error("unhandled exception in thread: " + t.getId() + ":" + t.getName(), e);
                }
            });
        }
        
        //
        return t;
    }

    /**
     * 
     */
    private String getInvoker(int depth) {
        Exception e = new Exception();
        StackTraceElement[] stes = e.getStackTrace();
        if(stes.length > depth) {
            return ClassUtils.getShortClassName(stes[depth].getClassName());
        }
        return getClass().getSimpleName();
    }
    
    private long getSequence(String invoker) {
        AtomicLong r = this.sequences.get(invoker);
        if(r == null) {
            r = new AtomicLong(0);
            AtomicLong previous = this.sequences.putIfAbsent(invoker, r);
            if(previous != null) {
                r = previous;
            }
        }
        return r.incrementAndGet();
    }
}

public class DummyFuture<V> implements Future<V> {
    //
    private volatile V result;
    private volatile Throwable throwable;

    /**
     * 
     */
    public boolean isDone() {
        return true;
    }
    
    public boolean isCancelled() {
        return false;
    }
    
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    public V get() throws InterruptedException, ExecutionException {
        if(throwable != null) {
            throw new ExecutionException(throwable);
        } else {
            return result;
        }
    }

    public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return get();
    }
    
    /**
     * 
     */
    public void setResult(V result) {
        this.result = result;
    }

    public void setThrowable(Throwable throwable) {
        this.throwable = throwable;
    }
}

public class XFuture<V> implements Future<V> {
    //
    private volatile V result;
    private volatile Object id;
    private volatile Throwable throwable;
    private CountDownLatch done = new CountDownLatch(1);

    /**
     * 
     */
    public XFuture() {
        this(null);
    }
    
    public XFuture(Object id) {
        this.id = id;
    }

    /**
     * 
     */
    public boolean isDone() {
        return done.getCount() != 1;
    }
    
    public boolean isCancelled() {
        return false;
    }
    
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    public V get() throws InterruptedException, ExecutionException {
        //
        done.await();
        
        //
        if(throwable != null) {
            throw new ExecutionException(throwable);
        } else {
            return result;
        }
    }

    public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        //
        if(!done.await(timeout, unit)) {
            throw new TimeoutException("failed to get in timeout: " + timeout + ", unit: " + unit);
        }
        
        //
        if(throwable != null) {
            throw new ExecutionException(throwable);
        } else {
            return result;
        }
    }
    
    /**
     * 
     */
    public Object getId() {
        return id;
    }
    
    public void setResult(V result) {
        this.result = result;
        this.done.countDown();
    }
    
    public void setThrowable(Throwable throwable) {
        this.throwable = throwable;
        this.done.countDown();
    }
}

评论

2 楼 whitesock 2010-06-15  

比如说通过数据库的某个值生成ID
1 每次对其加1，生成100个ID至少需要100次数据库访问。
2 每次对其加100， 那么生成100个ID只需要1次数据库访问。

1 楼 hellojinjie 2010-06-15  

恕我愚钝，id 的生成缓不缓存和性能有什么关系呢？