双队列的一种实现

介绍

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

设计

接下来具体介绍双队列的设计，并且会粘贴少量方法代码，帮助介绍。

本文中讲述的双队列，本质上是两个数组保存写入的Object，一个数组负责写入，另一个被消费者读出，两个数组都对应一个重入锁。数组内写入的数据会被计数。

public class DoubleCachedQueue<T> extends AbstractQueue<T> implements
		BlockingQueue<T>, java.io.Serializable {
	private static final long serialVersionUID = 1L;
	
	private static int default_line_limit = 1000;
	
	private static long max_cache_size = 67108864L;

	private int lineLimit;
	
	private long cacheSize;

	private T[] itemsA;

	private T[] itemsB;

	private ReentrantLock readLock, writeLock;

	private Condition notFull;

	private Condition awake;

	/**
	 * writeArray : in reader's eyes, reader get data from data source and write
	 * data to this line array. readArray : in writer's eyes, writer put data to
	 * data destination from this line array.
	 * 
	 * Because of this is doubleQueue mechanism, the two line will exchange when
	 * time is suitable.
	 * 
	 */
	private T[] writeArray, readArray;

	private volatile int writeCount, readCount;

	private int writeArrayTP, readArrayHP;
	
	private volatile boolean closed = false;

	private int spillSize = 0;

	private long lineRx = 0;

	private long lineTx = 0;

队列实现了阻塞队列的接口，所以在向队列offer数据的时候是阻塞的，同样，取出操作poll也会阻塞。两个数组会在适当的时候进行queueSwitch操作。queueSwitch的条件就是当读者把queue读空了之后，且写入的queue此时不为空的时候，两个queue就会进行交换。在交换的时候，写入queue会被上锁，此时生产者不能让队列里写入数据。一般情况下，queue互换其实就是两个数组的引用互换，将相应的计数器也重置，写队列的计数器此时就清零了，因为queue交换是因为读队列已经被读空。

private long queueSwitch(long timeout, boolean isInfinite)
			throws InterruptedException {
		System.out.println("queue switch");
		writeLock.lock();
		try {
			if (writeCount <= 0) {
				if (closed) {
					return -2;
				}
				try {
					if (isInfinite && timeout <= 0) {
						awake.await();
						return -1;
					} else {
						return awake.awaitNanos(timeout);
					}
				} catch (InterruptedException ie) {
					awake.signal();
					throw ie;
				}
			} else {
				T[] tmpArray = readArray;
				readArray = writeArray;
				writeArray = tmpArray;

				readCount = writeCount;
				readArrayHP = 0;

				writeCount = 0;
				writeArrayTP = 0;

				notFull.signal();
				// logger.debug("Queue switch successfully!");
				return -1;
			}
		} finally {
			writeLock.unlock();
		}
	}

上面queue交换的时候，可以看到当要被交换的写队列也已经为空的时候，会做一次检查。如果此时queue已经被显示地关闭了，那么poll操作就会返回空，读者此时应该检查queue是否已经被closed了，若已经closed了，那么读者已经把queue里的数据读完了。这里的显示close是我们给双队列加的一个状态，close这件事的作用是为了让读者知道：生产者已经停止往queue里写新数据了，但是queue里其实可能还有未取完的数据（在写queue里，此时还差一次queue switch），你往queue poll取数据的时候，如果取到空了，那么应该做一次check，如果queue已经关闭了，那么读者就知道本次读的任务完全结束了。反过来，close状态其实不影响写，生产者如果还想写的话，其实也是可以的，但是我不推荐这么做。

public void close() {
		writeLock.lock();
		try {
			closed = true;
			//System.out.println(this);
			awake.signalAll();
		} finally {
			writeLock.unlock();
		}
	}

如果没有这个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循环重试直到写入成功。

@Override
public void run() {
	long start = System.currentTimeMillis();
	log.debug(Thread.currentThread() + " Unpacker started at " + start);
		
	Random r = new Random(start);
	Bundle bundle = null;
	boolean shoudShutdown = false;
	try {
		while(!shoudShutdown) {
			bundle = (Bundle) fromQueue.poll();
			if (bundle == null) {
				if (seeEOF.get()) {
					// 当取到空，并且其他线程已经取到EOF，那么本线程将Latch减1，并退出循环
					latch.countDown();
					shoudShutdown = true;
				} else {
					// 如果EOF还没被取到，本线程小睡一会后继续取
					try {
						sleep(r.nextInt(10));
					} catch (InterruptedException e) {
						log.error("Interrupted when taking a nap", e);
					}
				}
			} else if (!bundle.isEof()) {
			// bundle非空且非EOF，则往双队列写入一个Bundle
			byte[] lineBytes = BundleUtil.getDecompressedData(bundle);
			// 放入双队列时，若offer失败则重试
			while (!toQueue.offer(new UnCompressedBundle(bundle.getId(), ByteUtil.bytes2Lines(lineBytes, lineDelim), bundle.getIndex(), bundle.getJobId()))) {
				log.info("Unpacker put failed, will retry");
			}
			log.info("After enqueue, queue size is " + toQueue.size());
			} else {
				// Unpacker获得到了EOF
				seeEOF.set(true);
				// 自己将Lacth减1，并等待其他线程退出
				latch.countDown();
				try {
					latch.await();
				} catch (InterruptedException e) {
					log.error("Interrupted when waiting the latch ");
				}
				// 其他线程已经退出，本线程放入EOF
				while (!toQueue.offer(new UnCompressedBundle(-1L, new Line[0], -1L, -1L))) {
					log.info("Unpacker put EOF failed, will retry");
				}
				// 关闭Queue
				toQueue.close();
				// 退出循环
				shoudShutdown = true;
			}
		}
		log.debug(Thread.currentThread() + " Unpacker finished in " + (System.currentTimeMillis()-start) + " ms");
	} catch (Exception e) {
		log.error("Exception when unpacker is running ", e);
		// 将latch减1，表示自己异常退出，且不再工作
		// latch.countDown();
		log.debug(Thread.currentThread() + " Unpacker occured exception and stopped. ");
	} finally {
		
	}
}

多对一

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

总结分析

本文介绍了一种经典双队列的设计和实现，也给出了一些代码演示。文章末尾我会贴出整个双队列的代码实现，需要的同学也可以留言，我把.java发给你。如果使用的时候有发现问题，不吝赐教，这个双队列的实现也还不是很完美。使用的时候也存在需要注意的地方。

其实双队列的目的还是在于让写和读互相没有影响，而且更加照顾了写的速度。因为一般写的速度可能会比较快，而读的人读出之后还会做一些额外的处理，所以写的这一方借助双队列，可以持续写的过程，而且如果读的一方慢的话，可以多起几个消费者线程，就像"一对多"场景里阐述的那样来使用双队列。

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

import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


import lombok.ToString;
import lombok.extern.log4j.Log4j;


/**
 * Represents a region with two swap spaces, one for storing data which from
 * data source, the other one for storing data which will be transferred to data
 * destination.
 * <br>
 * A classical DoubleCachedQueue, In beginning, space A and space B both
 * empty, then loading task begin to load data to space A, when A is almost
 * full, let the data from data source being loaded to space B, then dumping
 * task begin to dump data from space A to data source. When space A is empty,
 * switch the two spaces for load and dump task. Repeat the above operation.
 * 
 */
@Log4j
@ToString
public class DoubleCachedQueue<T> extends AbstractQueue<T> implements
		BlockingQueue<T>, java.io.Serializable {
	private static final long serialVersionUID = 1L;
	
	private static int default_line_limit = 1000;
	
	private static long max_cache_size = 67108864L;

	private int lineLimit;
	
	private long cacheSize;

	private T[] itemsA;

	private T[] itemsB;

	private ReentrantLock readLock, writeLock;

	private Condition notFull;

	private Condition awake;

	/**
	 * writeArray : in reader's eyes, reader get data from data source and write
	 * data to this line array. readArray : in writer's eyes, writer put data to
	 * data destination from this line array.
	 * 
	 * Because of this is doubleQueue mechanism, the two line will exchange when
	 * time is suitable.
	 * 
	 */
	private T[] writeArray, readArray;

	private volatile int writeCount, readCount;

	private int writeArrayTP, readArrayHP;
	
	private volatile boolean closed = false;

	private int spillSize = 0;

	private long lineRx = 0;

	private long lineTx = 0;

	/**
	 * Get info of line number in {@link DoubleCachedQueue} space.
	 * 
	 * @return Information of line number.
	 * 
	 */
	public String info() {
		return String.format("Write Array: %s/%s; Read Array: %s/%s", writeCount, writeArray.length, readCount, readArray.length);
	}

	/**
	 * Use the two parameters to construct a {@link DoubleCachedQueue} which hold the
	 * swap areas.
	 * 
	 * @param lineLimit
	 *            Limit of the line number the {@link DoubleCachedQueue} can hold.
	 * 
	 * @param byteLimit
	 *            Limit of the bytes the {@link DoubleCachedQueue} can hold.
	 * 
	 */
	public DoubleCachedQueue(int lineLimit) {
		if (lineLimit <= 0) {
			this.lineLimit = default_line_limit;
		}else{
			this.lineLimit = lineLimit;
		}
		itemsA = (T[])new Object[lineLimit];
		itemsB = (T[])new Object[lineLimit];

		readLock = new ReentrantLock();
		writeLock = new ReentrantLock();

		notFull = writeLock.newCondition();
		awake = writeLock.newCondition();

		readArray = itemsA;
		writeArray = itemsB;
		spillSize = lineLimit * 8 / 10;
	}
	
	public DoubleCachedQueue(long cacheSize){
		if (cacheSize <= 0) {
			throw new IllegalArgumentException(
					"Queue initial capacity can't less than 0!");
		}
		this.cacheSize = cacheSize > max_cache_size ? max_cache_size : cacheSize;
		
		readLock = new ReentrantLock();
		writeLock = new ReentrantLock();

		notFull = writeLock.newCondition();
		awake = writeLock.newCondition();

		readArray = itemsA;
		writeArray = itemsB;
		spillSize = lineLimit * 8 / 10;
	}

	/**
	 * Get line number of the {@link DoubleCachedQueue}
	 * 
	 * @return lineLimit Limit of the line number the {@link DoubleCachedQueue} can
	 *         hold.
	 * 
	 */
	public int getLineLimit() {
		return lineLimit;
	}

	/**
	 * Set line number of the {@link DoubleCachedQueue}.
	 * 
	 * @param capacity
	 *            Limit of the line number the {@link DoubleCachedQueue} can hold.
	 * 
	 */
	public void setLineLimit(int capacity) {
		this.lineLimit = capacity;
	}

	/**
	 * Insert one line of record to a apace which buffers the swap data.
	 * 
	 * @param line
	 *            The inserted line.
	 * 
	 */
	private void insert(T line) {
		writeArray[writeArrayTP] = line;
		++writeArrayTP;
		++writeCount;
		++lineRx;
	}

	/**
	 * Insert a line array(appointed the limit of array size) of data to a apace
	 * which buffers the swap data.
	 * 
	 * @param lines
	 *            Inserted line array.
	 * 
	 * @param size
	 *            Limit of inserted size of the line array.
	 * 
	 */
	private void insert(T[] lines, int size) {
		if(size > 0){
			System.arraycopy(lines, 0, writeArray, writeArrayTP, size);
			writeArrayTP = writeArrayTP + size;
			writeCount = writeCount + size;
			lineRx = lineRx + size;
		}
//		for (int i = 0; i < size; ++i) {
//			writeArray[writeArrayTP] = lines[i];
//			++writeArrayTP;
//			++writeCount;
//			++lineRx;
//			if(lines[i] != null && lines[i].getLine() != null){
//				byteRx += lines[i].getLine().length();
//			}
//		}
	}

	/**
	 * Extract one line of record from the space which contains current data.
	 * 
	 * @return line A line of data.
	 * 
	 */
	private T extract() {
		T e = readArray[readArrayHP];
		readArray[readArrayHP] = null;
		++readArrayHP;
		--readCount;
		++lineTx;
		return e;
	}

	/**
	 * Extract a line array of data from the space which contains current data.
	 * 
	 * @param ea
	 * @return Extracted line number of data.
	 * 
	 */
	private int extract(T[] ea) {
		int readsize = Math.min(ea.length, readCount);
		if(readsize > 0){
			readCount = readCount - readsize;
			lineTx = lineTx + readsize;
			System.arraycopy(readArray, readArrayHP, ea, 0, readsize);
			readArrayHP = readArrayHP + readsize;
		}
//		for (int i = 0; i < readsize; ++i) {
//			ea[i] = readArray[readArrayHP];
//			readArray[readArrayHP] = null;
//			++readArrayHP;
//			--readCount;
//			++lineTx;
//		}
		return readsize;
	}

	/**
	 * switch condition: read queue is empty && write queue is not empty.
	 * Notice:This function can only be invoked after readLock is grabbed,or may
	 * cause dead lock.
	 * 
	 * @param timeout
	 * 
	 * @param isInfinite
	 *            whether need to wait forever until some other thread awake it.
	 * 
	 * @return
	 * 
	 * @throws InterruptedException
	 * 
	 */

	private long queueSwitch(long timeout, boolean isInfinite)
			throws InterruptedException {
		System.out.println("queue switch");
		writeLock.lock();
		try {
			if (writeCount <= 0) {
				if (closed) {
					return -2;
				}
				try {
					if (isInfinite && timeout <= 0) {
						awake.await();
						return -1;
					} else {
						return awake.awaitNanos(timeout);
					}
				} catch (InterruptedException ie) {
					awake.signal();
					throw ie;
				}
			} else {
				T[] tmpArray = readArray;
				readArray = writeArray;
				writeArray = tmpArray;

				readCount = writeCount;
				readArrayHP = 0;

				writeCount = 0;
				writeArrayTP = 0;

				notFull.signal();
				// logger.debug("Queue switch successfully!");
				return -1;
			}
		} finally {
			writeLock.unlock();
		}
	}

	/**
	 * If exists write space, it will return true, and write one line to the
	 * space. otherwise, it will try to do that in a appointed time,when time is
	 * out if still failed, return false.
	 * 
	 * @param line
	 *            a Line.
	 * 
	 * @param timeout
	 *            appointed limit time
	 * 
	 * @param unit
	 *            time unit
	 * 
	 * @return True if success,False if failed.
	 * 
	 */
	public boolean offer(T line, long timeout, TimeUnit unit)
			throws InterruptedException {
		if (line == null) {
			throw new NullPointerException();
		}
		long nanoTime = unit.toNanos(timeout);
		writeLock.lockInterruptibly();
		if(itemsA == null || itemsB == null){
			initArray(line);
		}
		try {
			for (;;) {
				if (writeCount < writeArray.length) {
					insert(line);
					if (writeCount == 1) {
						awake.signal();
					}
					return true;
				}

				// Time out
				if (nanoTime <= 0) {
					return false;
				}
				// keep waiting
				try {
					nanoTime = notFull.awaitNanos(nanoTime);
				} catch (InterruptedException ie) {
					notFull.signal();
					throw ie;
				}
			}
		} finally {
			writeLock.unlock();
		}
	}

	private void initArray(T line) {
		
		long recordLength = computeSize(line);
		long size = cacheSize/recordLength;
		if(size <= 0){
			size = default_line_limit;
		}
		lineLimit = (int) size;
		itemsA = (T[])new Object[(int) size];
		itemsB = (T[])new Object[(int) size];
		readArray = itemsA;
		writeArray = itemsB;
		
	}

	public long computeSize(T line){
		return 1;
	}

	/**
	 * If exists write space, it will return true, and write a line array to the
	 * space.<br>
	 * otherwise, it will try to do that in a appointed time,when time out if
	 * still failed, return false.
	 * 
	 * @param lines
	 *            line array contains lines of data
	 * 
	 * @param size
	 *            Line number needs to write to the space.
	 * 
	 * @param timeout
	 *            appointed limit time
	 * 
	 * @param unit
	 *            time unit
	 * 
	 * @return status of this operation, true or false.
	 * 
	 * @throws InterruptedException
	 *             if being interrupted during the try limit time.
	 * 
	 */
	public boolean offer(T[] lines, int size, long timeout, TimeUnit unit)
			throws InterruptedException {
		if (lines == null || lines.length == 0) {
			throw new NullPointerException();
		}
		long nanoTime = unit.toNanos(timeout);
		writeLock.lockInterruptibly();
		if(itemsA == null || itemsB == null){
			initArray(lines[0]);
		}
		try {
			for (;;) {
				if (writeCount + size <= writeArray.length) {
					insert(lines, size);
					if (writeCount >= spillSize) {
						awake.signalAll();
					}
					return true;
				}

				// Time out
				if (nanoTime <= 0) {
					return false;
				}
				// keep waiting
				try {
					nanoTime = notFull.awaitNanos(nanoTime);
				} catch (InterruptedException ie) {
					notFull.signal();
					throw ie;
				}
			}
		} finally {
			writeLock.unlock();
		}
	}

	/**
	 * Close the synchronized lock and one inner state.
	 * 
	 */
	public void close() {
		writeLock.lock();
		try {
			closed = true;
			//System.out.println(this);
			awake.signalAll();
		} finally {
			writeLock.unlock();
		}
	}

	public boolean isClosed() {
		return closed;
	}

	/**
	 * 
	 * 
	 * @param timeout
	 *            appointed limit time
	 * 
	 * @param unit
	 *            time unit
	 */
	public T poll(long timeout, TimeUnit unit) throws InterruptedException {
		long nanoTime = unit.toNanos(timeout);
		readLock.lockInterruptibly();

		try {
			for (;;) {
				if (readCount > 0) {
					return extract();
				}

				if (nanoTime <= 0) {
					return null;
				}
				nanoTime = queueSwitch(nanoTime, true);
			}
		} finally {
			readLock.unlock();
		}
	}

	/**
	 * 
	 * @param ea
	 *            line buffer
	 * 
	 * 
	 * @param timeout
	 *            a appointed limit time
	 * 
	 * @param unit
	 *            a time unit
	 * 
	 * @return line number of data.if less or equal than 0, means fail.
	 * 
	 * @throws InterruptedException
	 *             if being interrupted during the try limit time.
	 */
	public int poll(T[] ea, long timeout, TimeUnit unit)
			throws InterruptedException {
		long nanoTime = unit.toNanos(timeout);
		readLock.lockInterruptibly();

		try {
			for (;;) {
				if (readCount > 0) {
					return extract(ea);
				}

				if (nanoTime == -2) {
					return -1;
				}

				if (nanoTime <= 0) {
					return 0;
				}
				nanoTime = queueSwitch(nanoTime, false);
			}
		} finally {
			readLock.unlock();
		}
	}

	public Iterator<T> iterator() {
		return null;
	}

	/**
	 * Get size of {@link Storage} in bytes.
	 * 
	 * @return Storage size.
	 * 
	 * */
	@Override
	public int size() {
		return (writeCount + readCount);
	}

	@Override
	public int drainTo(Collection<? super T> c) {
		return 0;
	}

	@Override
	public int drainTo(Collection<? super T> c, int maxElements) {
		return 0;
	}

	/**
	 * If exists write space, it will return true, and write one line to the
	 * space.<br>
	 * otherwise, it will try to do that in a appointed time(20
	 * milliseconds),when time out if still failed, return false.
	 * 
	 * @param line
	 *            a Line.
	 * 
	 * @see DoubleCachedQueue#offer(Line, long, TimeUnit)
	 * 
	 */
	@Override
	public boolean offer(T line) {
		try {
			return offer(line, 20, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e1) {
			log.debug(e1.getMessage(), e1);
		}
		return false;
	}

	@Override
	public void put(T e) throws InterruptedException {
	}

	@Override
	public int remainingCapacity() {
		return 0;
	}

	@Override
	public T take() throws InterruptedException {
		return null;
	}

	@Override
	public T peek() {
		return null;
	}

	@Override
	public T poll() {
		try {
			return poll(1*1000, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e) {
			log.debug(e.getMessage(), e);
		}
		return null;
	}

}

(全文完)