双缓冲队列尝试

提出背景：
在C/S模式的系统里面，服务器端的主线程，除了要接收消息之外，还要处理消息。这使得主线程的工作量不但很大，而且工作很繁杂。这种情况在软件设计的角度来看，是很不好的：第一，这样让主线程类看起来异常的臃肿和难易阅读，第二，软件设计追求的目标是“尽量让每个类处理的工作都很单一，这样便于以后的调试和进一步对程序的扩展和移植”，这样的设计背离了软件设计中“模块化设计”的原则。
为此，很多程序员会设计一个全局的LinkedBlockingQueue对象来存放服务器主线程所收到的消息，即主线程收到消息之后，不做任何业务逻辑处理，直接put到阻塞队列中去，然后再启动一个处理消息的线程来从阻塞队列中take消息，这样就让服务器端程序大大解耦了，也便于阅读了，当然，最重要的是，效率也提高了不少。
但是在负荷量很大的系统里面，这样做仍然不是最好的办法。上面的解决办法中，仍然存在两个线程竞争一个资源的情况，即仍然存在同步的问题，只是LinkedBlockingQueue在底层进行了处理，在负荷量很大，并且要频繁访问同一个LinkedBlockingQueue对象的情况下，会使得效率较原始情况没有多大的提高。分析一下：主线程每次拿到消息M，都要put到LinkedBlockingQueue对象中，这样在LinkedBlockingQueue底层就要处理一次线程同步问题，当工作量很大很频繁的时候，这样的操作未免就显得有些累赘了，浪费了很多的时间和资源。同样在处理线程，每次take一个消息M，就会处理一次线程同步，同样降低了效率。注意，只是在负荷量很大的情况下(如中国移动CMPP短信网关)，这样的弱点才会很明显地暴露出来。

问题分析：
暴露出了问题，就需要我们想办法去解决，通过上面的分析，发现瓶颈在于，我们是对同一个LinkedBlockingQueue进行的操作，才会引起这么多的滞后。那么能不能试着去用两个LinkedBlockingQueue对象进行操作呢？一个专门给主线程put，另外一个专门给处理线程take和处理，当他们处理完了，再来交换？这样的话，就保障了基本上在每个时刻，每个线程都是在处理自己独自占有的资源，没有引发同步问题。这称之为：双缓冲队列。
当然，上面的解决方案是在一种很理想的情况下，即两个队列，在同一时刻被放满和取空，这样他们在程序运行过程中所达到的效率是最高的，因为没有任何消息被延后处理。所有消息都按照原定计划被处理了。现实情况往往没有这般顺利，但是上面的方案也不失为我们提供了一种解决思路，这其中还涉及到很多的处理机制。为了分析方便，我们暂且将两个LinkedBlockingQueue对象分别称之为LP(主线程控制的专门put的队列)，LT(处理线程控制的，专门take的队列)。
第一，    LP和LT队列什么时候才进行交换？肯定是在LT将所有的消息处理完了之后才进行交换，如果LT里面的消息都没有处理完就被交换了，那有可能LT里面的消息有些几年甚至永远不会被处理，这样的系统是无法投入使用的。
第二，    如果在LT处理完了，LP中还没有收到任何消息会怎么办？这是一种很极端的情况，但是也必须考虑到，延伸一下，如果LT里面的消息处理完了，而LP里面却只有一两条消息，该怎么办？要继续交换，还是等待一段时间以后再进行交换？在这里，笔者采用的是前者。目的是为了让消息都能及时地被处理。

问题解决：

首先我们来看看不使用双缓冲机制的程序：

package cn.netjava.lbq.copy;
//玩具类
public class Toy {
private String name;

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}

}

 

package cn.netjava.lbq.copy;

