使用并发在提高程序运行速度的同时，也会带来更多的问题和风险。
生产者和消费者模式
在并发种使用生产者和消费者模式能够解决绝大多数并发问题。
由于生产者和消费者的对数据的处理速度不同，会导致相互制约，所以需要解耦。生产者和消费者解耦依靠一个阻塞队列即可，生产者和消费者不直接联系，一个把数据扔给队列，一个从队列取数据，完美的平衡了生产者和消费者的处理能力。
一个生产者线程进行抽取邮件的任务，然后把邮件放到阻塞队列中，一个消费者线程池进行把抽取的邮件入到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));
}
// 每五分钟执行一次
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{
}
多生产者和多消费者场景
生产者1强消息存放在阻塞队列1里，消费者1从队列里读取消息，然后通过消息ID来散列得到N个队列中的一个，根据编号将消息放到不同的队列里，每个阻塞队列会分配一个线程来消费阻塞队列里的数据。
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;
//启动一个消息分发线程。在这个线程里子队列自动去总队列里获取消息
public class Main {
static class DispatchMessageTask implements Runnable{
//分发消息，负责把消息从大队列塞到小队列里
@Override
public void run() {
BlockingQueue<Message> subQueue;
while(true){
try {
//如果没有数据，那么阻塞在这里
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){
//出现错误，在获取队列大小之后，队列进行了一次删除操作
if(++errorCount < 3){
continue;
}
}
}
}
}
线上问题定位
-
Linux使用top命令查看每个进程的情况：
我们的程序是java应用，所以只需要关注Command是java的性能数据，Command表示启动当前进程的命令，在java进程这一行里可以看到cpu利用率，这是当前机器所有核加在一起的CPU利用率（所以超出100%正常）。
top
-
再使用top的交互命令数字1查看每个CPU的性能数据。
命令行只显示了cpu0，说明这是个1核的虚拟机，cpu利用率在3.0%，如果是100%，那么可能是产生死循环了。
命令1
cpu参数含义
| 参数 | 描述 |
|---|---|
us |
用户空间占用cpu百分比 |
1.0% sy |
内核空间占用cpu百分比 |
0.0% ni |
用户进程空间改变过优先级的进程占用cpu百分比 |
98.7% id |
空闲cpu百分比 |
0.0% wa |
等待输入/输出的cpu时间百分比 |
- 使用top的交互命令H查看每个线程的性能信息
- 某个线程cpu利用率一直100%，这个线程可能有死循环，记住这个pid。
- 某个线程一直在top 10位置，这个线程可能有性能问题。
- CPU利用率高的几个线程在不停变化，说明并不是由某一个线程导致cpu偏高。
H
P60820-124012(1).jpg
P60820-124118(1).jpg
P60820-124201(1).jpg
P60820-124248(1).jpg
网友评论