JAVA線程池應用的DEMO

時間 2019-11-18

標籤 java 線程應用 demo 欄目 Java 简体版

原文原文鏈接

在作不少高併發應用的時候，單線程的瓶頸已經知足不了咱們的需求，此時使用多線程來提升處理速度已是比較常規的方案了。在使用多線程的時候，咱們可使用線程池來管理咱們的線程，至於使用線程池的優勢就很少說了。java

Java線程池提及來也簡單，簡單說下繼承關係：
ThreadPoolExecutor extends AbstractExecutorService implements ExecutorService extends Executor緩存

還有一個支持延時執行線程和能夠重複執行線程的實現類：
ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService多線程

你們把這些類中的相關方法弄清楚，使用線程池就不在話下了。其實弄清楚裏面各個方法的功能也就夠了。
最重要的仍是在實踐中總結經驗，企業須要的是能實際解決問題的人。併發

下面是我寫的一個例子，包括3個Java文件，分別是：
ExecutorServiceFactory.java
ExecutorProcessPool.java
ExecutorTest.javadom

一、ExecutorServiceFactory異步

package test;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 線程池構造工廠
 *
 * @author allan
 * @date   2017年3月20日
 */
public class ExecutorServiceFactory {
    private static ExecutorServiceFactory executorFactory = new ExecutorServiceFactory();
    /**
     * 定時任務線程池
     */
    private ExecutorService executors;

    private ExecutorServiceFactory() {
    }

    /**
     * 獲取ExecutorServiceFactory
     * 
     * @return
     */
    public static ExecutorServiceFactory getInstance() {
        return executorFactory;
    }

    /**
     * 建立一個線程池，它可安排在給定延遲後運行命令或者按期地執行。
     * 
     * @return
     */
    public ExecutorService createScheduledThreadPool() {
        // CPU個數
        int availableProcessors = Runtime.getRuntime().availableProcessors();
        // 建立
        executors = Executors.newScheduledThreadPool(availableProcessors * 10, getThreadFactory());
        return executors;
    }

    /**
     * 建立一個使用單個 worker 線程的
     * Executor，以無界隊列方式來運行該線程。（注意，若是由於在關閉前的執行期間出現失敗而終止了此單個線程，
     * 那麼若是須要，一個新線程將代替它執行後續的任務）。可保證順序地執行各個任務，而且在任意給定的時間不會有多個線程是活動的。與其餘等效的
     * newFixedThreadPool(1) 不一樣，可保證無需從新配置此方法所返回的執行程序便可使用其餘的線程。
     * 
     * @return
     */
    public ExecutorService createSingleThreadExecutor() {
        // 建立
        executors = Executors.newSingleThreadExecutor(getThreadFactory());
        return executors;
    }

    /**
     * 建立一個可根據須要建立新線程的線程池，可是在之前構造的線程可用時將重用它們。對於執行不少短時間異步任務的程序而言，這些線程池一般可提升程序性能。調用
     * execute 將重用之前構造的線程（若是線程可用）。若是現有線程沒有可用的，則建立一個新線程並添加到池中。終止並從緩存中移除那些已有 60
     * 秒鐘未被使用的線程。所以，長時間保持空閒的線程池不會使用任何資源。注意，可使用 ThreadPoolExecutor
     * 構造方法建立具備相似屬性但細節不一樣（例如超時參數）的線程池。
     * 
     * @return
     */
    public ExecutorService createCachedThreadPool() {
        // 建立
        executors = Executors.newCachedThreadPool(getThreadFactory());
        return executors;
    }

    /**
     * 建立一個可重用固定線程數的線程池，以共享的無界隊列方式來運行這些線程。在任意點，在大多數 nThreads
     * 線程會處於處理任務的活動狀態。若是在全部線程處於活動狀態時提交附加任務
     * ，則在有可用線程以前，附加任務將在隊列中等待。若是在關閉前的執行期間因爲失敗而致使任何線程終止
     * ，那麼一個新線程將代替它執行後續的任務（若是須要）。在某個線程被顯式地關閉以前，池中的線程將一直存在。
     * 
     * @return
     */
    public ExecutorService createFixedThreadPool(int count) {
        // 建立
        executors = Executors.newFixedThreadPool(count, getThreadFactory());
        return executors;
    }


