JDK1.5 Queue

JDK1.5 Queue

LinkedList：

LinkedList不是同步的。如果多个线程同时访问列表，而其中至少一个线程从结构上修改了该列表，则它必须 保持外部同步。（结构修改指添加或删除一个或多个元素的任何操作；仅设置元素的值不是结构修改。）这一般通过对自然封装该列表的对象进行同步操作来完成。如果不存在这样的对象，则应该使用 Collections.synchronizedList 方法来“包装”该列表。最好在创建时完成这一操作，以防止对列表进行意外的不同步访问，如下所示：

List list = Collections.synchronizedList(new LinkedList(...));

将指定元素追加到列表的结尾有add、offer两个方法，区别是offer是接口 Queue<E>中的，add是接口 Collection<E> 中的。offer使用可能有插入限制（例如容量限定）的队列时，offer方法通常要优于 Collection.add(E) 方法，因为add只能通过抛出异常使插入元素失败。而offer方法的返回值是：如果可以向此队列添加元素，则返回 true；否则返回 false。

找到并移除列表的头（第一个元素）有poll、remove两个方法，区别是poll是接口 Queue<E>中的，remove是接口 Queue<E>中的。如果队列为空，poll返回null，而remove抛出NoSuchElementException。

找到但不移除列表的头（第一个元素）是peek方法

 

实例：

package com.bijian.thread;

import java.util.LinkedList;
import java.util.Queue;

public class UseLinkedList {

	public static void main(String[] args) {
		Queue<String> queue = new LinkedList<String>();
		queue.offer("Hello");
		queue.offer("World!");
		queue.offer("你好！");
		System.out.println(queue.size());
		String str;
		while ((str = queue.poll()) != null) {
			System.out.println(str);
		}
		System.out.println();
		System.out.println(queue.size());
	}
}

 

运行结果：

3
Hello
World!
你好！

0

 

ArrayBlockingQueue：

       一个由数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素。新元素插入到队列的尾部，队列检索操作则是从队列头部开始获得元素。

ArrayBlockingQueue的常用方法：

       put：将指定的元素添加到此队列的尾部，如果必要，将等待可用的空间。

       take：检索并移除此队列的头部，如果此队列不存在任何元素，则一直等待。

       offer：将指定的元素插入到此队列的尾部（如果可能），如果此队列已满，则立即返回。

       poll：检索并移除此队列的头，如果此队列为空，则返回null。

说明：put、offer都是向队列中添加元素，take、poll都是从队列中移除元素。put和take对应一组，是阻塞式的；offer、poll对应一组，是非阻塞式的。

 

实例：

 

