`
dingjob
  • 浏览: 183654 次
  • 性别: Icon_minigender_1
  • 来自: 武汉
社区版块
存档分类
最新评论

双向缓冲列表用法

thread设计模式中国移动F#工作 
阅读更多

提出背景：
   在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();
}
}
}
}

 

Java代码 复制代码
  1. package cn.netjava.lbq.copy;   
  2. //从单缓冲队列中取出玩具对象   
  3.   
  4. public class Kid extends Thread {   
  5.     long time1 = System.currentTimeMillis();   
  6.     int count = 0;   
  7.     public void run() {   
  8.         while(true){   
  9.         try {   
  10.             Tools.lT.take();   
  11.             count++;   
  12. //计算取100000个所需要的时间   
  13.             if(count==100000){   
  14.                 System.out.println(System.currentTimeMillis()-time1);   
  15.                 Thread.sleep(100000);   
  16.             }   
  17.             System.out.println("玩完一个玩具");   
  18.         } catch (InterruptedException e) {   
  19.             e.printStackTrace();   
  20.         }   
  21.         }   
  22.         }   
  23.   
  24. }  
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();
		}
		}
		}

}

 

Java代码 复制代码
  1. package cn.netjava.lbq.copy;   
  2. //存放玩具对象的的队列   
  3. import java.util.concurrent.LinkedBlockingQueue;   
  4.   
  5. public class Tools {   
  6.     public static LinkedBlockingQueue<Tool>lT = new LinkedBlockingQueue<Tool>(10000);   
  7.        
  8. }  
package cn.netjava.lbq.copy;
//存放玩具对象的的队列
import java.util.concurrent.LinkedBlockingQueue;

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

 

Java代码 复制代码
  1. package cn.netjava.lbq.copy;   
  2.   
  3. public class MainTest {   
  4.   
  5.     /**  
  6.      * @param args  
  7.      */  
  8.     public static void main(String[] args) {   
  9.         Factory f = new Factory();   
  10.         f.start();   
  11.         Kid k = new Kid();   
  12.         k.start();   
  13.            
  14.     }   
  15.   
  16. }  
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类就不再贴出了。

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

Java代码 复制代码
  1. package cn.netjava.lbq;   
  2.   
  3. import java.util.concurrent.LinkedBlockingQueue;   
  4.   
  5. public class Tools {   
  6.     public static LinkedBlockingQueue<Tool>lT = new LinkedBlockingQueue<Tool>(10000);   
  7.     public static LinkedBlockingQueue<Tool>lP = new LinkedBlockingQueue<Tool>(10000);   
  8. }  
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);
}

 

Java代码 复制代码
  1. package cn.netjava.lbq;   
  2.   
  3.   
  4. public class Kid extends Thread {   
  5.     long time1 = System.currentTimeMillis();   
  6.     int count = 0;   
  7.     public void run() {   
  8.         while(true){   
  9.         try {   
  10.              Tools.lT.take();   
  11.             count++;   
  12.                
  13.             if(count==10000){   
  14.             System.out.println(System.currentTimeMillis()-time1);   
  15.             Thread.sleep(100000);   
  16.             }   
  17.             System.out.println("玩完一个玩具");   
  18.         } catch (InterruptedException e) {   
  19.             e.printStackTrace();   
  20.         }   
  21.         }   
  22.         }   
  23.   
  24. }  
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();
		}
		}
		}

}

 

Java代码 复制代码
  1. package cn.netjava.lbq;   
  2.   
  3. import java.util.concurrent.LinkedBlockingQueue;   
  4.   
  5. public class Factory extends Thread{   
  6.        
  7.   
  8.     public void run(){   
  9.         while(true){   
  10.         Toy t = new Toy ();   
  11.         t.setName("玩具");   
  12.         try {   
  13.             Tools.lP.put(t);   
  14.             System.out.println("生产出一个玩具");   
  15.         } catch (InterruptedException e) {   
  16.             e.printStackTrace();   
  17.         }   
  18.          }   
  19.     }   
  20. }  
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();
	 	}
		 }
	}
}

 

Java代码 复制代码
  1. package cn.netjava.lbq;   
  2.   
  3. mport java.util.concurrent.LinkedBlockingQueue;   
  4. /**  
  5.  * 定义一个双缓冲队列  
  6.  * @author Irving  
  7.  * @time  2009-9-16----上午10:25:49  
  8.  *  
  9.  */  
  10. public class DoubleBufferList {   
  11.   
  12.     private LinkedBlockingQueue<Object> lP;   
  13.     private LinkedBlockingQueue<Object> lT;   
  14.     private int gap;   
  15.     /**  
  16.      * 构造方法  
  17.      * @param lP 用来存放对象的阻塞队列  
  18.      * @param lT 用来取对象的阻塞队列  
  19.      * @param gap 交换的间隔  
  20.      */  
  21.        
  22.     public DoubleBufferList(LinkedBlockingQueue  lP   
  23.             ,LinkedBlockingQueue  lT,int gap){   
  24.         this.lP=lP;   
  25.         this.lT=lT;   
  26.         this.gap=gap;   
  27.     }   
  28.        
  29.     public void check(){   
  30.         Runnable runner = new Runnable(){   
  31.             public void run() {   
  32.                 while(true){   
  33.             if(lT.size()==0){   
  34.                 synchronized(lT){   
  35.                 synchronized(lP){   
  36.                 lT.addAll(lP);   
  37.                 }   
  38.                 lP.clear();   
  39.                 }   
  40.             }               try {   
  41.                 Thread.sleep(gap);   
  42.             } catch (InterruptedException e) {   
  43.                 e.printStackTrace();   
  44.             }   
  45.                 }      
  46.                    
  47.             }                  
  48.         };   
  49.         Thread thread = new Thread(runner);   
  50.         thread.start();   
  51.     }   
  52.        
  53. }  
	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();
		}
		
	}

 

Java代码 复制代码
  1. package cn.netjava.lbq;   
  2.   
  3. public class MainTest {   
  4.   
  5.     /**  
  6.      * @param args  
  7.      */  
  8.     public static void main(String[] args) {   
  9.         Factory f = new Factory();   
  10.         f.start();   
  11.         Kid k = new Kid();   
  12.         k.start();   
  13.         new DoubleBufferList(Tools.lP,Tools.lT,1).check();   
  14.     }   
  15.   
  16. }  
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

 

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

分享到：
| 基于Lucene的商业搜索应用架构研究(上)— ...
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    STM32 环形缓冲

    串口自发自收（Self-Send and Receive）意味着STM32可以同时作为发送端和接收端，通过环形缓冲实现数据的双向传输。 3. **C程序实现**：在C语言中，可以定义一个固定大小的数组来创建环形缓冲，并用两个整数变量来...

    内存缓冲池的实现

    若要在整个程序中使用内存缓冲池，需要重定义全局的`operator new`和`operator delete`，以及对应的数组版本。这样，当代码中的`new`和`delete`调用执行时，它们会自动使用内存缓冲池。此外，为了获取分配时的位置...

    单片机之间双向通信

    本文将详细探讨如何实现单片机之间的双向通信，主要以甲机和乙机为例，并使用Keil C编程语言进行阐述。 首先，我们需要理解什么是双向通信。双向通信是指两个或多个设备之间可以同时进行数据发送和接收的过程，它...

    Windows同步管道双向通信

    在Windows操作系统中，同步管道是一种进程间通信（IPC，...通过分析和理解这些源代码，开发者可以更深入地学习和掌握Windows同步管道双向通信的工作原理和实现方法，这对于进行多进程间的高效数据交互具有重要意义。

    Android 通话双向录音

    这通常涉及到对音频硬件的独占访问，因此可能需要使用到AudioManager的setMode()方法，将其设置为MODE_IN_COMMUNICATION模式，以便在通话期间使用音频硬件。 ```java audioManager = (AudioManager) ...

    iperf使用方法及流程介绍.docx

    - `-l &lt;size&gt;`：设置发送数据的缓冲区大小，默认为8KB。 - `-m`：显示TCP的最大MTU值。 - `-o &lt;file&gt;`：将输出信息重定向到指定文件。 - `-p &lt;port&gt;`：指定服务器端或客户端使用的端口号。 - `-u`：使用UDP协议而非...

    vc 实现 双列队缓冲池

    `DLinkBuffer.cpp`和`DLinkBuffer.h`可能包含了双列队缓冲池的核心实现，其中`DLinkBuffer`可能是缓冲池类的名称，使用双向链表作为底层数据结构。双向链表允许在链表头尾进行快速插入和删除，方便内存块的管理和...

    C#词法分析器之输入缓冲和代码定位的应用分析

    为有效管理这些缓冲区，作者采用了双向链表的形式，将多个缓冲区链接成一个环。链表中的每个节点代表一个缓冲区，三个指针current、first和last分别指向当前正在读取的缓冲区、最早的缓冲区和最新的缓冲区。通过移动...

    缓冲器安装方法

    二、使用这种方法通常是因为不愿意把闭门器安装在建筑物之外的朝外开启的外门上，从而保持外观的整洁：是把闭门器安装在与铰链侧相反的、门关闭方向的一面。此外，如果门的上沿很窄，导致没有足够的空余间容纳要安装...

    php实现的双向队列类实例

    代码中使用了数组来存储队列元素，并且有相应的方法来操作数组，确保操作的正确性和数据结构的完整性。代码还考虑了对队列长度的限制，当队列达到最大长度后，将不允许进一步的添加操作。 通过以上详细的知识点解析...

    深入浅出linux内核源代码之双向链表list_head(list.h).pdf

    ### 深入解析Linux内核中的双向链表机制 #### 一、双向链表的基础概念与作用 双向链表是一种常见的数据结构，在...无论是对于Linux内核的贡献者还是对C语言学习者而言，掌握双向链表的概念和使用方法都是十分重要的。

    27 串口1-232与TCP服务器双向通信_串口1-232与TCP服务器双向通信_串口TCP_broken8nf_stm32f4

    "broken8nf"可能是指一个特定的开发板或项目名，它也可能暗示了某种特定的编程或调试方法。我们首先理解串口1-232与TCP/IP协议的基础，然后介绍如何在STM32F4平台上实现它们的交互。 串口1-232（UART1）是STM32F4微...

    LRU.rar_LRU缓冲

    6. 效率优化：为了提高性能，LRU算法的实现可能会采用一些技巧，例如使用双向链表结合哈希表，以支持O(1)时间复杂度的插入和查找操作。 7. 系统调优：在实际应用中，数据库管理员需要根据工作负载和硬件资源调整LRU...

    mfc聊天程序，利用TCP/IP完成双向数据收发

    为了保证聊天的流畅性，可以使用缓冲区来暂存待发送或已接收到的数据，并使用适当的错误处理机制来应对网络异常。此外，为了提高用户体验，还可以添加消息提示、用户界面更新等功能。 总的来说，这个"Mfc聊天程序，...

    UDP.zip_javaUDP双向收发

    总结来说，Java UDP双向收发涉及到的主要知识点包括：`DatagramSocket`的使用、`DatagramPacket`的构建和发送、接收数据的处理，以及多线程技术以实现双向通信。理解和实践这些内容对于进行Java网络编程至关重要。

    Go-buffstreams-通过TCP实现流协议缓冲消息

    4. `BufferManager`：管理缓冲区的分配和释放，优化内存使用。 项目结构可能如下： - `buffstreams-master` - `server`：服务器端代码，包括主程序和服务器逻辑。 - `client`：客户端代码，包括主程序和客户端...

    双机通信-实现双向通信功能

    我们将主要关注使用流式套接字（Stream Sockets）通过TCP/IP协议进行通信的方法。 首先，TCP/IP协议是互联网上最常用的一种通信协议，它由四层组成：链路层、网络层、传输层和应用层。在这个案例中，我们主要关注...

    spi.rar_SPI VHDL_VHDL SPI 双向_spi_spi 双向通讯_spi函数

    同时，需要定义数据缓冲区来存储待发送和接收的数据，并使用计数器来跟踪时钟周期。MISO和MOSI信号通过数据寄存器和时钟同步逻辑来处理，而CS信号则根据当前的通信需求进行控制。 双向SPI通讯： 常规的SPI协议通常...

Magicbox
Global site tag (gtag.js) - Google Analytics