效果：這種方案導(dǎo)致的現(xiàn)象是接收到的消息會迅速堆積，我們從消息隊(duì)列（或者其他地方）取出了大量消息，但是處理線程的速度又跟不上，所以導(dǎo)致的問題是大量的Task會堆積在線程池底層維護(hù)的一個(gè)阻塞隊(duì)列中，這會極大的耗費(fèi)存儲空間，影響系統(tǒng)的性能。

分析：當(dāng)execute()一個(gè)任務(wù)的時(shí)候，如果有空閑的worker線程，那么投入運(yùn)行，否則看設(shè)置的***線程個(gè)數(shù)，沒有達(dá)到線程個(gè)數(shù)限制就創(chuàng)建新線程，接新任務(wù)，否則就把任務(wù)緩沖到一個(gè)阻塞隊(duì)列中，問題就是這個(gè)隊(duì)列，默認(rèn)的大小是沒有限制的，所以就會大量的堆積任務(wù)，必然耗費(fèi)heap空間。

public static ExecutorService newFixedThreadPool(int nThreads) { 
        return new ThreadPoolExecutor(nThreads, nThreads, 
                                      0L, TimeUnit.MILLISECONDS, 
                                      new LinkedBlockingQueue<Runnable>()); 
    } 
 
public LinkedBlockingQueue() { 
        this(Integer.MAX_VALUE); // capacity 
    } 

計(jì)數(shù)限制

面對上述問題，想到了要限制消息接收的速度，自然就想到了各種線程同步的原語，不過在這里最簡單的就是使用一個(gè)Volatile的計(jì)數(shù)器。

消息接收者：

public class Receiver { 
    private static volatile boolean inited = false; 
    private static volatile boolean shutdown = false; 
    private static volatile int cnt = 0; 
    private MessageHandler messageHandler; 
    public void start(){ 
        Executors.newSingleThreadExecutor().execute(new Runnable() { 
            @Override 
            public void run() { 
                while(!shutdown){ 
                    init(); 
                    recv(); 
                } 
            } 
        }); 
    } 
 
    /** 
     * 模擬消息接收 
     */ 
    public void recv(){ 
            Message msg = new Message("Msg" + System.currentTimeMillis()); System.out.println(String.format("接收到消息(%d): %s", ++cnt, msg)); messageHandler.handle(msg); } public void init(){ if(!inited){ messageHandler = new MessageHandler(); inited = true; } } public static void main(String[] args) { new Receiver().start(); 
    } 
} 

消息處理：

public class MessageHandler { 
    private static final int THREAD_POOL_SIZE = 1; 
    private ExecutorService service = Executors.newFixedThreadPool(THREAD_POOL_SIZE); 
 
    public void handle(Message msg){ 
        try { 
            service.execute(new Runnable() { 
 
                @Override 
                public void run() { 
                    parseMsg(msg); 
                } 
            }); 
        } catch (Throwable e) { 
            System.out.println("消息處理異常" + e); } } /** * 比較耗時(shí)的消息處理流程 */ public void parseMsg(Message message){ try { Thread.sleep(10000); System.out.println("解析消息：" + message); } catch (InterruptedException e) { e.printStackTrace(); }finally { 
            Receiver.limit --; 
        } 
 
    } 
} 

public class MessageHandler { 
    private static final int THREAD_POOL_SIZE = 4; 
 
    ThreadPoolExecutor service = new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE, 0L, TimeUnit.MILLISECONDS, 
            new SynchronousQueue<Runnable>(), new RejectedExecutionHandler() { 
        @Override 
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { 
            System.out.println("自定義拒絕策略"); try { executor.getQueue().put(r); System.out.println("重新放任務(wù)回隊(duì)列"); } catch (InterruptedException e) { e.printStackTrace(); } } }); public void handle(Message msg) { try { System.out.println(service.getTaskCount()); System.out.println(service.getQueue().size()); System.out.println(service.getCompletedTaskCount()); service.execute(new Runnable() { @Override public void run() { parseMsg(msg); } }); } catch (Throwable e) { System.out.println("消息處理異常" + e); } } /** * 比較耗時(shí)的消息處理流程 */ public void parseMsg(Message message) { while (true) { try { System.out.println("線程名:" + Thread.currentThread().getName()); System.out.println("解析消息：" + message); Thread.sleep(1000); } catch (InterruptedException e) { 
                e.printStackTrace(); 
            } 
        } 
    } 
} 

public class MessageHandler { 
    private static final int THREAD_POOL_SIZE = 4; 
    private static final int BLOCK_QUEUE_CAP = 500; 
    ThreadPoolExecutor service = new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE, 0L, TimeUnit.MILLISECONDS, 
            new LinkedBlockingQueue<Runnable>(BLOCK_QUEUE_CAP), new SimpleThreadFactory(), new RejectedExecutionHandler() { 
        @Override 
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { 
            System.out.println("自定義拒絕策略"); try { executor.getQueue().put(r); System.out.println("重新放任務(wù)回隊(duì)列"); } catch (InterruptedException e) { e.printStackTrace(); } } }); public void handle(Message msg) { try { service.execute(new Runnable() { @Override public void run() { parseMsg(msg); } }); } catch (Throwable e) { System.out.println("消息處理異常" + e); } } /** * 比較耗時(shí)的消息處理流程 */ public void parseMsg(Message message) { try { Thread.sleep(5000); System.out.println("線程名:" + Thread.currentThread().getName()); System.out.println("解析消息：" + message); } catch (InterruptedException e) { e.printStackTrace(); } } static class SimpleThreadFactory implements ThreadFactory { @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName("Thread-" + System.currentTimeMillis()); return thread; 
        } 
    } 
}