Activemq API使用(整合spring)

BlockingQueue，是一个接口，支持泛型。BlockingQueue接口继承了Queue接口，Queue是队列的意思。Queue接口又继承了Collection接口。阻塞队列是一个额外支持两个操作的队列：在队列为空时，从队列中取元素的线程会阻塞。以及在队列满时，往队列里添加元素的线程会阻塞。这两个方法分别是put(E e)和take()。

BlockingQueue接口方法：

方法	抛异常	返回特定值	阻塞	超时后返回
添加元素	add(E e)	offer(E e)	put(E e)	offer(E e, log timeout, TimeUnit unit)
取出队首元素	remove()	poll()	take()	poll(long timeout, TimeUnit unit)
查询队首元素	element()	peek()	-	-

抛异常，指的是在队列满时添加元素会抛IllegalStateException异常，或者在队列空时取元素会抛NoSuchElementException异常。其实感觉抛异常的方法不太友好，像offer(E e)一样，队列满时添加元素返回false，像poll()一样，队列空时取元素返回null，这多好。

在jdk中，BlockingQueue有7个实现类：

1、ArrayBlockingQueue：一个由数组实现的有界阻塞队列。ArrayBlockingQueue有3个构造器，ArrayBlockingQueue(int capacity)、ArrayBlockingQueue(int capacity, boolean fair)、ArrayBlockingQueue(int capacity, boolean fair, Collection c)。ArrayBlockingQueue有个Object[] items成员变量，数组的长度等于capacity。还有个ReentrantLock lock成员变量，是否公平由fair决定，默认非公平锁。第三个构造器额外有个Collection实例参数，表示可以构造一个有一些初始数据的实例，当Collection实例的长度大于capacity时，会抛出IllegalArgumentException异常。ArrayBlockingQueue是线程安全的，所有操作元素的方法都由同一个ReentrantLock实例加持。

2、LinkedBlockingQueue：一个由链表实现的有界或者无界阻塞队列。在Executors.newFixedThreadPool()和Executors.newSingleThreadExecutor()中用到。LinkedBlockingQueue有3个构造器：LinkedBlockQueue()、LinkedBlockQueue(int capacity)、LinkedBlockQueue(Collection c)。LinkedBlockQueue(int capacity)是有界队列，最多只能放这么多个元素。LinkedBlockQueue()和LinkedBlockQueue(Collection c)内部实现都是LinkedBlockQueue(Integer.MAX_VALUE)，队列中可以一直放元素，直到Integer.MAX_VALUE，这就是无界队列了。LinkedBlockQueue内部有2个可重入锁，takeLock和putLock，且都是非公平锁，没得选。LinkedBlockingQueue也是线程安全的，所有的添加元素方法都由一个非公平的ReentrantLock实例putLock加持，所有的取出元素方法、查看元素方法都由一个非公平的ReentrantLock实例takeLock加持。线程池中假如用无界阻塞队列，则核心线程之外的线程不会启动，且系统有OOM的风险。

3、PriorityBlockingQueue：一个支持优先级的无界阻塞队列。PriorityBlockingQueue始终保证出队的元素是优先级最高的元素，内部使用二叉树最小堆算法维护一个Object数组，这个数组是可扩容的，在添加元素时，如果队列中元素的数量大于等于内部数组的长度，就会扩容，所以是无界队列。在构造队列时，可以传一个Comparable对象，如果不传的话，就会用元素的compareTo方法排序。如果元素不是Comparable实例，就会报ClassCastException异常。PriorityBlockingQueue有4个构造器：PriorityBlockingQueue()、PriorityBlockingQueue(int initialCapacity)、PriorityBlockingQueue(int initialCapacity, Comparable comparator)、PriorityBlockingQueue(Collection c)。PriorityBlockingQueue()内部实现是PriorityBlockingQueue(11，null)，即初始容量是11，不传入比较器。因为是无界队列，所以不会有队列满的情况，所以put(E e)等同于offer(E e)方法，add(E e)也等同于offer(E e)方法。但是，remove()还是抛异常的，poll()还是返回值的，take()还是阻塞的。PriorityBlockingQueue也是线程安全的，所有操作元素的方法都由同一个非公平的ReentrantLock实例加持。

4、DelayQueue：一个支持优先级、支持延时获取元素的无界阻塞队列。队列中的元素必须实现java.util.concurrent.Delayed接口。在创建元素时，可以指定多久才能从队列中获取元素。当不到指定时间(元素的getDelay(TimeUnit unit)方法返回值大于0)或者队列为空时，remove()会抛异常，poll()会返回null，take()会阻塞，直到时间到。DelayQueue有2个构造器：DelayQueue()、DelayQueue(Collection<Delayed> c)。DelayQueue有一个PriorityQueue类型的成员变量，值是new PriorityQueue()，指定PriorityQueue初始容量是默认值11，无比较器。比较器是Delayed实例，因为Delayed接口继承了Comparable<Delayed>接口，DelayQueue中元素顺序会由元素的compareTo(Delayed d)方法决定。PriorityQueue同PriorityBlockingQueue一样，内部也是使用二叉树最小堆算法维护一个Object数组，数组可扩容。直接遍历DelayQueue队列是无序的，通过remove()/poll()/take()方法取队首元素是有序的。DelayQueue也是线程安全的，所有操作元素的方法都由同一个非公平的ReentrantLock实例加持。

DelayQueue可以用于以下场景：

缓存系统：在把数据放入缓存时，把数据的缓存到期时间戳和id放到队列中，用一个线程持续take，如果取到元素不为null，则表示对应数据到了deadline，就从缓存中删除对应数据。

定时系统：把下次执行时间戳放到队列中，用一个线程持续take，如果取到元素不为null，则表示对应任务到时间了，执行即可。

ScheduledThreadPoolExecutor类中有一个ScheduledFutureTask内部类，实现了Delayed接口。

5、SynchronousQueue：一个不存储任何元素的伪队列。在Executors.newCachedThreadPool()中用到。SynchronousQueue有2个构造器：SynchronousQueue()、SynchronousQueue(boolean fair)。SynchronousQueue()内部实现是SynchronousQueue(false)。fair同样表示阻塞的线程是否公平地访问队列。因为不存储元素，所以add(E e)、remove()恒抛异常，offer(E e)恒返回false，poll()恒返回null，put(E e)、take()恒阻塞。

Executors.newScheduledThreadPool()和Executors.newSingleThreadScheduledExecutor()用到的是DelayedWorkQueue，是ScheduledThreadPoolExecutor的内部类，继承了BlockingQueue接口。

巴特西

Activemq API使用(整合spring)

最新文章

热门文章