程序師世界是廣大編程愛好者互助、分享、學習的平台,程序師世界有你更精彩!
首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫知識
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
 程式師世界 >> 編程語言 >> JAVA編程 >> JAVA綜合教程 >> java並發:線程池、飽和策略、定制、擴展,java線程

java並發:線程池、飽和策略、定制、擴展,java線程

編輯:JAVA綜合教程

java並發:線程池、飽和策略、定制、擴展,java線程


一、序言

  當我們需要使用線程的時候,我們可以新建一個線程,然後顯式調用線程的start()方法,這樣實現起來非常簡便,但在某些場景下存在缺陷:如果需要同時執行多個任務(即並發的線程數量很多),頻繁地創建線程會降低系統的效率,因為創建和銷毀線程均需要一定的時間。

  線程池可以使線程得到復用,所謂線程復用就是線程在執行完一個任務後並不被銷毀,該線程可以繼續執行其他的任務。java.lang.concurrent包中的Executors類為我們創建線程池提供了方便。

 

二、Executors的簡單使用示例

此處我們先來看一個簡單的例子,如下:

package com.soft;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorsDemo {
    
    public static void main(String[] args) throws InterruptedException, ExecutionException {
//      ExecutorService executor = Executors.newSingleThreadExecutor();
//      ExecutorService executor = Executors.newCachedThreadPool();
        ExecutorService executor = Executors.newFixedThreadPool(5);
        Thread.sleep(5*1000);//方便監控工具能捕獲到
        for (int i = 0; i < 10; i++) {
            final int no = i;
            Runnable runnable = new Runnable() {
                public void run() {
                    try {
                        System.out.println("into" + no);
                        Thread.sleep(1000L);
                        System.out.println("end" + no);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            executor.execute(runnable);//ExecutorService有一個execute()方法,這個方法的參數是Runnable類型,通過execute(Runnable)方法即可將一個任務添加到線程池,任務的執行方法是Runnable類型對象的run()方法。
        }//End for
        
        executor.shutdown();
        System.out.println("Thread Main End!");
    }
}

其運行結果如下:

into0
into3
Thread Main End!
into4
into1
into2
end0
into5
end3
end1
end4
into8
into6
into7
end2
into9
end5
end7
end8
end6
end9

解說:這個例子應該很容易看懂,從運行結果來看,在任意某一時刻只有5個線程在執行,這是因為上述代碼通過Executors.newFixedThreadPool(5)語句創建了一個固定長度的線程池(長度為5),一個結束之後另再一個才開始執行。

 

三、Executors提供的線程池

  Executors是線程的工廠類,也可以說是一個線程池工具類,它調用其內部靜態方法(如newFixedThreadPool()等)即可創建一個線程池,通過參數設置,Executors提供不同的線程池機制。

 

四、簡述線程池的屬性

 

五、詳解ThreadPoolExecutor

  上文提到可以通過顯式的ThreadPoolExecutor構造函數來構造特定形式的線程池,ThreadPoolExecutor是java.util.concurrent包以內部線程池的形式對外提供線程池管理、線程調度等服務,此處我們來了解一下ThreadPoolExecutor

(1)一般使用方式:

ExecutorService exec = new ThreadPoolExecutor(8,
                8, 
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(100),
                new ThreadPoolExecutor.CallerRunsPolicy());

下文詳解此示例涉及的一些內容

(2)構造函數的聲明:

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;  
    }

(3)函數參數說明:

參數名 代表含義 corePoolSize 線程池的基本大小(核心線程池大小) maximumPoolSize 線程池的最大大小 keepAliveTime 線程池中超過corePoolSize數目的空閒線程的最大存活時間 unit keepAliveTime參數的時間單位 workQueue 任務阻塞隊列 threadFactory 新建線程的工廠 handler 當提交的任務數超過maxmumPoolSize與workQueue之和時,任務會交給RejectedExecutionHandler來處理

 

 

 

 

 

 

 

進一步解說:

A、當提交新任務時,若線程池大小小於corePoolSize,將創建一個新的線程來執行任務,即使此時線程池中存在空閒線程;

B、當提交新任務時,若線程池達到corePoolSize大小,新提交的任務將被放入workQueue中,等待線程池調度執行;

C、當提交新任務時,若workQueue已滿,且maximumPoolSize>corePoolSize,將創建新的線程來執行任務;

D、當提交新任務時,若任務總數超過maximumPoolSize,新提交的任務將由RejectedExecutionHandler來處理;

E、當線程池中的線程數超過corePoolSize時,若線程的空閒時間達到keepAliveTime,則關閉空閒線程

(4)任務阻塞隊列選擇機制

 (5)簡述SynchronousQueue

注:此處貼出SynchronousQueue的使用示例,示例中使用了Semaphore,更多關於SynchronousQueue及Semaphore的內容請參考其他文章

package com.test;

import java.util.concurrent.Semaphore;
import java.util.concurrent.SynchronousQueue;

/*
 * 程序中有10個線程來消費生成者產生的數據,這些消費者都調用TestDo.doSome()方法去進行處理,
 * 每個消費者都需要一秒才能處理完,程序應保證這些消費者線程依次有序地消費數據,只有上一個消費者消費完後,
 * 下一個消費者才能消費數據,下一個消費者是誰都可以,但要保證這些消費者線程拿到的數據是有順序的。
 */
public class SynchronousQueueTest {
    
    public static void main(String[] args) {
        
        System.out.println("begin:" + (System.currentTimeMillis() / 1000));
        // 定義一個Synchronous
        final SynchronousQueue<String> sq = new SynchronousQueue<String>();
// 定義一個數量為1的信號量,其作用相當於一個互斥鎖
        final Semaphore sem = new Semaphore(1);
for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                public void run() {
                    try {
                        sem.acquire();
                        String input = sq.take();
                        String output = TestDo.doSome(input);//內部類
                        System.out.println(Thread.currentThread().getName()+ ":" + output);
                        sem.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
        
        for (int i = 0; i < 10; i++) {
            String input = i + ""; //此處將i變成字符串
            try {
                sq.put(input);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
    }//End main
    
}

class TestDo {
    public static String doSome(String input) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String output = input + ":" + (System.currentTimeMillis() / 1000);
        return output;
    }
}

上述代碼的運行結果如下:

begin:1458954798
Thread-0:0:1458954799
Thread-1:1:1458954800
Thread-2:2:1458954801
Thread-3:3:1458954802
Thread-4:4:1458954803
Thread-5:5:1458954804
Thread-6:6:1458954805
Thread-7:7:1458954806
Thread-8:8:1458954807
Thread-9:9:1458954808

從上述結果看,上例在任意某一時刻只有一個線程在執行,且只有前一個線程執行完下一個線程才開始

 

六、飽和策略(線程池任務拒絕策略)

上文提到ThreadPoolExecutor構造函數的RejectedExecutionHandler handler參數,該參數表示當提交的任務數超過maxmumPoolSize與workQueue之和時,任務會交給RejectedExecutionHandler來處理,此處我們來具體了解一下

(1)四種飽和策略

(2)源碼分析:

RejectedExecutionHandler這個接口是用來處理被丟棄的線程的異常處理接口,其源碼如下:

public interface RejectedExecutionHandler{
    //被線程池丟棄的線程處理機制
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) ;
}

AbortPolicy(中止策略)繼承RejectedExecutionHandler接口,其源碼如下:

public static class AbortPolicy implements RejectedExecutionHandler{

    public AbortPolicy(){}
    
    //直接拋出異常
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        throw new RejectedExecutionException("Task"+r.toString()+"rejected from"+executor.toString());
    }
    
}

我們可以自己實現RejectedExecutionHandler接口,將實現類作為線程丟棄處理類,代碼如下:

package com.test;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class RejectedExecutionHandlerDemo implements RejectedExecutionHandler{

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // TODO Auto-generated method stub
        System.out.println("線程信息"+r.toString()+"被遺棄的線程池:"+executor.toString());
    }
    
}

 

 

七、定制ThreadPoolExecutor

(1)通過修改參數的方式達到定制目的

(2)通過自定義方式(封裝各種參數)達到定制目的

示例(摘自網絡):

import java.util.concurrent.ArrayBlockingQueue;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.RejectedExecutionHandler;  
import java.util.concurrent.ThreadFactory;  
import java.util.concurrent.ThreadPoolExecutor;  
import java.util.concurrent.TimeUnit;  
import java.util.concurrent.atomic.AtomicInteger;  
  
  
public class CustomThreadPoolExecutor {  
  
      
    private ThreadPoolExecutor pool = null;  
      
      
    /** 
     * 線程池初始化方法 
     *  
     * corePoolSize 核心線程池大小----10 
     * maximumPoolSize 最大線程池大小----30 
     * keepAliveTime 線程池中超過corePoolSize數目的空閒線程最大存活時間----30+單位TimeUnit 
     * TimeUnit keepAliveTime時間單位----TimeUnit.MINUTES 
     * workQueue 阻塞隊列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞隊列 
     * threadFactory 新建線程工廠----new CustomThreadFactory()====定制的線程工廠 
     * rejectedExecutionHandler 當提交任務數超過maxmumPoolSize+workQueue之和時, 
     *                          即當提交第41個任務時(前面線程都沒有執行完,此測試方法中用sleep(100)), 
     *                          任務會交給RejectedExecutionHandler來處理 
     */  
    public void init() {  
        pool = new ThreadPoolExecutor(  
                10,  
                30,  
                30,  
                TimeUnit.MINUTES,  
                new ArrayBlockingQueue<Runnable>(10),  
                new CustomThreadFactory(),  
                new CustomRejectedExecutionHandler());  
    }  
  
      
    public void destory() {  
        if(pool != null) {  
            pool.shutdownNow();  
        }  
    }  
      
      
    public ExecutorService getCustomThreadPoolExecutor() {  
        return this.pool;  
    }  
      
    private class CustomThreadFactory implements ThreadFactory {  
  
        private AtomicInteger count = new AtomicInteger(0);  
          
        @Override  
        public Thread newThread(Runnable r) {  
            Thread t = new Thread(r);  
            String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);  
            System.out.println(threadName);  
            t.setName(threadName);  
            return t;  
        }  
    }  
      
      
    private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {  
  
        @Override  
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {  
            // 記錄異常  
            // 報警處理等  
            System.out.println("error.............");  
        }  
    }  
      
      
      
    // 測試構造的線程池  
    public static void main(String[] args) {  
        CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor();  
        // 1.初始化  
        exec.init();  
          
        ExecutorService pool = exec.getCustomThreadPoolExecutor();  
        for(int i=1; i<100; i++) {  
            System.out.println("提交第" + i + "個任務!");  
            pool.execute(new Runnable() {  
                @Override  
                public void run() {  
                    try {  
                        Thread.sleep(300);  
                    } catch (InterruptedException e) {  
                        e.printStackTrace();  
                    }  
                    System.out.println("running=====");  
                }  
            });  
        }  
          
          
          
        // 2.銷毀----此處不能銷毀,因為任務沒有提交執行完,如果銷毀線程池,任務也就無法執行了  
        // exec.destory();  
          
        try {  
            Thread.sleep(10000);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}

 

八、擴展ThreadPoolExecutor

 

九、源碼視角

從源碼視角分析Executors、ThreadPoolExecutor、ExecuteService、Executor之間的關系,此處簡單提及,讀者可查看下一節“參考資料”以了解相關內容

(1)Executors

  從Java5開始新增了Executors類,它有幾個靜態工廠方法用來創建線程池,這些靜態工廠方法返回一個ExecutorService類型的值,此值即為線程池的引用。

(2)Executor

Executor是一個接口,裡面只有一個方法

public interface Executor {
    void execute(Runnable command);
}

(3)ExecuteService

ExecuteService也是一個接口,其定義如下:

public interface ExecutorService extends Executor {...}

(4)ThreadPoolExecutor繼承AbstractExecutorService,AbstractExecutorService實現ExecutorService接口

public class ThreadPoolExecutor extends AbstractExecutorService {...}
public abstract class AbstractExecutorService implements ExecutorService {...}

 

十、ExecutorService的生命周期

在本文最開始的那個示例中,有一句代碼,如下:

executor.shutdown();

該語句並不是終止線程的運行,而是禁止在這個executor中添加新的任務,下文描述了該語句對於ExecutorService的意義。

 

十一、參考資料

本文僅簡單闡述了Java並發中關於Executors及ThreadPoolExecutor的內容,此處貼出一些優質文章以供讀者閱覽

(1)http://blog.csdn.net/xiamizy/article/details/40781939

(2)http://www.cnblogs.com/dolphin0520/p/3932921.html

(3)http://www.cnblogs.com/yezhenhan/archive/2012/01/07/2315645.html

  1. 上一頁:
  2. 下一頁:
Copyright © 程式師世界 All Rights Reserved