标签：
在研究Smack的源码的时候，我对它的连接Connection以及派生类XMPPConnection的关注是最多的，由于一个即时通信程序，它的网络模块必是它的核心。
而我非常在乎它是怎样实现的。
在收发数据包的时候，我看到了队列的身影。BlockingQueue和ArrayBlockingQueue。所以，我认为用到什么然后去查阅。去记录，这样的方法是比較高效率的。
BlockingQueue是在Java的新的Concurrent包中的。
Reference：
http://www.cnblogs.com/jackyuj/archive/2010/11/24/1886553.html
在新增的Concurrent包中。BlockingQueue非常好的攻克了多线程中。怎样高效安全“传输”数据的问题。通过这些高效而且线程安全的队列类。为我们高速搭建高质量的多线程程序带来极大的便利。本文具体介绍了BlockingQueue家庭中的全部成员，包含他们各自的功能以及常见使用场景。
假设我们有若干生产者线程，另外又有若干个消费者线程。假设生产者线程须要把准备好的数据共享给消费者线程。利用队列的方式来传递数据，就能够非常方便地解决他们之间的数据共享问题。但假设生产者和消费者在某个时间段内。万一发生数据处理速度不匹配的情况呢？理想情况下，假设生产者产出数据的速度大于消费者消费的速度，而且当生产出来的数据累积到一定程度的时候。那么生产者必须暂停等待一下（堵塞生产者线程）。以便等待消费者线程把累积的数据处理完成，反之亦然。然而，在concurrent包公布曾经。在多线程环境下，我们每一个程序猿都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。好在此时。强大的concurrent包横空出世了，而他也给我们带来了强大的BlockingQueue。（在多线程领域：所谓堵塞。在某些情况下会挂起线程（即堵塞）。一旦条件满足。被挂起的线程又会自己主动被唤醒）
如上图所看到的：当队列中没有数据的情况下。消费者端的全部线程都会被自己主动堵塞（挂起）。直到有数据放入队列。
作为BlockingQueue的使用者。我们再也不须要关心什么时候须要堵塞线程，什么时候须要唤醒线程。由于这一切BlockingQueue都给你一手包办了。既然BlockingQueue如此神通广大，让我们一起来见识下它的经常用法：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
/**
* @author jackyuj
*/
public class BlockingQueueTest {
public static void main(String[] args) throws InterruptedException {
// 声明一个容量为10的缓存队列
BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
Producer producer1 = new Producer(queue);
Producer producer2 = new Producer(queue);
Producer producer3 = new Producer(queue);
Consumer consumer = new Consumer(queue);
// 借助Executors
ExecutorService service = Executors.newCachedThreadPool();
// 启动线程
service.execute(producer1);
service.execute(producer2);
service.execute(producer3);
service.execute(consumer);
// 运行10s
Thread.sleep(10 * 1000);
producer1.stop();
producer2.stop();
producer3.stop();
Thread.sleep(2000);
// 退出Executor
service.shutdown();
}
}import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* 消费者线程
*
* @author jackyuj
*/
public class Consumer implements Runnable {
public Consumer(BlockingQueue<String> queue) {
this.queue = queue;
}
public void run() {
System.out.println("启动消费者线程！");
Random r = new Random();
boolean isRunning = true;
try {
while (isRunning) {
System.out.println("正从队列获取数据...");
String data = queue.poll(2, TimeUnit.SECONDS);
if (null != data) {
System.out.println("拿到数据：" + data);
System.out.println("正在消费数据：" + data);
Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
} else {
// 超过2s还没数据。觉得全部生产线程都已经退出，自己主动退出消费线程。
isRunning = false;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
} finally {
System.out.println("退出消费者线程！
");
}
}
private BlockingQueue<String> queue;
private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
}
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 生产者线程
*
* @author jackyuj
*/
public class Producer implements Runnable {
public Producer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
String data = null;
Random r = new Random();
System.out.println("启动生产者线程！");
try {
while (isRunning) {
System.out.println("正在生产数据...");
Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
data = "data:" + count.incrementAndGet();
System.out.println("将数据：" + data + "放入队列...");
if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
System.out.println("放入数据失败：" + data);
}
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
} finally {
System.out.println("退出生产者线程！");
}
}
public void stop() {
isRunning = false;
}
private volatile boolean isRunning = true;
private BlockingQueue queue;
private static AtomicInteger count = new AtomicInteger();
private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
}
相对于有缓冲的BlockingQueue来说，少了一个中间经销商的环节（缓冲区），假设有经销商。生产者直接把产品批发给经销商，而无需在意经销商终于会将这些产品卖给那些消费者。由于经销商能够库存一部分商品。因此相对于直接交易模式，整体来说採用中间经销商的模式会吞吐量高一些（能够批量买卖）；但还有一方面，又由于经销商的引入，使得产品从生产者到消费者中间添加了额外的交易环节。单个产品的及时响应性能可能会减少。
package org.jivesoftware.smack;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.jivesoftware.smack.filter.PacketFilter;
import org.jivesoftware.smack.packet.Packet;
/**
* Provides a mechanism to collect packets into a result queue that pass a
* specified filter. The collector lets you perform blocking and polling
* operations on the result queue. So, a PacketCollector is more suitable to
* use than a {@link PacketListener} when you need to wait for a specific
* result.<p>
*
* Each packet collector will queue up a configured number of packets for processing before
* older packets are automatically dropped. The default number is retrieved by
* {@link SmackConfiguration#getPacketCollectorSize()}.
*
* @see Connection#createPacketCollector(PacketFilter)
* @author Matt Tucker
*/
public class PacketCollector {
private PacketFilter packetFilter;
private ArrayBlockingQueue<Packet> resultQueue;
private Connection connection;
private boolean cancelled = false;
/**
* Creates a new packet collector. If the packet filter is <tt>null</tt>, then
* all packets will match this collector.
*
* @param conection the connection the collector is tied to.
* @param packetFilter determines which packets will be returned by this collector.
*/
protected PacketCollector(Connection conection, PacketFilter packetFilter) {
this(conection, packetFilter, SmackConfiguration.getPacketCollectorSize());
}
/**
* Creates a new packet collector. If the packet filter is <tt>null</tt>, then
* all packets will match this collector.
*
* @param conection the connection the collector is tied to.
* @param packetFilter determines which packets will be returned by this collector.
* @param maxSize the maximum number of packets that will be stored in the collector.
*/
protected PacketCollector(Connection conection, PacketFilter packetFilter, int maxSize) {
this.connection = conection;
this.packetFilter = packetFilter;
this.resultQueue = new ArrayBlockingQueue<Packet>(maxSize);
}
/**
* Explicitly cancels the packet collector so that no more results are
* queued up. Once a packet collector has been cancelled, it cannot be
* re-enabled. Instead, a new packet collector must be created.
*/
public void cancel() {
// If the packet collector has already been cancelled, do nothing.
if (!cancelled) {
cancelled = true;
connection.removePacketCollector(this);
}
}
/**
* Returns the packet filter associated with this packet collector. The packet
* filter is used to determine what packets are queued as results.
*
* @return the packet filter.
*/
public PacketFilter getPacketFilter() {
return packetFilter;
}
/**
* Polls to see if a packet is currently available and returns it, or
* immediately returns <tt>null</tt> if no packets are currently in the
* result queue.
*
* @return the next packet result, or <tt>null</tt> if there are no more
* results.
*/
public Packet pollResult() {
return resultQueue.poll();
}
/**
* Returns the next available packet. The method call will block (not return)
* until a packet is available.
*
* @return the next available packet.
*/
public Packet nextResult() {
try {
return resultQueue.take();
}
catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
/**
* Returns the next available packet. The method call will block (not return)
* until a packet is available or the <tt>timeout</tt> has elapased. If the
* timeout elapses without a result, <tt>null</tt> will be returned.
*
* @param timeout the amount of time to wait for the next packet (in milleseconds).
* @return the next available packet.
*/
public Packet nextResult(long timeout) {
try {
return resultQueue.poll(timeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
/**
* Processes a packet to see if it meets the criteria for this packet collector.
* If so, the packet is added to the result queue.
*
* @param packet the packet to process.
*/
protected void processPacket(Packet packet) {
if (packet == null) {
return;
}
if (packetFilter == null || packetFilter.accept(packet)) {
while (!resultQueue.offer(packet)) {
// Since we know the queue is full, this poll should never actually block.
resultQueue.poll();
}
}
}
}
标签：
原文地址：http://www.cnblogs.com/mengfanrong/p/5146368.html
