bytefifo.java

java多线程编程实例_Source

public class ByteFIFO extends Object {
	private byte[] queue;
	private int capacity;
	private int size;
	private int head;
	private int tail;

	public ByteFIFO(int cap) {
		capacity = ( cap > 0 ) ? cap : 1; // at least 1
		queue = new byte[capacity];
		head = 0;
		tail = 0;
		size = 0;
	}

	public int getCapacity() {
		return capacity;
	}

	public synchronized int getSize() {
		return size;
	}

	public synchronized boolean isEmpty() {
		return ( size == 0 );
	}

	public synchronized boolean isFull() {
		return ( size == capacity );
	}

	public synchronized void add(byte b) 
			throws InterruptedException {

		waitWhileFull();

		queue[head] = b;
		head = ( head + 1 ) % capacity;
		size++;

		notifyAll(); // let any waiting threads know about change
	}

	public synchronized void add(byte[] list) 
			throws InterruptedException {

		// For efficiency, the bytes are copied in blocks
		// instead of one at a time. As space becomes available,
		// more bytes are copied until all of them have been
		// added.

		int ptr = 0;

		while ( ptr < list.length ) {
			// If full, the lock will be released to allow 
			// another thread to come in and remove bytes.
			waitWhileFull();

			int space = capacity - size;
			int distToEnd = capacity - head;
			int blockLen = Math.min(space, distToEnd);

			int bytesRemaining = list.length - ptr;
			int copyLen = Math.min(blockLen, bytesRemaining);

			System.arraycopy(list, ptr, queue, head, copyLen);
			head = ( head + copyLen ) % capacity;
			size += copyLen;
			ptr += copyLen;

			// Keep the lock, but let any waiting threads 
			// know that something has changed.
			notifyAll();
		}
	}

	public synchronized byte remove() 
			throws InterruptedException {

		waitWhileEmpty();
		
		byte b = queue[tail];
		tail = ( tail + 1 ) % capacity;
		size--;

		notifyAll(); // let any waiting threads know about change

		return b;
	}

	public synchronized byte[] removeAll() {
		// For efficiency, the bytes are copied in blocks
		// instead of one at a time. 

		if ( isEmpty() ) {
			// Nothing to remove, return a zero-length
			// array and do not bother with notification
			// since nothing was removed.
			return new byte[0]; 
		}

		// based on the current size
		byte[] list = new byte[size]; 

		// copy in the block from tail to the end
		int distToEnd = capacity - tail;
		int copyLen = Math.min(size, distToEnd);
		System.arraycopy(queue, tail, list, 0, copyLen);

		// If data wraps around, copy the remaining data
		// from the front of the array.
		if ( size > copyLen ) {
			System.arraycopy(
					queue, 0, list, copyLen, size - copyLen);
		}

		tail = ( tail + size ) % capacity;
		size = 0; // everything has been removed

		// Signal any and all waiting threads that 
		// something has changed.
		notifyAll(); 

		return list; 
	}

	public synchronized byte[] removeAtLeastOne() 
			throws InterruptedException {

		waitWhileEmpty(); // wait for a least one to be in FIFO
		return removeAll();
	}

	public synchronized boolean waitUntilEmpty(long msTimeout) 
			throws InterruptedException {

		if ( msTimeout == 0L ) {
			waitUntilEmpty();  // use other method
			return true;
		}

		// wait only for the specified amount of time
		long endTime = System.currentTimeMillis() + msTimeout;
		long msRemaining = msTimeout;

		while ( !isEmpty() && ( msRemaining > 0L ) ) {
			wait(msRemaining);
			msRemaining = endTime - System.currentTimeMillis();
		}

		// May have timed out, or may have met condition, 
		// calc return value.
		return isEmpty();
	}

	public synchronized void waitUntilEmpty() 
			throws InterruptedException {

		while ( !isEmpty() ) {
			wait();
		}
	}

	public synchronized void waitWhileEmpty() 
			throws InterruptedException {

		while ( isEmpty() ) {
			wait();
		}
	}

	public synchronized void waitUntilFull() 
			throws InterruptedException {

		while ( !isFull() ) {
			wait();
		}
	}

	public synchronized void waitWhileFull() 
			throws InterruptedException {

		while ( isFull() ) {
			wait();
		}
	}
}