package com.bijian.thread.blockqueue;

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

	private final BlockingQueue<Character> queue;

	Producer(BlockingQueue<Character> q) {
		queue = q;
	}

	public void run() {
		try {
			for (int i = 0; i < 10; i++) {
				queue.put(produce());
				System.out.println("生产后：" + queue.peek());
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	Character produce() {
		char c = (char) (Math.random() * 26 + 'A');
		System.out.println("生产前：" + c);
		return c;
	}
}

 

package com.bijian.thread.blockqueue;

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {
	
	private String name;
	private final BlockingQueue<Character> queue;
	
	public Consumer(String name, BlockingQueue<Character> q) {
		this.name = name;
		this.queue = q;
	}

	@Override
	public void run() {
		
		try {
			while(true) {
				consume(queue.take());
			}
		} catch(InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public void consume(Character c) {
		System.out.println("[" + name + "]消费: " + c);
	}
}

 

package com.bijian.thread.blockqueue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class Main {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		
		BlockingQueue q = new ArrayBlockingQueue(10);
		
		Producer p = new Producer(q);
		Consumer c1 = new Consumer("tom", q);
		Consumer c2 = new Consumer("jacky", q);
		new Thread(c1).start();
		new Thread(c2).start();
		new Thread(p).start();
	}
}

 运行结果（打印结束后程序一直未结束，但也未输出，处于阻塞状态）：

生产前：L
生产后：L
生产前：Q
[tom]消费: L
[jacky]消费: Q
生产后：Q
生产前：G
生产后：G
[tom]消费: G
生产前：O
生产后：O
[jacky]消费: O
生产前：M
生产后：M
[tom]消费: M
生产前：I
生产后：I
[jacky]消费: I
生产前：B
生产后：B
生产前：Q
[tom]消费: B
[jacky]消费: Q
生产后：Q
生产前：D
生产后：D
[tom]消费: D
生产前：A
生产后：A
[jacky]消费: A

 

 

实例2：

package com.bijian.thread.blockqueue;

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

	private final BlockingQueue<Character> queue;

	Producer(BlockingQueue<Character> q) {
		queue = q;
	}

	public void run() {
		try {
			for (int i = 0; i < 10; i++) {
				queue.offer(produce());
				//System.out.println("生产后：" + queue.peek());
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	Character produce() {
		char c = (char) (Math.random() * 26 + 'A');
		System.out.println("生产前：" + c);
		return c;
	}
}

 

package com.bijian.thread.blockqueue;

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {
	
	private String name;
	private final BlockingQueue<Character> queue;
	
	public Consumer(String name, BlockingQueue<Character> q) {
		this.name = name;
		this.queue = q;
	}

	@Override
	public void run() {
		
		while(true) {
			consume(queue.poll());
			try {
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
	public void consume(Character c) {
		System.out.println("[" + name + "]消费: " + c);
	}
}

 

package com.bijian.thread.blockqueue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class Main {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		
		BlockingQueue q = new ArrayBlockingQueue(10);
		
		Producer p = new Producer(q);
		Consumer c1 = new Consumer("tom", q);
		Consumer c2 = new Consumer("jacky", q);
		
		new Thread(p).start();
		new Thread(c1).start();
		new Thread(c2).start();
	}
}

  

运行结果（打印结束后程序一直未结束，消费者一直输出null，处于非阻塞状态）：

[tom]消费: null
[jacky]消费: null
生产前：S
生产前：R
生产前：N
生产前：S
生产前：P
生产前：U
生产前：J
生产前：Y
生产前：C
生产前：H
[jacky]消费: S
[tom]消费: R
[jacky]消费: N
[tom]消费: S
[tom]消费: P
[jacky]消费: U
[jacky]消费: J
[tom]消费: Y
[jacky]消费: C
[tom]消费: H
[jacky]消费: null
[tom]消费: null
[tom]消费: null
[jacky]消费: null
……

 

 

DelayQueue：

       Delayed 元素的一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的 Delayed 元素。如果延迟都还没有期满，则队列没有头部，并且 poll 将返回 null。当一个元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于或等于零的值时，则出现期满。此队列不允许使用 null 元素。

DelayQueue的常用方法：

       add：将指定元素添加到此队列中。

       offer：将指定元素插入到此延迟队列中。

       put：将指定元素添加到此延迟队列中。由于该队列是无界的，所以此方法永远不会阻塞。

       take：检索并移除此队列的头部，如果此队列不存在未到期延迟的元素，则等待它。

       poll：检索并移除此队列的头部，如果此队列不存在未到期延迟的元素，则返回null。

       remove：从此队列中移除指定元素的单个实例（如果存在）。

       drainTo：移除此队列中所有可用的元素，并将它们添加到给定 collection 中。

 

实例：

package com.bijian.thread.blockqueue.delayqueue;

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class NanoDelay implements Delayed {
	
	private String name;
	private long targetTime;
	
	public NanoDelay(String name, long targetTime) {
		super();
		this.name = name;
		this.targetTime = TimeUnit.NANOSECONDS.convert(targetTime, TimeUnit.NANOSECONDS);
	}

	@Override
	public int compareTo(Delayed o) {
		NanoDelay that = (NanoDelay)o;
		
		return targetTime > that.targetTime?1:(targetTime < that.targetTime?-1:0);
	}

	@Override
	public long getDelay(TimeUnit unit) {
		
		return unit.convert(targetTime, unit.NANOSECONDS);
	}

	@Override
	public String toString() {
		StringBuilder builder = new StringBuilder();
		builder.append("NanoDelay [name=");
		builder.append(name);
		builder.append(", targetTime=");
		builder.append(targetTime);
		builder.append("]");
		return builder.toString();
	}
}

 

package com.bijian.thread.blockqueue.delayqueue;

import java.util.Random;
import java.util.concurrent.DelayQueue;

public class Main {
	
	public static void main(String args[]) {
		
		Random random = new Random();
		DelayQueue<NanoDelay> queue = new DelayQueue<NanoDelay>();
		
		for (int i = 0; i < 5; i++) {
			queue.add(new NanoDelay("T" + i, random.nextInt(5)));
		}
		
		System.out.println("start..");
		System.out.println(queue.size());
		
		for(int i = 0; i < 5; i++) {
			try {
				NanoDelay delay = queue.take();
				System.out.println(delay);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}

 

运行结果（打印结束后程序一直未结束，但也未输出，处于阻塞状态）：

start..
5
NanoDelay [name=T1, targetTime=0]