使用并發(fā)在提高程序運行速度的同時，也會帶來更多的問題和風險。

生產(chǎn)者和消費者模式

在并發(fā)種使用生產(chǎn)者和消費者模式能夠解決絕大多數(shù)并發(fā)問題。

由于生產(chǎn)者和消費者的對數(shù)據(jù)的處理速度不同，會導致相互制約，所以需要解耦。生產(chǎn)者和消費者解耦依靠一個阻塞隊列即可，生產(chǎn)者和消費者不直接聯(lián)系，一個把數(shù)據(jù)扔給隊列，一個從隊列取數(shù)據(jù)，完美的平衡了生產(chǎn)者和消費者的處理能力。

一個生產(chǎn)者線程進行抽取郵件的任務，然后把郵件放到阻塞隊列中，一個消費者線程池進行把抽取的郵件入到Wiki中。

package com.yuna;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class QuickEmailToWikiExtractor{
    private ThreadPoolExecutor threadPool;
    private BlockingQueue<ExchangeEmailShallowDTO> emailQueue;
    
    public QuickEmailToWikiExtractor() {
        emailQueue = new LinkedBlockingQueue<ExchangeEmailShallowDTO>();
        int corePoolSize = Runtime.getRuntime().availableProcessors() * 2;
        threadPool = new ThreadPoolExecutor(corePoolSize, corePoolSize,101, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(2000));
    }
    // 每五分鐘執(zhí)行一次
    public void exetract(){
        System.out.println("開始");
        long start = System.currentTimeMillis();
        
        //抽取所有的郵件放到隊列中
        new ExtractEmailTask().start();
        //把隊列的文章插入Wiki
        insertToWiki();
        
        long end = System.currentTimeMillis();
        double cost = (end - start) / 1000;
        System.out.println("完成，花費時間=" + cost + "秒。");
    }
    // 把隊列中的文章插入到Wiki
    private void insertToWiki(){
        while(true){
            //2s讀取不到就退出
            ExchangeEmailShallowDTO email;
            try {
                email = emailQueue.poll(2,  TimeUnit.SECONDS);
                if(email == null){
                    break;
                }
                threadPool.execute(new InsertToWikiTask(email));
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    protected void extractEmail(){
        List<ExchangeEmailShallowDTO> allEmails = new ArrayList();//這里是抽取出的所有郵件
        int i = 0;
        while(i++<10){
            allEmails.add(new ExchangeEmailShallowDTO());
        }
        if(allEmails == null){
            return;
        }
        for(ExchangeEmailShallowDTO email: allEmails){
            emailQueue.offer(email);
        }
    }
    //抓取郵件線程
    class ExtractEmailTask extends Thread{
        public void run(){
            System.out.println("抽取郵件。。。");
            extractEmail();
        }
    }
    class InsertToWikiTask implements Runnable{
        private ExchangeEmailShallowDTO email;
        public InsertToWikiTask(ExchangeEmailShallowDTO email){
            this.email = email;
        }
        public void run() {
            System.out.println(Thread.currentThread().getName() + "把一個郵件插入到Wiki=" + System.currentTimeMillis());
        }
    }
    
    public static void main(String[] args) {
        QuickEmailToWikiExtractor mail = new QuickEmailToWikiExtractor();
        mail.exetract();
    }
}

class ExchangeEmailShallowDTO{
    
}

多生產(chǎn)者和多消費者場景

生產(chǎn)者1強消息存放在阻塞隊列1里，消費者1從隊列里讀取消息，然后通過消息ID來散列得到N個隊列中的一個，根據(jù)編號將消息放到不同的隊列里，每個阻塞隊列會分配一個線程來消費阻塞隊列里的數(shù)據(jù)。

package com.yuna.queue;

public class Message {

}

package com.yuna.queue;

public interface IMsgQueue {
    void put(Message msg);
    Message take();
}

package com.yuna.queue;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedTransferQueue;

//總消息隊列管理
public class MsgQueueManager implements IMsgQueue {
    public final static BlockingQueue<Message> messageQueue = new LinkedTransferQueue<Message>();
    
    @Override
    public void put(Message msg) {
        try {
            messageQueue.put(msg);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @Override
    public Message take() {
        try {
            return messageQueue.take();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return null;
    }

}

package com.yuna.queue;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

//啟動一個消息分發(fā)線程。在這個線程里子隊列自動去總隊列里獲取消息
public class Main {
    static class DispatchMessageTask implements Runnable{
        //分發(fā)消息，負責把消息從大隊列塞到小隊列里
        @Override
        public void run() {
            BlockingQueue<Message> subQueue;
            while(true){
                try {
                    //如果沒有數(shù)據(jù)，那么阻塞在這里
                    Message msg = MsgQueueManager.messageQueue.take();
                    while( (subQueue = getSubQueue()) == null){//沒有獲取到子隊列，等待
                        try{
                            Thread.sleep(1000);
                        }catch(InterruptedException e){
                            Thread.currentThread().interrupt();
                        }
                    }
                    //把消息放到小隊列里
                    try{
                        subQueue.put(msg);
                    }catch(InterruptedException e){
                        Thread.currentThread().interrupt();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                
            }
        }
        
    }
    //使用散列（hash）算法獲取一個子隊列
    public static BlockingQueue<Message> getSubQueue(){
        List<BlockingQueue<Message>> subMsgQueues = new ArrayList();//假裝這是一個全局的子隊列
        int errorCount = 0;
        for(;;){
            if(subMsgQueues.isEmpty()){
                return null;
            }
            int index = (int)(System.nanoTime() % subMsgQueues.size());
            try{
                return subMsgQueues.get(index);
            }catch(Exception e){
                //出現(xiàn)錯誤，在獲取隊列大小之后，隊列進行了一次刪除操作
                if(++errorCount < 3){
                    continue;
                }
            }
        }
    }
}

線上問題定位

• Linux使用top命令查看每個進程的情況：
  我們的程序是java應用，所以只需要關(guān)注Command是java的性能數(shù)據(jù)，Command表示啟動當前進程的命令，在java進程這一行里可以看到cpu利用率，這是當前機器所有核加在一起的CPU利用率（所以超出100%正常）。

  
  
  top

• 再使用top的交互命令數(shù)字1查看每個CPU的性能數(shù)據(jù)。
  命令行只顯示了cpu0，說明這是個1核的虛擬機，cpu利用率在3.0%，如果是100%，那么可能是產(chǎn)生死循環(huán)了。

  
  
  命令1

cpu參數(shù)含義

• 使用top的交互命令H查看每個線程的性能信息
  • 某個線程cpu利用率一直100%，這個線程可能有死循環(huán)，記住這個pid。
  • 某個線程一直在top 10位置，這個線程可能有性能問題。
  • CPU利用率高的幾個線程在不停變化，說明并不是由某一個線程導致cpu偏高。

H

P60820-124012(1).jpg

P60820-124118(1).jpg

P60820-124201(1).jpg

P60820-124248(1).jpg