//在线程中生产玩具，放入单缓冲队列
public class Factory extends Thread{


public void run(){
while(true){
Toy t = new Toy ();
t.setName("玩具");
try {
Tools.lT.put(t);
System.out.println("生产出一个玩具");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

 

package cn.netjava.lbq.copy;
//从单缓冲队列中取出玩具对象

public class Kid extends Thread {
	long time1 = System.currentTimeMillis();
	int count = 0;
	public void run() {
		while(true){
		try {
		    Tools.lT.take();
			count++;
//计算取100000个所需要的时间
			if(count==100000){
				System.out.println(System.currentTimeMillis()-time1);
				Thread.sleep(100000);
			}
			System.out.println("玩完一个玩具");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		}
		}

}

 

package cn.netjava.lbq.copy;
//存放玩具对象的的队列
import java.util.concurrent.LinkedBlockingQueue;

public class Tools {
	public static LinkedBlockingQueue<Tool>lT = new LinkedBlockingQueue<Tool>(10000);
	
}

 

package cn.netjava.lbq.copy;

public class MainTest {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		Factory f = new Factory();
		f.start();
		Kid k = new Kid();
		k.start();
		
	}

}

 取出100000个玩具的时间为：3609ms

 

 

再来看看使用双缓冲队列的情况，toy类就不再贴出了。

看其他类，中间做了一些改动

package cn.netjava.lbq;

import java.util.concurrent.LinkedBlockingQueue;

public class Tools {
	public static LinkedBlockingQueue<Tool>lT = new LinkedBlockingQueue<Tool>(10000);
	public static LinkedBlockingQueue<Tool>lP = new LinkedBlockingQueue<Tool>(10000);
}

 

package cn.netjava.lbq;


public class Kid extends Thread {
	long time1 = System.currentTimeMillis();
	int count = 0;
	public void run() {
		while(true){
		try {
			 Tools.lT.take();
			count++;
			
			if(count==10000){
			System.out.println(System.currentTimeMillis()-time1);
			Thread.sleep(100000);
			}
			System.out.println("玩完一个玩具");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		}
		}

}

 

package cn.netjava.lbq;

import java.util.concurrent.LinkedBlockingQueue;

public class Factory extends Thread{
	

	public void run(){
		while(true){
		Toy t = new Toy ();
		t.setName("玩具");
		try {
			Tools.lP.put(t);
			System.out.println("生产出一个玩具");
		} catch (InterruptedException e) {
			e.printStackTrace();
	 	}
		 }
	}
}

 

	package cn.netjava.lbq;

import java.util.concurrent.LinkedBlockingQueue;
	/**
	 * 定义一个双缓冲队列
	 * @author Irving
	 * @time  2009-9-16----上午10:25:49
	 *
	 */
	public class DoubleBufferList {
	
		private LinkedBlockingQueue<Object> lP;
		private LinkedBlockingQueue<Object> lT;
		private int gap;
		/**
		 * 构造方法
		 * @param lP 用来存放对象的阻塞队列
		 * @param lT 用来取对象的阻塞队列
		 * @param gap 交换的间隔
		 */
		
		public DoubleBufferList(LinkedBlockingQueue  lP
				,LinkedBlockingQueue  lT,int gap){
			this.lP=lP;
			this.lT=lT;
			this.gap=gap;
		}
		
		public void check(){
			Runnable runner = new Runnable(){
				public void run() {
					while(true){
				if(lT.size()==0){
					synchronized(lT){
					synchronized(lP){
					lT.addAll(lP);
					}
					lP.clear();
					}
				}				try {
					Thread.sleep(gap);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
					}	
					
				}				
			};
			Thread thread = new Thread(runner);
			thread.start();
		}
		
	}

 

package cn.netjava.lbq;

public class MainTest {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		Factory f = new Factory();
		f.start();
		Kid k = new Kid();
		k.start();
		new DoubleBufferList(Tools.lP,Tools.lT,1).check();
	}

}

同样100000个玩具对象，使用的时间是3469ms，大约节约了200ms

 

在使用双缓冲队列的时候，要注意调试监测线程的监控时间间隔和阻塞队列的大小，这些都是影响运行速度的关键因素。

 

评论

3 楼 大隐于市 2011-09-13  

既然用了同步，就没必要再用LinkedBlockingQueue了吧，用普通的list效率会更高。

2 楼 Irving_wei 2009-09-16  

sahero 写道

本人有两个问题:
1.博主提出的性能问题是不是由于lT.take(),lT.put()都作用于同一个对象造成的?
2.对于
if(lT.size()==0){  
                lT.addAll(lP);  
                lP.clear();  
            }  
当程序已经执行了lT.addAll(lP)还未执行lP.clear(),这时候如果lP.put(t)又放进了一个对象,那么这个对象不就会被清理掉了?

呵呵，第一点，我个人觉得性能问题是由于lT.take(),lT.put()都作用于同一个对象造成的。
      第二点，程序应该这样改下才合适    if(lT.size()==0){
synchronized(lT){
synchronized(lP){
lT.addAll(lP);
}
lP.clear();
}
}
谢谢提醒啊。我现在正在做一个关于CMPP的模拟程序，关于改善线程同步来提高性能方面也在考虑当中，希望有机会可以和你讨论讨论啊。

1 楼 sahero 2009-09-16  

本人有两个问题:
1.博主提出的性能问题是不是由于lT.take(),lT.put()都作用于同一个对象造成的?
2.对于
if(lT.size()==0){  
                lT.addAll(lP);  
                lP.clear();  
            }  
当程序已经执行了lT.addAll(lP)还未执行lP.clear(),这时候如果lP.put(t)又放进了一个对象,那么这个对象不就会被清理掉了?