队列和优先队列是我们十分熟悉的数据结构。提供了所谓的“先进先出”功能，优先队列则按照某种规则“先进先出”。但是他们都没有提供：“固定大小的队列”和“固定大小的优先队列”的功能。
比如我们要实现：记录按照时间排序的最近的登录网站的20个人；按照分数排序的最高的30个人；比如在游戏中一场两两PK的战斗，得分最高的6个人；要实现这些功能时，需要的数据结构，在java类库中没有现成的类。我们需要利用现有的类库来实现它们。
1. 固定大小的“先进先出”队列
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue; public class TopQueue<E> {
private final LinkedBlockingQueue<E> blockQueue; public TopQueue(int size){
this.blockQueue = new LinkedBlockingQueue<E>(size);
} public synchronized void put(E e) throws InterruptedException{
if(blockQueue.offer(e)){
return;
}else{
blockQueue.take();
blockQueue.offer(e);
}
} public List<E> getAll(){
return new ArrayList<E>(blockQueue);
} public static void main(String[] args) throws InterruptedException{
TopQueue<Integer> tq = new TopQueue<Integer>(3);
tq.put(1);
tq.put(2);
tq.put(3);
System.out.println(Arrays.toString(tq.getAll().toArray())); tq.put(4);
System.out.println(Arrays.toString(tq.getAll().toArray())); tq.put(5);
System.out.println(Arrays.toString(tq.getAll().toArray())); tq.put(6);
System.out.println(Arrays.toString(tq.getAll().toArray()));
}
}
输出的结果为：
[1, 2, 3]
[2, 3, 4]
[3, 4, 5]
[4, 5, 6]
上面的TopQueue实现了“固定大小的线程安全的”队列。无论有多少个线程，向TopQueue中放入了多少个元素，在TopQueue中只保留最后放进去的n个元素。
2. 固定大小的优先队列（实现一）
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.PriorityBlockingQueue;
import com.alibaba.fastjson.JSON; public class TopPriorityQueue<E> {
private final PriorityBlockingQueue<E> blockQueue;
private final int size; public TopPriorityQueue(int size){
this.blockQueue = new PriorityBlockingQueue<E>(size + 1);
this.size = size + 1; // 这里多加1的原因是防止put方法中将大的删除了，反而降小的插入了，所以多加1个用做"哨卡"
} public synchronized void put(E e) throws InterruptedException{
if(blockQueue.size() >= size)
blockQueue.take();
blockQueue.offer(e);
} public List<E> getAll() throws InterruptedException{
synchronized(this){
if(blockQueue.size() >= size)
blockQueue.take(); // 前面构造函数中多加了1，这里减掉一个
}
return new ArrayList<E>(blockQueue);
} public static void main(String[] args) throws InterruptedException{
final TopPriorityQueue<User> tq = new TopPriorityQueue<User>(3);
User u1 = new User(1, "bbb", 10);
User u2 = new User(2, "ccc", 20);
User u3 = new User(3, "ddd", 30);
User u4 = new User(4, "fff", 40);
User u5 = new User(5, "fff", 50);
User u6 = new User(6, "ddd", 60);
User u7 = new User(7, "ggg", 70);
User u8 = new User(8, "hhh", 80); tq.put(u4); //
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u8); //4,8
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u7); //4,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u5); //5,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u2); //5,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u3); //5,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u1); //5,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u6); //6,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
}
}
User类：
import java.util.Comparator;
public class User implements Comparable<User>{
private int id;
private String name;
private long score; // 得分
// ... ...
public User(int id, String name, long score){
this.id = id;
this.name = name;
this.score = score;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public long getScore() {
return score;
}
public void setScore(long score) {
this.score = score;
}
@Override
public int compareTo(User o) {
return this.getScore() > o.getScore() ? 1 : this.getScore() < o.getScore() ? -1 : 0;
}
}
输入的结果为：
[{"id":4,"name":"fff","score":40}]
[{"id":4,"name":"fff","score":40},{"id":8,"name":"hhh","score":80}]
[{"id":4,"name":"fff","score":40},{"id":8,"name":"hhh","score":80},{"id":7,"name":"ggg","score":70}]
[{"id":5,"name":"fff","score":50},{"id":8,"name":"hhh","score":80},{"id":7,"name":"ggg","score":70}]
[{"id":5,"name":"fff","score":50},{"id":8,"name":"hhh","score":80},{"id":7,"name":"ggg","score":70}]
[{"id":5,"name":"fff","score":50},{"id":8,"name":"hhh","score":80},{"id":7,"name":"ggg","score":70}]
[{"id":5,"name":"fff","score":50},{"id":8,"name":"hhh","score":80},{"id":7,"name":"ggg","score":70}]
[{"id":6,"name":"ddd","score":60},{"id":8,"name":"hhh","score":80},{"id":7,"name":"ggg","score":70}]
TopPriorityQueue实现了“固定大小的优先队列”，的实现原理是：
public synchronized void put(E e) throws InterruptedException{
if(blockQueue.size() >= size)
blockQueue.take();
blockQueue.offer(e);
}
当队列满了，还要插入时，就删除队列中最小的一个，然后再插入。但是这里涉及到一个问题，如果这个要被插入的元素优先级要比那个被删除的元素优先级低呢？那岂不是将大的删除了，反而将小的插入了。所以这里我们采取的办法是，比实际要求的size的基础上多保留一个，用做“哨卡”。当队列满了时，我们将“哨卡”删掉，然后再插入我们的元素，然后队列中新的最小的元素就成为了新的“哨卡”。而“哨卡”因为是最小的一个，不是我们需要的，返回最终结果时会被删除掉。所以不会出现删除了大的，插入了小的问题。这里有点小技巧。
3. 固定大小的优先队列（实现二）
上面的实现，需要我们插入队列的元素Comparable这个接口，但是实际环境中，我们不太可能去进行这样的修改，所以我们还有另外一种方法——使用Comparator来搞定，看代码：
import java.util.Comparator;
import com.coin.User; public class MyComparator implements Comparator<User> {
@Override
public int compare(User u1, User u2) {
if(u1.getScore() > u2.getScore())
return 1;
if(u1.getScore() < u2.getScore())
return -1;
return 0;
}
}
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.PriorityBlockingQueue; import com.alibaba.fastjson.JSON; public class TopPriorityQueue<E> {
private final PriorityBlockingQueue<E> blockQueue;
private final int size; public TopPriorityQueue(int size, Comparator<E> comparator){
this.blockQueue = new PriorityBlockingQueue<E>(size + 1, comparator);
this.size = size + 1; // 这里多加1的原因是防止put方法中将大的删除了，反而降小的插入了，所以多加1个用做"哨卡"
} public synchronized void put(E e) throws InterruptedException{
if(blockQueue.size() >= size)
blockQueue.take();
blockQueue.offer(e);
} public List<E> getAll() throws InterruptedException{
synchronized(this){
if(blockQueue.size() >= size)
blockQueue.take(); // 前面构造函数中多加了1，这里减掉一个
} return new ArrayList<E>(blockQueue);
} public static void main(String[] args) throws InterruptedException{
MyComparator myComparator = new MyComparator();
final TopPriorityQueue<User> tq = new TopPriorityQueue<User>(3, myComparator);
User u1 = new User(1, "bbb", 10);
User u2 = new User(2, "ccc", 20);
User u3 = new User(3, "ddd", 30);
User u4 = new User(4, "fff", 40);
User u5 = new User(5, "fff", 50);
User u6 = new User(6, "ddd", 60);
User u7 = new User(7, "ggg", 70);
User u8 = new User(8, "hhh", 80); tq.put(u4); //
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u8); //4,8
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u7); //4,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u5); //5,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u2); //5,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u3); //5,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u1); //5,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
tq.put(u6); //6,8,7
System.out.println(JSON.toJSONString(tq.getAll()));
}
}
所以我们在使用PriorityBlockingQueue时，要么我们插入的元素实现了Comparable这个接口，要么我定义一个Comparator，传入到PriorityBlockingQueue的构造函数中，我们可以看下PriorityBlockingQueue.offer(e)方法的源码，它会对这两种情况进行判断：
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
int n, cap;
Object[] array;
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
其中的代码：
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
就是判断我们是否在PriorityBlockingQueue的构造函数中是否传入了Comparator。这样User类就不需要实现Comparable接口了。
另外我们要注意 LinkedBlockingQueue 和 PriorityBlockingQueue 有一点不同，BlockingQueue.offer(e)在队列满了时，会返回false，而PriorityBlockingQueue.offer()即使队列满了，它会进行扩展，永远只返回true.
LinkedBlockingQueue .offer() 的源码如下：
/**
* Inserts the specified element at the tail of this queue if it is
* possible to do so immediately without exceeding the queue's capacity,
* returning {@code true} upon success and {@code false} if this queue
* is full.
* When using a capacity-restricted queue, this method is generally
* preferable to method {@link BlockingQueue#add add}, which can fail to
* insert an element only by throwing an exception.
*
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
if (count.get() < capacity) {
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return c >= 0;
}
当满了时返回false:
if (count.get() == capacity)
return false;
PriorityBlockingQueue.offer() 的源码如下：
/**
* Inserts the specified element into this priority queue.
* As the queue is unbounded, this method will never return {@code false}.
*
* @param e the element to add
* @return {@code true} (as specified by {@link Queue#offer})
* @throws ClassCastException if the specified element cannot be compared
* with elements currently in the priority queue according to the
* priority queue's ordering
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
int n, cap;
Object[] array;
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
当满了时，会扩容：
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
As the queue is unbounded, this method will never return {@code false}.
另外TopQueue 和 TopPriorityQueue 都是线程安全的，但是并不保证插入队列中的元素自身的线程安全性。