devbox
天景科技苑
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

双队列的一种实现

作者：天景科技苑 | 2024-07-27 21:49:44

赞

踩

两类pcd队列csdn

介绍

双队列是一种高效的内存数据结构，在多线程编程中，能保证生产者线程的写入和消费者的读出尽量做到最低的影响，避免了共享队列的锁开销。本文将介绍一种双队列的设计，并给出实现代码，然后会举例使用的场景。该双队列在项目中使用，性能也得到了验证。

设计

接下来具体介绍双队列的设计，并且会粘贴少量方法代码，帮助介绍。
本文中讲述的双队列，本质上是两个数组保存写入的Object，一个数组负责写入，另一个被消费者读出，两个数组都对应一个重入锁。数组内写入的数据会被计数。 
  1. public class DoubleCachedQueue<T> extends AbstractQueue<T> implements
  2. 		BlockingQueue<T>, java.io.Serializable {
  3. 	private static final long serialVersionUID = 1L;
  4. 	
  5. 	private static int default_line_limit = 1000;
  6. 	
  7. 	private static long max_cache_size = 67108864L;
  8.  
  9. 	private int lineLimit;
  10. 	
  11. 	private long cacheSize;
  12.  
  13. 	private T[] itemsA;
  14.  
  15. 	private T[] itemsB;
  16.  
  17. 	private ReentrantLock readLock, writeLock;
  18.  
  19. 	private Condition notFull;
  20.  
  21. 	private Condition awake;
  22.  
  23. 	/**
  24. 	 * writeArray : in reader's eyes, reader get data from data source and write
  25. 	 * data to this line array. readArray : in writer's eyes, writer put data to
  26. 	 * data destination from this line array.
  27. 	 * 
  28. 	 * Because of this is doubleQueue mechanism, the two line will exchange when
  29. 	 * time is suitable.
  30. 	 * 
  31. 	 */
  32. 	private T[] writeArray, readArray;
  33.  
  34. 	private volatile int writeCount, readCount;
  35.  
  36. 	private int writeArrayTP, readArrayHP;
  37. 	
  38. 	private volatile boolean closed = false;
  39.  
  40. 	private int spillSize = 0;
  41.  
  42. 	private long lineRx = 0;
  43.  
  44. 	private long lineTx = 0;

队列实现了阻塞队列的接口，所以在向队列offer数据的时候是阻塞的，同样，取出操作poll也会阻塞。两个数组会在适当的时候进行queueSwitch操作。queueSwitch的条件就是当读者把queue读空了之后，且写入的queue此时不为空的时候，两个queue就会进行交换。在交换的时候，写入queue会被上锁，此时生产者不能让队列里写入数据。一般情况下，queue互换其实就是两个数组的引用互换，将相应的计数器也重置，写队列的计数器此时就清零了，因为queue交换是因为读队列已经被读空。 
  1. private long queueSwitch(long timeout, boolean isInfinite)
  2. 			throws InterruptedException {
  3. 		System.out.println("queue switch");
  4. 		writeLock.lock();
  5. 		try {
  6. 			if (writeCount <= 0) {
  7. 				if (closed) {
  8. 					return -2;
  9. 				}
  10. 				try {
  11. 					if (isInfinite && timeout <= 0) {
  12. 						awake.await();
  13. 						return -1;
  14. 					} else {
  15. 						return awake.awaitNanos(timeout);
  16. 					}
  17. 				} catch (InterruptedException ie) {
  18. 					awake.signal();
  19. 					throw ie;
  20. 				}
  21. 			} else {
  22. 				T[] tmpArray = readArray;
  23. 				readArray = writeArray;
  24. 				writeArray = tmpArray;
  25.  
  26. 				readCount = writeCount;
  27. 				readArrayHP = 0;
  28.  
  29. 				writeCount = 0;
  30. 				writeArrayTP = 0;
  31.  
  32. 				notFull.signal();
  33. 				// logger.debug("Queue switch successfully!");
  34. 				return -1;
  35. 			}
  36. 		} finally {
  37. 			writeLock.unlock();
  38. 		}
  39. 	}

上面queue交换的时候，可以看到当要被交换的写队列也已经为空的时候，会做一次检查。如果此时queue已经被显示地关闭了，那么poll操作就会返回空，读者此时应该检查queue是否已经被closed了，若已经closed了，那么读者已经把queue里的数据读完了。这里的显示close是我们给双队列加的一个状态，close这件事的作用是为了让读者知道：生产者已经停止往queue里写新数据了，但是queue里其实可能还有未取完的数据（在写queue里，此时还差一次queue switch），你往queue poll取数据的时候，如果取到空了，那么应该做一次check，如果queue已经关闭了，那么读者就知道本次读的任务完全结束了。反过来，close状态其实不影响写，生产者如果还想写的话，其实也是可以的，但是我不推荐这么做。 
  1. public void close() {
  2. 		writeLock.lock();
  3. 		try {
  4. 			closed = true;
  5. 			//System.out.println(this);
  6. 			awake.signalAll();
  7. 		} finally {
  8. 			writeLock.unlock();
  9. 		}
  10. 	}

如果没有这个close标志位的话，可能就需要消费者放入一个EOF让读者知道。这在只有一个生产者和一个消费者的情况下是可行的，但是如果是一个多对一，一对多，甚至多对多的情况呢？一对一的情况是最简单的，也是双队列被创造出来最合适的场景。因为双队列完全分离了一个生产者和一个消费者的锁争抢情况，各自只要获得自己的读/写队列的锁就可以了。在本文阐述的双队列中，唯一产生一些开销的就是queue swtich的情况，如果queue频繁交换的话，还是会产生一些性能开销的。


一对多

上面已经大致介绍了双队列的读写。在实际项目中，一对多的场景需要注意的地方有两：
  • 单个生产者需要在结束的时候关闭queue
  • 多个消费者需要知道任务结束（知道其他线程已经完成任务）
第一点很简单，比如读文件的话，当生产者readLine()最后为空的时候，就认为数据源已经读完，调用方法把queue close()。而消费者在读queue的时候，有时候可能会由于延迟、queue交换等原因取到空数据，此时就如上面一节所说，消费者线程拿到空数据后应该检查queue的状态，如果queue没有关闭，那么应该等待一小会儿后继续poll数据；如果queue关闭了，那么其实说明该线程已经完成了任务。同理，其他消费者线程也应该在取到空的时候做这样的操作。


消费者之间或者外部有一方需要知道各个消费者线程的存活情况，这样才能知道本次任务完成。比如如果外面有一个上帝的话，可以加一个CountDownLatch计数，每个消费者完成后就countDown一次，外部调用await()直到大家都已经退出，那么整个任务结束。如果没有上帝，线程之间互相知道对方情况的话，我的做法是让生产者放入一个EOF，当某线程取到EOF的时候，他知道自己是第一个遇到尽头的人，他会置一个布尔，而其他线程在取到空的时候会检查该布尔值，这样就能知道是否已经有小伙伴已经拿到EOF了，那么这时候就可以countDown了，而拿到EOF的线程进程countDown后就await()，最后退出。
下面是我自己针对这种场景，使用双队列的方式，其中的fromQueue是一个ConcurrentLinkedQueue，大家可以忽略，toQueue是双队列，可以注意一下用法。特别是往里面写的时候，需要while循环重试直到写入成功。 
  1. @Override
  2. public void run() {
  3. 	long start = System.currentTimeMillis();
  4. 	log.debug(Thread.currentThread() + " Unpacker started at " + start);
  5. 		
  6. 	Random r = new Random(start);
  7. 	Bundle bundle = null;
  8. 	boolean shoudShutdown = false;
  9. 	try {
  10. 		while(!shoudShutdown) {
  11. 			bundle = (Bundle) fromQueue.poll();
  12. 			if (bundle == null) {
  13. 				if (seeEOF.get()) {
  14. 					// 当取到空，并且其他线程已经取到EOF，那么本线程将Latch减1，并退出循环
  15. 					latch.countDown();
  16. 					shoudShutdown = true;
  17. 				} else {
  18. 					// 如果EOF还没被取到，本线程小睡一会后继续取
  19. 					try {
  20. 						sleep(r.nextInt(10));
  21. 					} catch (InterruptedException e) {
  22. 						log.error("Interrupted when taking a nap", e);
  23. 					}
  24. 				}
  25. 			} else if (!bundle.isEof()) {
  26. 			// bundle非空且非EOF，则往双队列写入一个Bundle
  27. 			byte[] lineBytes = BundleUtil.getDecompressedData(bundle);
  28. 			// 放入双队列时，若offer失败则重试
  29. 			while (!toQueue.offer(new UnCompressedBundle(bundle.getId(), ByteUtil.bytes2Lines(lineBytes, lineDelim), bundle.getIndex(), bundle.getJobId()))) {
  30. 				log.info("Unpacker put failed, will retry");
  31. 			}
  32. 			log.info("After enqueue, queue size is " + toQueue.size());
  33. 			} else {
  34. 				// Unpacker获得到了EOF
  35. 				seeEOF.set(true);
  36. 				// 自己将Lacth减1，并等待其他线程退出
  37. 				latch.countDown();
  38. 				try {
  39. 					latch.await();
  40. 				} catch (InterruptedException e) {
  41. 					log.error("Interrupted when waiting the latch ");
  42. 				}
  43. 				// 其他线程已经退出，本线程放入EOF
  44. 				while (!toQueue.offer(new UnCompressedBundle(-1L, new Line[0], -1L, -1L))) {
  45. 					log.info("Unpacker put EOF failed, will retry");
  46. 				}
  47. 				// 关闭Queue
  48. 				toQueue.close();
  49. 				// 退出循环
  50. 				shoudShutdown = true;
  51. 			}
  52. 		}
  53. 		log.debug(Thread.currentThread() + " Unpacker finished in " + (System.currentTimeMillis()-start) + " ms");
  54. 	} catch (Exception e) {
  55. 		log.error("Exception when unpacker is running ", e);
  56. 		// 将latch减1，表示自己异常退出，且不再工作
  57. 		// latch.countDown();
  58. 		log.debug(Thread.currentThread() + " Unpacker occured exception and stopped. ");
  59. 	} finally {
  60. 		
  61. 	}
  62. }

多对一

多个生产者的情况下，写入队列无可避免发送锁争抢，但是能保证消费者的稳定读出过程。没有什么特殊处理的地方，这里就不啰嗦了。

总结分析

本文介绍了一种经典双队列的设计和实现，也给出了一些代码演示。文章末尾我会贴出整个双队列的代码实现，需要的同学也可以留言，我把.java发给你。如果使用的时候有发现问题，不吝赐教，这个双队列的实现也还不是很完美。使用的时候也存在需要注意的地方。
其实双队列的目的还是在于让写和读互相没有影响，而且更加照顾了写的速度。因为一般写的速度可能会比较快，而读的人读出之后还会做一些额外的处理，所以写的这一方借助双队列，可以持续写的过程，而且如果读的一方慢的话，可以多起几个消费者线程，就像"一对多"场景里阐述的那样来使用双队列。

下面是整个实现。各位可以仔细看看，发现问题一定记得通知我 :) 

  1. import java.util.AbstractQueue;
  2. import java.util.Collection;
  3. import java.util.Iterator;
  4. import java.util.concurrent.BlockingQueue;
  5. import java.util.concurrent.TimeUnit;
  6. import java.util.concurrent.locks.Condition;
  7. import java.util.concurrent.locks.ReentrantLock;
  8.  
  9.  
  10. import lombok.ToString;
  11. import lombok.extern.log4j.Log4j;
  12.  
  13.  
  14. /**
  15.  * Represents a region with two swap spaces, one for storing data which from
  16.  * data source, the other one for storing data which will be transferred to data
  17.  * destination.
  18.  * <br>
  19.  * A classical DoubleCachedQueue, In beginning, space A and space B both
  20.  * empty, then loading task begin to load data to space A, when A is almost
  21.  * full, let the data from data source being loaded to space B, then dumping
  22.  * task begin to dump data from space A to data source. When space A is empty,
  23.  * switch the two spaces for load and dump task. Repeat the above operation.
  24.  * 
  25.  */
  26. @Log4j
  27. @ToString
  28. public class DoubleCachedQueue<T> extends AbstractQueue<T> implements
  29. 		BlockingQueue<T>, java.io.Serializable {
  30. 	private static final long serialVersionUID = 1L;
  31. 	
  32. 	private static int default_line_limit = 1000;
  33. 	
  34. 	private static long max_cache_size = 67108864L;
  35.  
  36. 	private int lineLimit;
  37. 	
  38. 	private long cacheSize;
  39.  
  40. 	private T[] itemsA;
  41.  
  42. 	private T[] itemsB;
  43.  
  44. 	private ReentrantLock readLock, writeLock;
  45.  
  46. 	private Condition notFull;
  47.  
  48. 	private Condition awake;
  49.  
  50. 	/**
  51. 	 * writeArray : in reader's eyes, reader get data from data source and write
  52. 	 * data to this line array. readArray : in writer's eyes, writer put data to
  53. 	 * data destination from this line array.
  54. 	 * 
  55. 	 * Because of this is doubleQueue mechanism, the two line will exchange when
  56. 	 * time is suitable.
  57. 	 * 
  58. 	 */
  59. 	private T[] writeArray, readArray;
  60.  
  61. 	private volatile int writeCount, readCount;
  62.  
  63. 	private int writeArrayTP, readArrayHP;
  64. 	
  65. 	private volatile boolean closed = false;
  66.  
  67. 	private int spillSize = 0;
  68.  
  69. 	private long lineRx = 0;
  70.  
  71. 	private long lineTx = 0;
  72.  
  73. 	/**
  74. 	 * Get info of line number in {@link DoubleCachedQueue} space.
  75. 	 * 
  76. 	 * @return Information of line number.
  77. 	 * 
  78. 	 */
  79. 	public String info() {
  80. 		return String.format("Write Array: %s/%s; Read Array: %s/%s", writeCount, writeArray.length, readCount, readArray.length);
  81. 	}
  82.  
  83. 	/**
  84. 	 * Use the two parameters to construct a {@link DoubleCachedQueue} which hold the
  85. 	 * swap areas.
  86. 	 * 
  87. 	 * @param lineLimit
  88. 	 *            Limit of the line number the {@link DoubleCachedQueue} can hold.
  89. 	 * 
  90. 	 * @param byteLimit
  91. 	 *            Limit of the bytes the {@link DoubleCachedQueue} can hold.
  92. 	 * 
  93. 	 */
  94. 	public DoubleCachedQueue(int lineLimit) {
  95. 		if (lineLimit <= 0) {
  96. 			this.lineLimit = default_line_limit;
  97. 		}else{
  98. 			this.lineLimit = lineLimit;
  99. 		}
  100. 		itemsA = (T[])new Object[lineLimit];
  101. 		itemsB = (T[])new Object[lineLimit];
  102.  
  103. 		readLock = new ReentrantLock();
  104. 		writeLock = new ReentrantLock();
  105.  
  106. 		notFull = writeLock.newCondition();
  107. 		awake = writeLock.newCondition();
  108.  
  109. 		readArray = itemsA;
  110. 		writeArray = itemsB;
  111. 		spillSize = lineLimit * 8 / 10;
  112. 	}
  113. 	
  114. 	public DoubleCachedQueue(long cacheSize){
  115. 		if (cacheSize <= 0) {
  116. 			throw new IllegalArgumentException(
  117. 					"Queue initial capacity can't less than 0!");
  118. 		}
  119. 		this.cacheSize = cacheSize > max_cache_size ? max_cache_size : cacheSize;
  120. 		
  121. 		readLock = new ReentrantLock();
  122. 		writeLock = new ReentrantLock();
  123.  
  124. 		notFull = writeLock.newCondition();
  125. 		awake = writeLock.newCondition();
  126.  
  127. 		readArray = itemsA;
  128. 		writeArray = itemsB;
  129. 		spillSize = lineLimit * 8 / 10;
  130. 	}
  131.  
  132. 	/**
  133. 	 * Get line number of the {@link DoubleCachedQueue}
  134. 	 * 
  135. 	 * @return lineLimit Limit of the line number the {@link DoubleCachedQueue} can
  136. 	 *         hold.
  137. 	 * 
  138. 	 */
  139. 	public int getLineLimit() {
  140. 		return lineLimit;
  141. 	}
  142.  
  143. 	/**
  144. 	 * Set line number of the {@link DoubleCachedQueue}.
  145. 	 * 
  146. 	 * @param capacity
  147. 	 *            Limit of the line number the {@link DoubleCachedQueue} can hold.
  148. 	 * 
  149. 	 */
  150. 	public void setLineLimit(int capacity) {
  151. 		this.lineLimit = capacity;
  152. 	}
  153.  
  154. 	/**
  155. 	 * Insert one line of record to a apace which buffers the swap data.
  156. 	 * 
  157. 	 * @param line
  158. 	 *            The inserted line.
  159. 	 * 
  160. 	 */
  161. 	private void insert(T line) {
  162. 		writeArray[writeArrayTP] = line;
  163. 		++writeArrayTP;
  164. 		++writeCount;
  165. 		++lineRx;
  166. 	}
  167.  
  168. 	/**
  169. 	 * Insert a line array(appointed the limit of array size) of data to a apace
  170. 	 * which buffers the swap data.
  171. 	 * 
  172. 	 * @param lines
  173. 	 *            Inserted line array.
  174. 	 * 
  175. 	 * @param size
  176. 	 *            Limit of inserted size of the line array.
  177. 	 * 
  178. 	 */
  179. 	private void insert(T[] lines, int size) {
  180. 		if(size > 0){
  181. 			System.arraycopy(lines, 0, writeArray, writeArrayTP, size);
  182. 			writeArrayTP = writeArrayTP + size;
  183. 			writeCount = writeCount + size;
  184. 			lineRx = lineRx + size;
  185. 		}
  186. //		for (int i = 0; i < size; ++i) {
  187. //			writeArray[writeArrayTP] = lines[i];
  188. //			++writeArrayTP;
  189. //			++writeCount;
  190. //			++lineRx;
  191. //			if(lines[i] != null && lines[i].getLine() != null){
  192. //				byteRx += lines[i].getLine().length();
  193. //			}
  194. //		}
  195. 	}
  196.  
  197. 	/**
  198. 	 * Extract one line of record from the space which contains current data.
  199. 	 * 
  200. 	 * @return line A line of data.
  201. 	 * 
  202. 	 */
  203. 	private T extract() {
  204. 		T e = readArray[readArrayHP];
  205. 		readArray[readArrayHP] = null;
  206. 		++readArrayHP;
  207. 		--readCount;
  208. 		++lineTx;
  209. 		return e;
  210. 	}
  211.  
  212. 	/**
  213. 	 * Extract a line array of data from the space which contains current data.
  214. 	 * 
  215. 	 * @param ea
  216. 	 * @return Extracted line number of data.
  217. 	 * 
  218. 	 */
  219. 	private int extract(T[] ea) {
  220. 		int readsize = Math.min(ea.length, readCount);
  221. 		if(readsize > 0){
  222. 			readCount = readCount - readsize;
  223. 			lineTx = lineTx + readsize;
  224. 			System.arraycopy(readArray, readArrayHP, ea, 0, readsize);
  225. 			readArrayHP = readArrayHP + readsize;
  226. 		}
  227. //		for (int i = 0; i < readsize; ++i) {
  228. //			ea[i] = readArray[readArrayHP];
  229. //			readArray[readArrayHP] = null;
  230. //			++readArrayHP;
  231. //			--readCount;
  232. //			++lineTx;
  233. //		}
  234. 		return readsize;
  235. 	}
  236.  
  237. 	/**
  238. 	 * switch condition: read queue is empty && write queue is not empty.
  239. 	 * Notice:This function can only be invoked after readLock is grabbed,or may
  240. 	 * cause dead lock.
  241. 	 * 
  242. 	 * @param timeout
  243. 	 * 
  244. 	 * @param isInfinite
  245. 	 *            whether need to wait forever until some other thread awake it.
  246. 	 * 
  247. 	 * @return
  248. 	 * 
  249. 	 * @throws InterruptedException
  250. 	 * 
  251. 	 */
  252.  
  253. 	private long queueSwitch(long timeout, boolean isInfinite)
  254. 			throws InterruptedException {
  255. 		System.out.println("queue switch");
  256. 		writeLock.lock();
  257. 		try {
  258. 			if (writeCount <= 0) {
  259. 				if (closed) {
  260. 					return -2;
  261. 				}
  262. 				try {
  263. 					if (isInfinite && timeout <= 0) {
  264. 						awake.await();
  265. 						return -1;
  266. 					} else {
  267. 						return awake.awaitNanos(timeout);
  268. 					}
  269. 				} catch (InterruptedException ie) {
  270. 					awake.signal();
  271. 					throw ie;
  272. 				}
  273. 			} else {
  274. 				T[] tmpArray = readArray;
  275. 				readArray = writeArray;
  276. 				writeArray = tmpArray;
  277.  
  278. 				readCount = writeCount;
  279. 				readArrayHP = 0;
  280.  
  281. 				writeCount = 0;
  282. 				writeArrayTP = 0;
  283.  
  284. 				notFull.signal();
  285. 				// logger.debug("Queue switch successfully!");
  286. 				return -1;
  287. 			}
  288. 		} finally {
  289. 			writeLock.unlock();
  290. 		}
  291. 	}
  292.  
  293. 	/**
  294. 	 * If exists write space, it will return true, and write one line to the
  295. 	 * space. otherwise, it will try to do that in a appointed time,when time is
  296. 	 * out if still failed, return false.
  297. 	 * 
  298. 	 * @param line
  299. 	 *            a Line.
  300. 	 * 
  301. 	 * @param timeout
  302. 	 *            appointed limit time
  303. 	 * 
  304. 	 * @param unit
  305. 	 *            time unit
  306. 	 * 
  307. 	 * @return True if success,False if failed.
  308. 	 * 
  309. 	 */
  310. 	public boolean offer(T line, long timeout, TimeUnit unit)
  311. 			throws InterruptedException {
  312. 		if (line == null) {
  313. 			throw new NullPointerException();
  314. 		}
  315. 		long nanoTime = unit.toNanos(timeout);
  316. 		writeLock.lockInterruptibly();
  317. 		if(itemsA == null || itemsB == null){
  318. 			initArray(line);
  319. 		}
  320. 		try {
  321. 			for (;;) {
  322. 				if (writeCount < writeArray.length) {
  323. 					insert(line);
  324. 					if (writeCount == 1) {
  325. 						awake.signal();
  326. 					}
  327. 					return true;
  328. 				}
  329.  
  330. 				// Time out
  331. 				if (nanoTime <= 0) {
  332. 					return false;
  333. 				}
  334. 				// keep waiting
  335. 				try {
  336. 					nanoTime = notFull.awaitNanos(nanoTime);
  337. 				} catch (InterruptedException ie) {
  338. 					notFull.signal();
  339. 					throw ie;
  340. 				}
  341. 			}
  342. 		} finally {
  343. 			writeLock.unlock();
  344. 		}
  345. 	}
  346.  
  347. 	private void initArray(T line) {
  348. 		
  349. 		long recordLength = computeSize(line);
  350. 		long size = cacheSize/recordLength;
  351. 		if(size <= 0){
  352. 			size = default_line_limit;
  353. 		}
  354. 		lineLimit = (int) size;
  355. 		itemsA = (T[])new Object[(int) size];
  356. 		itemsB = (T[])new Object[(int) size];
  357. 		readArray = itemsA;
  358. 		writeArray = itemsB;
  359. 		
  360. 	}
  361.  
  362. 	public long computeSize(T line){
  363. 		return 1;
  364. 	}
  365.  
  366. 	/**
  367. 	 * If exists write space, it will return true, and write a line array to the
  368. 	 * space.<br>
  369. 	 * otherwise, it will try to do that in a appointed time,when time out if
  370. 	 * still failed, return false.
  371. 	 * 
  372. 	 * @param lines
  373. 	 *            line array contains lines of data
  374. 	 * 
  375. 	 * @param size
  376. 	 *            Line number needs to write to the space.
  377. 	 * 
  378. 	 * @param timeout
  379. 	 *            appointed limit time
  380. 	 * 
  381. 	 * @param unit
  382. 	 *            time unit
  383. 	 * 
  384. 	 * @return status of this operation, true or false.
  385. 	 * 
  386. 	 * @throws InterruptedException
  387. 	 *             if being interrupted during the try limit time.
  388. 	 * 
  389. 	 */
  390. 	public boolean offer(T[] lines, int size, long timeout, TimeUnit unit)
  391. 			throws InterruptedException {
  392. 		if (lines == null || lines.length == 0) {
  393. 			throw new NullPointerException();
  394. 		}
  395. 		long nanoTime = unit.toNanos(timeout);
  396. 		writeLock.lockInterruptibly();
  397. 		if(itemsA == null || itemsB == null){
  398. 			initArray(lines[0]);
  399. 		}
  400. 		try {
  401. 			for (;;) {
  402. 				if (writeCount + size <= writeArray.length) {
  403. 					insert(lines, size);
  404. 					if (writeCount >= spillSize) {
  405. 						awake.signalAll();
  406. 					}
  407. 					return true;
  408. 				}
  409.  
  410. 				// Time out
  411. 				if (nanoTime <= 0) {
  412. 					return false;
  413. 				}
  414. 				// keep waiting
  415. 				try {
  416. 					nanoTime = notFull.awaitNanos(nanoTime);
  417. 				} catch (InterruptedException ie) {
  418. 					notFull.signal();
  419. 					throw ie;
  420. 				}
  421. 			}
  422. 		} finally {
  423. 			writeLock.unlock();
  424. 		}
  425. 	}
  426.  
  427. 	/**
  428. 	 * Close the synchronized lock and one inner state.
  429. 	 * 
  430. 	 */
  431. 	public void close() {
  432. 		writeLock.lock();
  433. 		try {
  434. 			closed = true;
  435. 			//System.out.println(this);
  436. 			awake.signalAll();
  437. 		} finally {
  438. 			writeLock.unlock();
  439. 		}
  440. 	}
  441.  
  442. 	public boolean isClosed() {
  443. 		return closed;
  444. 	}
  445.  
  446. 	/**
  447. 	 * 
  448. 	 * 
  449. 	 * @param timeout
  450. 	 *            appointed limit time
  451. 	 * 
  452. 	 * @param unit
  453. 	 *            time unit
  454. 	 */
  455. 	public T poll(long timeout, TimeUnit unit) throws InterruptedException {
  456. 		long nanoTime = unit.toNanos(timeout);
  457. 		readLock.lockInterruptibly();
  458.  
  459. 		try {
  460. 			for (;;) {
  461. 				if (readCount > 0) {
  462. 					return extract();
  463. 				}
  464.  
  465. 				if (nanoTime <= 0) {
  466. 					return null;
  467. 				}
  468. 				nanoTime = queueSwitch(nanoTime, true);
  469. 			}
  470. 		} finally {
  471. 			readLock.unlock();
  472. 		}
  473. 	}
  474.  
  475. 	/**
  476. 	 * 
  477. 	 * @param ea
  478. 	 *            line buffer
  479. 	 * 
  480. 	 * 
  481. 	 * @param timeout
  482. 	 *            a appointed limit time
  483. 	 * 
  484. 	 * @param unit
  485. 	 *            a time unit
  486. 	 * 
  487. 	 * @return line number of data.if less or equal than 0, means fail.
  488. 	 * 
  489. 	 * @throws InterruptedException
  490. 	 *             if being interrupted during the try limit time.
  491. 	 */
  492. 	public int poll(T[] ea, long timeout, TimeUnit unit)
  493. 			throws InterruptedException {
  494. 		long nanoTime = unit.toNanos(timeout);
  495. 		readLock.lockInterruptibly();
  496.  
  497. 		try {
  498. 			for (;;) {
  499. 				if (readCount > 0) {
  500. 					return extract(ea);
  501. 				}
  502.  
  503. 				if (nanoTime == -2) {
  504. 					return -1;
  505. 				}
  506.  
  507. 				if (nanoTime <= 0) {
  508. 					return 0;
  509. 				}
  510. 				nanoTime = queueSwitch(nanoTime, false);
  511. 			}
  512. 		} finally {
  513. 			readLock.unlock();
  514. 		}
  515. 	}
  516.  
  517. 	public Iterator<T> iterator() {
  518. 		return null;
  519. 	}
  520.  
  521. 	/**
  522. 	 * Get size of {@link Storage} in bytes.
  523. 	 * 
  524. 	 * @return Storage size.
  525. 	 * 
  526. 	 * */
  527. 	@Override
  528. 	public int size() {
  529. 		return (writeCount + readCount);
  530. 	}
  531.  
  532. 	@Override
  533. 	public int drainTo(Collection<? super T> c) {
  534. 		return 0;
  535. 	}
  536.  
  537. 	@Override
  538. 	public int drainTo(Collection<? super T> c, int maxElements) {
  539. 		return 0;
  540. 	}
  541.  
  542. 	/**
  543. 	 * If exists write space, it will return true, and write one line to the
  544. 	 * space.<br>
  545. 	 * otherwise, it will try to do that in a appointed time(20
  546. 	 * milliseconds),when time out if still failed, return false.
  547. 	 * 
  548. 	 * @param line
  549. 	 *            a Line.
  550. 	 * 
  551. 	 * @see DoubleCachedQueue#offer(Line, long, TimeUnit)
  552. 	 * 
  553. 	 */
  554. 	@Override
  555. 	public boolean offer(T line) {
  556. 		try {
  557. 			return offer(line, 20, TimeUnit.MILLISECONDS);
  558. 		} catch (InterruptedException e1) {
  559. 			log.debug(e1.getMessage(), e1);
  560. 		}
  561. 		return false;
  562. 	}
  563.  
  564. 	@Override
  565. 	public void put(T e) throws InterruptedException {
  566. 	}
  567.  
  568. 	@Override
  569. 	public int remainingCapacity() {
  570. 		return 0;
  571. 	}
  572.  
  573. 	@Override
  574. 	public T take() throws InterruptedException {
  575. 		return null;
  576. 	}
  577.  
  578. 	@Override
  579. 	public T peek() {
  580. 		return null;
  581. 	}
  582.  
  583. 	@Override
  584. 	public T poll() {
  585. 		try {
  586. 			return poll(1*1000, TimeUnit.MILLISECONDS);
  587. 		} catch (InterruptedException e) {
  588. 			log.debug(e.getMessage(), e);
  589. 		}
  590. 		return null;
  591. 	}
  592.  
  593. }


(全文完)
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/天景科技苑/article/detail/891796?site
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号