首页 > 代码库 > SynchronousQueue应用
SynchronousQueue应用
SynchronousQueue是无界的，是一种无缓冲的等待队列，但是由于该Queue本身的特性，在某次添加元素后必须等待其他线程取走后才能继续添加；可以认为SynchronousQueue是一个缓存值为1的阻塞队列，但是 isEmpty()方法永远返回是true，remainingCapacity() 方法永远返回是0，remove()和removeAll() 方法永远返回是false，iterator()方法永远返回空，peek()方法永远返回null。
声明一个SynchronousQueue有两种不同的方式，它们之间有着不太一样的行为。
公平模式和非公平模式的区别:如果采用公平模式：SynchronousQueue会采用公平锁，并配合一个FIFO队列来阻塞多余的生产者和消费者，从而体系整体的公平策略；
但如果是非公平模式（SynchronousQueue默认）：SynchronousQueue采用非公平锁，同时配合一个LIFO队列来管理多余的生产者和消费者，而后一种模式，如果生产者和消费者的处理速度有差距，则很容易出现饥渴的情况，即可能有某些生产者或者是消费者的数据永远都得不到处理。
=======================================================================================
SynchronousQueue是这样 一种阻塞队列，其中每个 put 必须等待一个 take，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。
不能在同步队列上进行 peek，因为仅在试图要取得元素时，该元素才存在；
除非另一个线程试图移除某个元素，否则也不能（使用任何方法）添加元素；
也不能迭代队列，因为其中没有元素可用于迭代。队列的头是尝试添加到队列中的首个已排队线程元素； 如果没有已排队线程，则不添加元素并且头为 null。
注意1：它一种阻塞队列，其中每个 put 必须等待一个 take，反之亦然。 同步队列没有任何内部容量，甚至连一个队列的容量都没有。 注意2：它是线程安全的，是阻塞的。 注意3:不允许使用 null 元素。 注意4：公平排序策略是指调用put的线程之间，或take的线程之间。 公平排序策略可以查考ArrayBlockingQueue中的公平策略。 注意5:SynchronousQueue的以下方法很有趣： * iterator() 永远返回空，因为里面没东西。 * peek() 永远返回null。 * put() 往queue放进去一个element以后就一直wait直到有其他thread进来把这个element取走。 * offer() 往queue里放一个element后立即返回，如果碰巧这个element被另一个thread取走了，offer方法返回true，认为offer成功；否则返回false。 * offer(2000, TimeUnit.SECONDS) 往queue里放一个element但是等待指定的时间后才返回，返回的逻辑和offer()方法一样。 * take() 取出并且remove掉queue里的element（认为是在queue里的。。。），取不到东西他会一直等。 * poll() 取出并且remove掉queue里的element（认为是在queue里的。。。），只有到碰巧另外一个线程正在往queue里offer数据或者put数据的时候，该方法才会取到东西。否则立即返回null。 * poll(2000, TimeUnit.SECONDS) 等待指定的时间然后取出并且remove掉queue里的element,其实就是再等其他的thread来往里塞。 * isEmpty()永远是true。 * remainingCapacity() 永远是0。 * remove()和removeAll() 永远是false。
Demo:
简化版：
import java.util.Random;import java.util.concurrent.SynchronousQueue;import java.util.concurrent.TimeUnit;public class Main { public static void main(String[] args) throws InterruptedException { SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>(); new Customer(queue).start(); new Product(queue).start(); } static class Product extends Thread{ SynchronousQueue<Integer> queue; public Product(SynchronousQueue<Integer> queue){ this.queue = queue; } @Override public void run(){ while(true){ int rand = new Random().nextInt(1000); System.out.println("生产了一个产品："+rand); System.out.println("等待三秒后运送出去..."); try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } queue.offer(rand); System.out.println("产品生成完成："+rand); } } } static class Customer extends Thread{ SynchronousQueue<Integer> queue; public Customer(SynchronousQueue<Integer> queue){ this.queue = queue; } @Override public void run(){ while(true){ try { System.out.println("消费了一个产品:"+queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("------------------------------------------"); } } }}
生产了一个产品：326等待三秒后运送出去...产品生成完成：326生产了一个产品：291等待三秒后运送出去...消费了一个产品:326------------------------------------------产品生成完成：291消费了一个产品:291------------------------------------------生产了一个产品：913等待三秒后运送出去...产品生成完成：913消费了一个产品:913------------------------------------------生产了一个产品：993等待三秒后运送出去...产品生成完成：993消费了一个产品:993------------------------------------------生产了一个产品：295等待三秒后运送出去...产品生成完成：295消费了一个产品:295------------------------------------------生产了一个产品：772等待三秒后运送出去...产品生成完成：772消费了一个产品:772------------------------------------------生产了一个产品：977等待三秒后运送出去...产品生成完成：977消费了一个产品:977------------------------------------------生产了一个产品：182等待三秒后运送出去...产品生成完成：182消费了一个产品:182------------------------------------------生产了一个产品：606等待三秒后运送出去...产品生成完成：606消费了一个产品:606------------------------------------------生产了一个产品：704等待三秒后运送出去...产品生成完成：704消费了一个产品:704------------------------------------------生产了一个产品：194等待三秒后运送出去...产品生成完成：194生产了一个产品：355等待三秒后运送出去...消费了一个产品:194------------------------------------------产品生成完成：355消费了一个产品:355------------------------------------------生产了一个产品：991等待三秒后运送出去...产品生成完成：991消费了一个产品:991------------------------------------------生产了一个产品：958等待三秒后运送出去...产品生成完成：958消费了一个产品:958------------------------------------------生产了一个产品：388等待三秒后运送出去..
从结果中可以看出如果已经生产但是还未消费的，那么会阻塞在生产一直等到消费才能生成下一个。
多线程版本：
import java.util.Random;import java.util.concurrent.SynchronousQueue;import java.util.concurrent.TimeUnit;public class Main { public static void main(String[] args) throws InterruptedException { SynchronousQueue<String> queue=new SynchronousQueue(); // TODO Auto-generated method stub for(int i=0;i<5;i++) new Thread(new ThreadProducer(queue)).start(); for(int i=0;i<5;i++) new Thread(new ThreadConsumer(queue)).start(); }}class ThreadProducer implements Runnable { ThreadProducer(SynchronousQueue<String> queue) { this.queue=queue; } SynchronousQueue<String> queue; static int cnt=0; public void run() { String name=""; int val=0; Random random =new Random(System.currentTimeMillis()); for(int i=0;i<2;i++){ cnt=(cnt+1)&0xFFFFFFFF; try{ val=random.nextInt()%15; if(val<5) { name="offer name:"+cnt; queue.offer(name); } else if(val<10) { name="put name:"+cnt; queue.put(name); } else { name="offer wait time and name:"+cnt; queue.offer(name, 1000, TimeUnit.MILLISECONDS); } Thread.sleep(1); }catch(InterruptedException e) { e.printStackTrace(); } } }}class ThreadConsumer implements Runnable { ThreadConsumer(SynchronousQueue<String> queue) { this.queue = queue; } SynchronousQueue<String> queue; public void run() { String name; for(int i=0;i<2;i++){ try { name = queue.take(); System.out.println("take " + name); Thread.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } } }}
take offer wait time and name:4take offer wait time and name:4take offer wait time and name:5take offer wait time and name:4take offer wait time and name:4take offer name:9
结果有很多种可能性，要自己尝试运行。
http://blog.csdn.net/hudashi/article/details/7076814
http://ifeve.com/java-synchronousqueue/
SynchronousQueue应用
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。