    /**
     * 獲取線程池工廠
     * 
     * @return
     */
    private ThreadFactory getThreadFactory() {
        return new ThreadFactory() {
            AtomicInteger sn = new AtomicInteger();
            public Thread newThread(Runnable r) {
                SecurityManager s = System.getSecurityManager();
                ThreadGroup group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
                Thread t = new Thread(group, r);
                t.setName("任務線程 - " + sn.incrementAndGet());
                return t;
            }
        };
    }
}

二、ExecutorProcessPoolide

package test;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/**
 * 線程處理類
 *
 * @author allan
 * @date   2017年3月20日
 */
public class ExecutorProcessPool {

    private ExecutorService executor;
    private static ExecutorProcessPool pool = new ExecutorProcessPool();
    private final int threadMax = 10;

    private ExecutorProcessPool() {
        System.out.println("threadMax>>>>>>>" + threadMax);
        executor = ExecutorServiceFactory.getInstance().createFixedThreadPool(threadMax);
    }

    public static ExecutorProcessPool getInstance() {
        return pool;
    }

    /**
     * 關閉線程池，這裏要說明的是：調用關閉線程池方法後，線程池會執行完隊列中的全部任務才退出
     * 
     * @author allan
     * @date   2017年3月20日
     */
    public void shutdown(){
        executor.shutdown();
    }

    /**
     * 提交任務到線程池，能夠接收線程返回值
     * 
     * @param task
     * @return
     * @author allan
     * @date   2017年3月20日
     */
    public Future<?> submit(Runnable task) {
        return executor.submit(task);
    }

    /**
     * 提交任務到線程池，能夠接收線程返回值
     * 
     * @param task
     * @return
     * @author allan
     * @date   2017年3月20日
     */
    public Future<?> submit(Callable<?> task) {
        return executor.submit(task);
    }

    /**
     * 直接提交任務到線程池，無返回值
     * 
     * @param task
     * @author allan
     * @date   2017年3月20日
     */
    public void execute(Runnable task){
        executor.execute(task);
    }

}

三、ExecutorTest高併發

package test;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * 測試類
 *
 * @author allan
 * @date   2017年3月20日
 */
public class ExecutorTest {

    public static void main(String[] args) {

        ExecutorProcessPool pool = ExecutorProcessPool.getInstance();

        for (int i = 0; i < 200; i++) {
            Future<?> future = pool.submit(new ExcuteTask1(i+""));
//          try {
//              若是接收線程返回值，future.get() 會阻塞，若是這樣寫就是一個線程一個線程執行。因此非特殊狀況不建議使用接收返回值的。
//              System.out.println(future.get());   
//          } catch (Exception e) {
//              e.printStackTrace();
//          }
        }

        for (int i = 0; i < 200; i++) {
            pool.execute(new ExcuteTask2(i+""));
        }

        //關閉線程池，若是是須要長期運行的線程池，不用調用該方法。
        //監聽程序退出的時候最好執行一下。
        pool.shutdown();
    }

    /**
     * 執行任務1，實現Callable方式
     *
     * @author allan
     * @date   2017年3月20日
     */
    static class ExcuteTask1 implements Callable<String> {
        private String taskName;

        public ExcuteTask1(String taskName) {
            this.taskName = taskName;
        }

        @Override
        public String call() throws Exception {
            try {
//              Java 6/7最佳的休眠方法爲TimeUnit.MILLISECONDS.sleep(100);
//              最好不要用 Thread.sleep(100);
                TimeUnit.MILLISECONDS.sleep((int)(Math.random() * 1000));// 1000毫秒之內的隨機數，模擬業務邏輯處理
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("-------------這裏執行業務邏輯，Callable TaskName = " + taskName + "-------------");
            return ">>>>>>>>>>>>>線程返回值，Callable TaskName = " + taskName + "<<<<<<<<<<<<<<";
        }
    }

    /**
     * 執行任務2，實現Runable方式
     *
     * @author allan
     * @date   2017年3月20日
     */
    static class ExcuteTask2 implements Runnable {
        private String taskName;

        public ExcuteTask2(String taskName) {
            this.taskName = taskName;
        }

        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep((int)(Math.random() * 1000));// 1000毫秒之內的隨機數，模擬業務邏輯處理
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("-------------這裏執行業務邏輯，Runnable TaskName = " + taskName + "-------------");
        }

    }
}