Java并發(fā)編程在各主流框架中怎么應(yīng)用

    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) {    //沒(méi)有使用ThreadLocal，在方法間共享session需要進(jìn)行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 {//定義一個(gè)靜態(tài)ThreadLocal變量session，就能夠保證各個(gè)線程有自己的一份，并且方法可以方便獲取，不用傳遞    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()方法 傳入當(dāng)前線程，獲得一個(gè)ThreadLocalMap對(duì)象，說(shuō)明每一個(gè)線程維護(hù)了     * 自己的一個(gè) map，保證讀取出來(lái)的value是自己線程的。     *     * ThreadLocalMap 是ThreadLocal靜態(tài)內(nèi)部類，存儲(chǔ)value的鍵值就是ThreadLocal本身。     *     * 因此可以斷定，每個(gè)線程維護(hù)一個(gè)ThreadLocalMap的鍵值對(duì)映射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);    }}

/** * 管理每個(gè)線程的資源和事務(wù)同步的中心幫助程序。供資源管理代碼使用，但不供典型應(yīng)用程序代碼使用。 * * 資源管理代碼應(yīng)該檢查線程綁定的資源，如，JDBC連接 或 Hibernate Sessions。 * 此類代碼通常不應(yīng)該將資源綁定到線程，因?yàn)檫@是事務(wù)管理器的職責(zé)。另一個(gè)選項(xiàng)是， * 如果事務(wù)同步處于活動(dòng)狀態(tài)，則在首次使用時(shí)延遲綁定，以執(zhí)行跨任意數(shù)量資源的事務(wù)。 */public abstract class TransactionSynchronizationManager {    /**     *  一般是一個(gè)線程持有一個(gè) 獨(dú)立的事務(wù)，以相互隔離地處理各自的事務(wù)。     *  所以這里使用了很多 ThreadLocal對(duì)象，為每個(gè)線程綁定 對(duì)應(yīng)的事務(wù)屬性及資源，     *  以便后續(xù)使用時(shí)能直接獲取。     */    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");    /**     * 為當(dāng)前線程 綁定 對(duì)應(yīng)的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();        // 如果當(dāng)前線程的 resources中，綁定的數(shù)據(jù)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() + "]");        }    }    /**     * 返回當(dāng)前線程綁定的所有資源     */    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 {    /**     * 核心線程數(shù)     * 當(dāng)向線程池提交一個(gè)任務(wù)時(shí)，若線程池已創(chuàng)建的線程數(shù)小于corePoolSize，即便此時(shí)存在空閑線程，     * 也會(huì)通過(guò)創(chuàng)建一個(gè)新線程來(lái)執(zhí)行該任務(wù)，直到已創(chuàng)建的線程數(shù)大于或等于corePoolSize     */    private volatile int corePoolSize;    /**     * 最大線程數(shù)     * 當(dāng)隊(duì)列滿了，且已創(chuàng)建的線程數(shù)小于maximumPoolSize，則線程池會(huì)創(chuàng)建新的線程來(lái)執(zhí)行任務(wù)。     * 另外，對(duì)于無(wú)界隊(duì)列，可忽略該參數(shù)     */    private volatile int maximumPoolSize;    /**     * 線程存活保持時(shí)間     * 當(dāng)線程池中線程數(shù) 超出核心線程數(shù)，且線程的空閑時(shí)間也超過(guò) keepAliveTime時(shí)，     * 那么這個(gè)線程就會(huì)被銷毀，直到線程池中的線程數(shù)小于等于核心線程數(shù)     */    private volatile long keepAliveTime;    /**     * 任務(wù)隊(duì)列     * 用于傳輸和保存等待執(zhí)行任務(wù)的阻塞隊(duì)列     */    private final BlockingQueue<Runnable> workQueue;    /**     * 線程工廠     * 用于創(chuàng)建新線程。threadFactory 創(chuàng)建的線程也是采用 new Thread() 方式，threadFactory     * 創(chuàng)建的線程名都具有統(tǒng)一的風(fēng)格：pool-m-thread-n（m為線程池的編號(hào)，n為線程池中線程的編號(hào)     */    private volatile ThreadFactory threadFactory;    /**     * 線程飽和策略     * 當(dāng)線程池和隊(duì)列都滿了，再加入的線程會(huì)執(zhí)行此策略     */    private volatile RejectedExecutionHandler handler;    /**     * 構(gòu)造方法提供了多種重載，但實(shí)際上都使用了最后一個(gè)重載 完成了實(shí)例化     */    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;    }    /**     * 執(zhí)行一個(gè)任務(wù)，但沒(méi)有返回值     */    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);    }    /**     * 提交一個(gè)線程任務(wù)，有返回值。該方法繼承自其父類 AbstractExecutorService，有多種重載，這是最常用的一個(gè)。     * 通過(guò)future.get()獲取返回值（阻塞直到任務(wù)執(zhí)行完）     */    public <T> Future<T> submit(Callable<T> task) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task);        execute(ftask);        return ftask;    }    /**     * 關(guān)閉線程池，不再接收新的任務(wù)，但會(huì)把已有的任務(wù)執(zhí)行完     */    public void shutdown() {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            checkShutdownAccess();            advanceRunState(SHUTDOWN);            interruptIdleWorkers();            onShutdown(); // hook for ScheduledThreadPoolExecutor        } finally {            mainLock.unlock();        }        tryTerminate();    }    /**     * 立即關(guān)閉線程池，已有的任務(wù)也會(huì)被拋棄     */    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> {    // 將給定元素設(shè)置到隊(duì)列中，如果設(shè)置成功返回true, 否則返回false。如果是往限定了長(zhǎng)度的隊(duì)列中設(shè)置值，推薦使用offer()方法。    boolean add(E e);    // 將給定的元素設(shè)置到隊(duì)列中，如果設(shè)置成功返回true, 否則返回false. e的值不能為空，否則拋出空指針異常。    boolean offer(E e);    // 將元素設(shè)置到隊(duì)列中，如果隊(duì)列中沒(méi)有多余的空間，該方法會(huì)一直阻塞，直到隊(duì)列中有多余的空間。    void put(E e) throws InterruptedException;    // 將給定元素在給定的時(shí)間內(nèi)設(shè)置到隊(duì)列中，如果設(shè)置成功返回true, 否則返回false.    boolean offer(E e, long timeout, TimeUnit unit)        throws InterruptedException;    // 從隊(duì)列中獲取值，如果隊(duì)列中沒(méi)有值，線程會(huì)一直阻塞，直到隊(duì)列中有值，并且該方法取得了該值。    E take() throws InterruptedException;    // 在給定的時(shí)間里，從隊(duì)列中獲取值，時(shí)間到了直接調(diào)用普通的 poll()方法，為null則直接返回null。    E poll(long timeout, TimeUnit unit)        throws InterruptedException;    // 獲取隊(duì)列中剩余的空間。    int remainingCapacity();    // 從隊(duì)列中移除指定的值。    boolean remove(Object o);    // 判斷隊(duì)列中是否擁有該值。    public boolean contains(Object o);    // 將隊(duì)列中值，全部移除，并發(fā)設(shè)置到給定的集合中。    int drainTo(Collection<? super E> c);    // 指定最多數(shù)量限制將隊(duì)列中值，全部移除，并發(fā)設(shè)置到給定的集合中。    int drainTo(Collection<? super E> c, int maxElements);}