event包

EventUtils

事件绑定和监听的工具类

//给事件源绑定监听器
    public static <L> void addEventListener(final Object eventSource, final Class<L> listenerType, final L listener) 

//绑定事件到某个对象的方法，即发生事件后，可以执行某个对象的方法
     public static <L> void bindEventsToMethod(final Object target, final String methodName, final Object eventSource,
            final Class<L> listenerType, final String... eventTypes) 

concurrent包

AtomicInitializer

用于初始化对象，调用get()方法后，要不只获得一个单例，要不重新出事化一个对象。

AtomicSafeInitializer

用于初始化对象，但是它能保证只初始化一次。

BackgroundInitializer

用于在后台初始化资源，调用start()方法后，可以做其他事情，直到调用get()方法，如果已经初始化完毕，则可以马上获取，否则线程会被阻塞

CallableBackgroundInitializer

带返回值的初始化

// a Callable that performs a complex computation
   Callable<Integer> computationCallable = new MyComputationCallable();
   // setup the background initializer
   CallableBackgroundInitializer<Integer> initializer =
       new CallableBackgroundInitializer(computationCallable);
   initializer.start();
   // Now do some other things. Initialization runs in a parallel thread
   ...
   // Wait for the end of initialization and access the result
   Integer result = initializer.get();

BasicThreadFactory

线程工厂类

 BasicThreadFactory factory = new BasicThreadFactory.Builder()
       .namingPattern("workerthread-%d")
       .daemon(true)
       .priority(Thread.MAX_PRIORITY)
       .build();
   ExecutorService exec = Executors.newSingleThreadExecutor(factory);

ThresholdCircuitBreaker

阈值电子回路。传递一个增量值，如果到达阈值，则表示开启

long threshold = 10L;
   ThresholdCircuitBreaker breaker = new ThresholdCircuitBreaker(10L);
   ...
   public void handleRequest(Request request) {
       long memoryUsed = estimateMemoryUsage(request);
       if (breaker.incrementAndCheckState(memoryUsed)) {
           // actually handle this request
       } else {
           // do something else, e.g. send an error code
       }
   }

AbstractCircuitBreaker

抽象电子回路。提供一些基础实现，另外实现了事件监听，即到达阈值后会告诉监听方。关键代码如下

//添加监听器
 public void addChangeListener(final PropertyChangeListener listener) {
        changeSupport.addPropertyChangeListener(listener);
    }

//状态变更，告诉监听方
 protected void changeState(final State newState) {
        if (state.compareAndSet(newState.oppositeState(), newState)) {
            changeSupport.firePropertyChange(PROPERTY_NAME, !isOpen(newState), isOpen(newState));
        }
    }

ConcurrentUtils

并发工具类。提供一些方法将检查型异常转为非检查型异常。

//不存在则初始化并放入map
 public static <K, V> V createIfAbsent(final ConcurrentMap<K, V> map, final K key,
            final ConcurrentInitializer<V> init) throws ConcurrentException {
        if (map == null || init == null) {
            return null;
        }

        final V value = map.get(key);
        if (value == null) {
            return putIfAbsent(map, key, init.get());
        }
        return value;
    }

ConstantInitializer

调用get()方法，总会返回同一个对象。不用使用同步。

Memoizer

记忆器? 为什么拼写不正确？
主要功能是缓存一个计算结果，下次来取，直接获取结果，不用重新计算

public O compute(final I arg) throws InterruptedException {
        while (true) {
            Future<O> future = cache.get(arg);
            if (future == null) {
                final Callable<O> eval = () -> computable.compute(arg);
                final FutureTask<O> futureTask = new FutureTask<>(eval);
                future = cache.putIfAbsent(arg, futureTask);
                //如果future是null，表示第一次执行
                if (future == null) {
                    future = futureTask;
                //第一次执行
                    futureTask.run();
                }
            }
            try {
                return future.get();
            } catch (final CancellationException e) {
                cache.remove(arg, future);
            } catch (final ExecutionException e) {
                if (recalculate) {
                    cache.remove(arg, future);
                }

                throw launderException(e.getCause());
            }
        }
    }

MultiBackgroundInitializer

名字叫多重后台初始化器。

protected MultiBackgroundInitializerResults initialize() throws Exception {
        Map<String, BackgroundInitializer<?>> inits;
        synchronized (this) {
            // create a snapshot to operate on
            inits = new HashMap<>(
                    childInitializers);
        }

        // start the child initializers
        final ExecutorService exec = getActiveExecutor();
        for (final BackgroundInitializer<?> bi : inits.values()) {
            if (bi.getExternalExecutor() == null) {
                // share the executor service if necessary
                bi.setExternalExecutor(exec);
            }
// 异步提交，提交完就会返回
            bi.start();
        }

        // collect the results
        final Map<String, Object> results = new HashMap<>();
        final Map<String, ConcurrentException> excepts = new HashMap<>();
        for (final Map.Entry<String, BackgroundInitializer<?>> e : inits.entrySet()) {
            try {
//可是在e.getValue().get()的时候会阻塞，其中一个没执行完，就会等待执行完，然后等待下一个执行完
                results.put(e.getKey(), e.getValue().get());
            } catch (final ConcurrentException cex) {
                excepts.put(e.getKey(), cex);
            }
        }

        return new MultiBackgroundInitializerResults(inits, results, excepts);
    }

TimedSemaphore

时间的信号量。超过时间，会自动恢复信号量。不用调用release()来释放信号量

TimedSemaphore sem = new TimedSemaphore(1, TimeUnit.SECOND, 10);
   StatisticsThread thread = new StatisticsThread(sem);
   thread.start();


 public class StatisticsThread extends Thread {
       //信号量用来限制数据库负载
       private final TimedSemaphore semaphore;

       public StatisticsThread(TimedSemaphore timedSemaphore) {
           semaphore = timedSemaphore;
       }

       public void run() {
           try {
               while (true) {
//获取信号量
                   semaphore.acquire();   // limit database load
//执行查询
                   performQuery();        // issue a query
               }
           } catch(InterruptedException) {
               // fall through
           }
       }
       ...
   }

实现逻辑是：起一个定时器，定时重置acquireCount变量（表示已被使用的信号量）为0，然后其他线程就可以拿到信号量，不用一直被阻塞住（因为之前在acquire()方法里调用了wait()方法）

//起一个定时器，定时执行endOfPeriod
 protected ScheduledFuture<?> startTimer() {
        return getExecutorService().scheduleAtFixedRate(() -> endOfPeriod(), getPeriod(), getPeriod(), getUnit());
    }

//使用synchronized来保证acquireCount的同步访问
 synchronized void endOfPeriod() {
        lastCallsPerPeriod = acquireCount;
        totalAcquireCount += acquireCount;
        periodCount++;
        acquireCount = 0;
//唤醒其他已阻塞的线程
        notifyAll();
    }