使用并發(fā)在提高程序運(yùn)行速度的同時(shí)，也會(huì)帶來(lái)更多的問(wèn)題和風(fēng)險(xiǎn)。
生產(chǎn)者和消費(fèi)者模式
在并發(fā)種使用生產(chǎn)者和消費(fèi)者模式能夠解決絕大多數(shù)并發(fā)問(wèn)題擦耀。
由于生產(chǎn)者和消費(fèi)者的對(duì)數(shù)據(jù)的處理速度不同运准，會(huì)導(dǎo)致相互制約，所以需要解耦碰凶。生產(chǎn)者和消費(fèi)者解耦依靠一個(gè)阻塞隊(duì)列即可暮芭，生產(chǎn)者和消費(fèi)者不直接聯(lián)系，一個(gè)把數(shù)據(jù)扔給隊(duì)列欲低，一個(gè)從隊(duì)列取數(shù)據(jù)辕宏，完美的平衡了生產(chǎn)者和消費(fèi)者的處理能力。
一個(gè)生產(chǎn)者線程進(jìn)行抽取郵件的任務(wù)砾莱，然后把郵件放到阻塞隊(duì)列中瑞筐，一個(gè)消費(fèi)者線程池進(jì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("開(kāi)始");
long start = System.currentTimeMillis();
//抽取所有的郵件放到隊(duì)列中
new ExtractEmailTask().start();
//把隊(duì)列的文章插入Wiki
insertToWiki();
long end = System.currentTimeMillis();
double cost = (end - start) / 1000;
System.out.println("完成腊瑟，花費(fèi)時(shí)間=" + cost + "秒聚假。");
}
// 把隊(duì)列中的文章插入到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() + "把一個(gè)郵件插入到Wiki=" + System.currentTimeMillis());
}
}
public static void main(String[] args) {
QuickEmailToWikiExtractor mail = new QuickEmailToWikiExtractor();
mail.exetract();
}
}
class ExchangeEmailShallowDTO{
}
多生產(chǎn)者和多消費(fèi)者場(chǎng)景
生產(chǎn)者1強(qiáng)消息存放在阻塞隊(duì)列1里，消費(fèi)者1從隊(duì)列里讀取消息财松，然后通過(guò)消息ID來(lái)散列得到N個(gè)隊(duì)列中的一個(gè)瘪贱，根據(jù)編號(hào)將消息放到不同的隊(duì)列里，每個(gè)阻塞隊(duì)列會(huì)分配一個(gè)線程來(lái)消費(fèi)阻塞隊(duì)列里的數(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;
//總消息隊(duì)列管理
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;
//啟動(dòng)一個(gè)消息分發(fā)線程政敢。在這個(gè)線程里子隊(duì)列自動(dòng)去總隊(duì)列里獲取消息
public class Main {
static class DispatchMessageTask implements Runnable{
//分發(fā)消息，負(fù)責(zé)把消息從大隊(duì)列塞到小隊(duì)列里
@Override
public void run() {
BlockingQueue<Message> subQueue;
while(true){
try {
//如果沒(méi)有數(shù)據(jù)胚迫，那么阻塞在這里
Message msg = MsgQueueManager.messageQueue.take();
while( (subQueue = getSubQueue()) == null){//沒(méi)有獲取到子隊(duì)列喷户，等待
try{
Thread.sleep(1000);
}catch(InterruptedException e){
Thread.currentThread().interrupt();
}
}
//把消息放到小隊(duì)列里
try{
subQueue.put(msg);
}catch(InterruptedException e){
Thread.currentThread().interrupt();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
//使用散列（hash）算法獲取一個(gè)子隊(duì)列
public static BlockingQueue<Message> getSubQueue(){
List<BlockingQueue<Message>> subMsgQueues = new ArrayList();//假裝這是一個(gè)全局的子隊(duì)列
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)錯(cuò)誤，在獲取隊(duì)列大小之后访锻，隊(duì)列進(jìn)行了一次刪除操作
if(++errorCount < 3){
continue;
}
}
}
}
}
線上問(wèn)題定位
-
Linux使用top命令查看每個(gè)進(jìn)程的情況：
我們的程序是java應(yīng)用褪尝，所以只需要關(guān)注Command是java的性能數(shù)據(jù)，Command表示啟動(dòng)當(dāng)前進(jìn)程的命令期犬，在java進(jìn)程這一行里可以看到cpu利用率河哑，這是當(dāng)前機(jī)器所有核加在一起的CPU利用率（所以超出100%正常）。
top -
再使用top的交互命令數(shù)字1查看每個(gè)CPU的性能數(shù)據(jù)龟虎。
命令行只顯示了cpu0璃谨，說(shuō)明這是個(gè)1核的虛擬機(jī)，cpu利用率在3.0%，如果是100%佳吞，那么可能是產(chǎn)生死循環(huán)了拱雏。
命令1
cpu參數(shù)含義
| 參數(shù) | 描述 |
|---|---|
us |
用戶空間占用cpu百分比 |
1.0% sy |
內(nèi)核空間占用cpu百分比 |
0.0% ni |
用戶進(jìn)程空間改變過(guò)優(yōu)先級(jí)的進(jìn)程占用cpu百分比 |
98.7% id |
空閑cpu百分比 |
0.0% wa |
等待輸入/輸出的cpu時(shí)間百分比 |
- 使用top的交互命令H查看每個(gè)線程的性能信息
- 某個(gè)線程cpu利用率一直100%，這個(gè)線程可能有死循環(huán)底扳，記住這個(gè)pid铸抑。
- 某個(gè)線程一直在top 10位置，這個(gè)線程可能有性能問(wèn)題衷模。
- CPU利用率高的幾個(gè)線程在不停變化鹊汛，說(shuō)明并不是由某一個(gè)線程導(dǎo)致cpu偏高。
H
P60820-124012(1).jpg
P60820-124118(1).jpg
P60820-124201(1).jpg
P60820-124248(1).jpg
