牛逼！一文看懂 Java 并发编程在各主流框架中的应用

    private volatile int ioRatio = 50;

public class ThreadStopExample {    private static boolean stop;    public static void main(String[] args) throws InterruptedException {        Thread workThread = new Thread(new Runnable() {            public void run() {                int i= 0;                while (!stop) {                    i++;                    try{                        TimeUnit.SECONDS.sleep(1);                    } catch (InterruptedException e) {                        e.printStackTrace();                    }                }            }        });        workThread.start();        TimeUnit.SECONDS.sleep(3);        stop = true;    }}

    if (!stop)    While(true)        ......

    final long ioTime = System.nanoTime() - ioStartTime;    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);

    public void setIoRatio(int ioRatio) {        if (ioRatio <= 0 || ioRatio > 100) {            throw new IllegalArgumentException("ioRatio: " + ioRatio + " (expected: 0 < ioRatio <= 100)");        }        this.ioRatio = ioRatio;    }

public class SessionBean {    public static class Session {        private String id;        public String getId() {            return id;        }        public void setId(String id) {            this.id = id;        }    }    public Session createSession() {        return new Session();    }    public void setId(Session session, String id) {        session.setId(id);    }    public String getId(Session session) {        return session.getId();    }    public static void main(String[] args) {    //没有使用ThreadLocal，在方法间共享session需要进行session在方法间的传递        new Thread(() -> {            SessionBean bean = new SessionBean();            Session session = bean.createSession();            bean.setId(session, "susan");            System.out.println(bean.getId(session));        }).start();    }}

public class SessionBean {//定义一个静态ThreadLocal变量session，就能够保证各个线程有自己的一份，并且方法可以方便获取，不用传递    private static ThreadLocal<Session> session = new ThreadLocal<>();    public static class Session {        private String id;        public String getId() {            return id;        }        public void setId(String id) {            this.id = id;        }    }    public void createSession() {        session.set(new Session());    }    public void setId(String id) {        session.get().setId(id);    }    public String getId() {        return session.get().getId();    }    public static void main(String[] args) {        new Thread(() -> {            SessionBean bean = new SessionBean();            bean.createSession();            bean.setId("susan");            System.out.println(bean.getId());        }).start();    }}

public class ThreadLocal<T> {    /**     * 下面的 getMap()方法 传入当前线程，获得一个ThreadLocalMap对象，说明每一个线程维护了     * 自己的一个 map，保证读取出来的value是自己线程的。     *     * ThreadLocalMap 是ThreadLocal静态内部类，存储value的键值就是ThreadLocal本身。     *     * 因此可以断定，每个线程维护一个ThreadLocalMap的键值对映射Map。不同线程的Map的 key值 是一样的，     * 都是ThreadLocal，但 value 是不同的。     */    public T get() {        Thread t = Thread.currentThread();        ThreadLocalMap map = getMap(t);        if (map != null) {            ThreadLocalMap.Entry e = map.getEntry(this);            if (e != null) {                @SuppressWarnings("unchecked")                T result = (T)e.value;                return result;            }        }        return setInitialValue();    }    public void set(T value) {        Thread t = Thread.currentThread();        ThreadLocalMap map = getMap(t);        if (map != null)            map.set(this, value);        else            createMap(t, value);    }}

/** * 管理每个线程的资源和事务同步的中心帮助程序。供资源管理代码使用，但不供典型应用程序代码使用。 * * 资源管理代码应该检查线程绑定的资源，如，JDBC连接 或 Hibernate Sessions。 * 此类代码通常不应该将资源绑定到线程，因为这是事务管理器的职责。另一个选项是， * 如果事务同步处于活动状态，则在首次使用时延迟绑定，以执行跨任意数量资源的事务。 */public abstract class TransactionSynchronizationManager {    /**     *  一般是一个线程持有一个 独立的事务，以相互隔离地处理各自的事务。     *  所以这里使用了很多 ThreadLocal对象，为每个线程绑定 对应的事务属性及资源，     *  以便后续使用时能直接获取。     */    private static final ThreadLocal<Map<Object, Object>> resources =            new NamedThreadLocal<Map<Object, Object>>("Transactional resources");    private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =            new NamedThreadLocal<Set<TransactionSynchronization>>("Transaction synchronizations");    private static final ThreadLocal<String> currentTransactionName =            new NamedThreadLocal<String>("Current transaction name");    private static final ThreadLocal<Boolean> currentTransactionReadOnly =            new NamedThreadLocal<Boolean>("Current transaction read-only status");    private static final ThreadLocal<Integer> currentTransactionIsolationLevel =            new NamedThreadLocal<Integer>("Current transaction isolation level");    private static final ThreadLocal<Boolean> actualTransactionActive =            new NamedThreadLocal<Boolean>("Actual transaction active");    /**     * 为当前线程 绑定 对应的resource资源     */    public static void bindResource(Object key, Object value) throws IllegalStateException {        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);        Assert.notNull(value, "Value must not be null");        Map<Object, Object> map = resources.get();        // 如果当前线程的 resources中，绑定的数据map为空，则为 resources 绑定 map        if (map == null) {            map = new HashMap<Object, Object>();            resources.set(map);        }        Object oldValue = map.put(actualKey, value);        if (oldValue instanceof ResourceHolder && ((ResourceHolder) oldValue).isVoid()) {            oldValue = null;        }        if (oldValue != null) {            throw new IllegalStateException("Already value [" + oldValue + "] for key [" +                    actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");        }        if (logger.isTraceEnabled()) {            logger.trace("Bound value [" + value + "] for key [" + actualKey + "] to thread [" +                    Thread.currentThread().getName() + "]");        }    }    /**     * 返回当前线程绑定的所有资源     */    public static Map<Object, Object> getResourceMap() {        Map<Object, Object> map = resources.get();        return (map != null ? Collections.unmodifiableMap(map) : Collections.emptyMap());    }}

public class SqlSessionManager implements SqlSessionFactory, SqlSession {  private final ThreadLocal<SqlSession> localSqlSession = new ThreadLocal<>();  public void startManagedSession() {    this.localSqlSession.set(openSession());  }  public void startManagedSession(boolean autoCommit) {    this.localSqlSession.set(openSession(autoCommit));  }  public void startManagedSession(Connection connection) {    this.localSqlSession.set(openSession(connection));  }  public void startManagedSession(TransactionIsolationLevel level) {    this.localSqlSession.set(openSession(level));  }  public void startManagedSession(ExecutorType execType) {    this.localSqlSession.set(openSession(execType));  }  public void startManagedSession(ExecutorType execType, boolean autoCommit) {    this.localSqlSession.set(openSession(execType, autoCommit));  }  public void startManagedSession(ExecutorType execType, TransactionIsolationLevel level) {    this.localSqlSession.set(openSession(execType, level));  }  public void startManagedSession(ExecutorType execType, Connection connection) {    this.localSqlSession.set(openSession(execType, connection));  }  public boolean isManagedSessionStarted() {    return this.localSqlSession.get() != null;  }  @Override  public Connection getConnection() {    final SqlSession sqlSession = localSqlSession.get();    if (sqlSession == null) {      throw new SqlSessionException("Error:  Cannot get connection.  No managed session is started.");    }    return sqlSession.getConnection();  }  @Override  public void clearCache() {    final SqlSession sqlSession = localSqlSession.get();    if (sqlSession == null) {      throw new SqlSessionException("Error:  Cannot clear the cache.  No managed session is started.");    }    sqlSession.clearCache();  }  @Override  public void commit() {    final SqlSession sqlSession = localSqlSession.get();    if (sqlSession == null) {      throw new SqlSessionException("Error:  Cannot commit.  No managed session is started.");    }    sqlSession.commit();  }  @Override  public void commit(boolean force) {    final SqlSession sqlSession = localSqlSession.get();    if (sqlSession == null) {      throw new SqlSessionException("Error:  Cannot commit.  No managed session is started.");    }    sqlSession.commit(force);  }  @Override  public void rollback() {    final SqlSession sqlSession = localSqlSession.get();    if (sqlSession == null) {      throw new SqlSessionException("Error:  Cannot rollback.  No managed session is started.");    }    sqlSession.rollback();  }  @Override  public void rollback(boolean force) {    final SqlSession sqlSession = localSqlSession.get();    if (sqlSession == null) {      throw new SqlSessionException("Error:  Cannot rollback.  No managed session is started.");    }    sqlSession.rollback(force);  }  @Override  public List<BatchResult> flushStatements() {    final SqlSession sqlSession = localSqlSession.get();    if (sqlSession == null) {      throw new SqlSessionException("Error:  Cannot rollback.  No managed session is started.");    }    return sqlSession.flushStatements();  }  @Override  public void close() {    final SqlSession sqlSession = localSqlSession.get();    if (sqlSession == null) {      throw new SqlSessionException("Error:  Cannot close.  No managed session is started.");    }    try {      sqlSession.close();    } finally {      localSqlSession.set(null);    }  }}

public class ThreadPoolExecutor extends AbstractExecutorService {    /**     * 核心线程数     * 当向线程池提交一个任务时，若线程池已创建的线程数小于corePoolSize，即便此时存在空闲线程，     * 也会通过创建一个新线程来执行该任务，直到已创建的线程数大于或等于corePoolSize     */    private volatile int corePoolSize;    /**     * 最大线程数     * 当队列满了，且已创建的线程数小于maximumPoolSize，则线程池会创建新的线程来执行任务。     * 另外，对于无界队列，可忽略该参数     */    private volatile int maximumPoolSize;    /**     * 线程存活保持时间     * 当线程池中线程数 超出核心线程数，且线程的空闲时间也超过 keepAliveTime时，     * 那么这个线程就会被销毁，直到线程池中的线程数小于等于核心线程数     */    private volatile long keepAliveTime;    /**     * 任务队列     * 用于传输和保存等待执行任务的阻塞队列     */    private final BlockingQueue<Runnable> workQueue;    /**     * 线程工厂     * 用于创建新线程。threadFactory 创建的线程也是采用 new Thread() 方式，threadFactory     * 创建的线程名都具有统一的风格：pool-m-thread-n（m为线程池的编号，n为线程池中线程的编号     */    private volatile ThreadFactory threadFactory;    /**     * 线程饱和策略     * 当线程池和队列都满了，再加入的线程会执行此策略     */    private volatile RejectedExecutionHandler handler;    /**     * 构造方法提供了多种重载，但实际上都使用了最后一个重载 完成了实例化     */    public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue) {        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             Executors.defaultThreadFactory(), defaultHandler);    }    public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue,                              ThreadFactory threadFactory) {        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             threadFactory, defaultHandler);    }    public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue,                              RejectedExecutionHandler handler) {        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             Executors.defaultThreadFactory(), handler);    }    public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue,                              ThreadFactory threadFactory,                              RejectedExecutionHandler handler) {        if (corePoolSize < 0 ||            maximumPoolSize <= 0 ||            maximumPoolSize < corePoolSize ||            keepAliveTime < 0)            throw new IllegalArgumentException();        if (workQueue == null || threadFactory == null || handler == null)            throw new NullPointerException();        this.corePoolSize = corePoolSize;        this.maximumPoolSize = maximumPoolSize;        this.workQueue = workQueue;        this.keepAliveTime = unit.toNanos(keepAliveTime);        this.threadFactory = threadFactory;        this.handler = handler;    }    /**     * 执行一个任务，但没有返回值     */    public void execute(Runnable command) {        if (command == null)            throw new NullPointerException();        int c = ctl.get();        if (workerCountOf(c) < corePoolSize) {            if (addWorker(command, true))                return;            c = ctl.get();        }        if (isRunning(c) && workQueue.offer(command)) {            int recheck = ctl.get();            if (! isRunning(recheck) && remove(command))                reject(command);            else if (workerCountOf(recheck) == 0)                addWorker(null, false);        }        else if (!addWorker(command, false))            reject(command);    }    /**     * 提交一个线程任务，有返回值。该方法继承自其父类 AbstractExecutorService，有多种重载，这是最常用的一个。     * 通过future.get()获取返回值（阻塞直到任务执行完）     */    public <T> Future<T> submit(Callable<T> task) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task);        execute(ftask);        return ftask;    }    /**     * 关闭线程池，不再接收新的任务，但会把已有的任务执行完     */    public void shutdown() {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            checkShutdownAccess();            advanceRunState(SHUTDOWN);            interruptIdleWorkers();            onShutdown(); // hook for ScheduledThreadPoolExecutor        } finally {            mainLock.unlock();        }        tryTerminate();    }    /**     * 立即关闭线程池，已有的任务也会被抛弃     */    public List<Runnable> shutdownNow() {        List<Runnable> tasks;        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            checkShutdownAccess();            advanceRunState(STOP);            interruptWorkers();            tasks = drainQueue();        } finally {            mainLock.unlock();        }        tryTerminate();        return tasks;    }    public boolean isShutdown() {        return ! isRunning(ctl.get());    }}

public interface BlockingQueue<E> extends Queue<E> {    // 将给定元素设置到队列中，如果设置成功返回true, 否则返回false。如果是往限定了长度的队列中设置值，推荐使用offer()方法。    boolean add(E e);    // 将给定的元素设置到队列中，如果设置成功返回true, 否则返回false. e的值不能为空，否则抛出空指针异常。    boolean offer(E e);    // 将元素设置到队列中，如果队列中没有多余的空间，该方法会一直阻塞，直到队列中有多余的空间。    void put(E e) throws InterruptedException;    // 将给定元素在给定的时间内设置到队列中，如果设置成功返回true, 否则返回false.    boolean offer(E e, long timeout, TimeUnit unit)        throws InterruptedException;    // 从队列中获取值，如果队列中没有值，线程会一直阻塞，直到队列中有值，并且该方法取得了该值。    E take() throws InterruptedException;    // 在给定的时间里，从队列中获取值，时间到了直接调用普通的 poll()方法，为null则直接返回null。    E poll(long timeout, TimeUnit unit)        throws InterruptedException;    // 获取队列中剩余的空间。    int remainingCapacity();    // 从队列中移除指定的值。    boolean remove(Object o);    // 判断队列中是否拥有该值。    public boolean contains(Object o);    // 将队列中值，全部移除，并发设置到给定的集合中。    int drainTo(Collection super E> c);    // 指定最多数量限制将队列中值，全部移除，并发设置到给定的集合中。    int drainTo(Collection super E> c, int maxElements);}