17.并发类容器设计

Posted 永恒之心

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了17.并发类容器设计相关的知识，希望对你有一定的参考价值。

并发类容器设计

    1.ConcurrentHashMap：代替散列普通的hashTable，添加了复合操作支持。

  1. private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
  2.  for (Map.Entry<String, Object> m : resultMap.entrySet()) {
  3.             count += (Long) m.getValue();
  4.  }


     2.CopyOnWriteArrayList：代替了voctor，并发的CopyOnWriteArraySet，以及并发的queue。


    3.ConcurrentLikedQueue：(高性能队列)、无阻塞形式、基于链表的无界队列，该队列不允许为空、先进先出原则。

  1. package demo6;
  3. import java.util.concurrent.ConcurrentLinkedQueue;
  5. /**高性能队列、无阻塞
  6.  */
  7. public class DemoConcurrentLinkedQueue {
  8.     /**
  9.      * 说明：list与Queue的区别
  10.      *      1.队列先进先出
  11.      *      2.list集合
  12.      */
  15. /*
  16.     使用take()函数，如果队列中没有数据，则线程wait释放CPU，而poll()则不会等待，直接返回null；同样，空间耗尽时offer()函数不会等待，直接返回false，
  17.     而put()则会wait，因此如果你使用while(true)来获得队列元素，千万别用poll()，CPU会100%的.
  18. */
  20.     public static  void main(String[] args){
  21.         ConcurrentLinkedQueue<String>  mq = new ConcurrentLinkedQueue<String>();
  22.         mq.add("小明");
  23.         mq.add("王老吉");
  24.         mq.add("大宝");
  25.         mq.add("哈士奇");
  27.         System.err.println(mq.poll());//1.取出该元素并删除
  28.         System.err.println(mq.size());
  29.         System.err.println(mq.peek());//1.取出不删除
  30.         System.err.println(mq.size());
  31.     }
  32. }
  33. 输出：
  34. 小明
3
王老吉
3

    4.LinkedBlockingQueue：(阻塞形式队列)、连链表构成、采用分离锁、读写分离、无界队列、可并行生产、消费。

    

  1. BlockingQueue<String> mq2 = new LinkedBlockingQueue<String>(5);//有界、无界
  2.         mq2.add("1+");
  3.         mq2.add("2+");
  4.         mq2.add("3+");
  5.         mq2.add("4+");
  6.         mq2.add("5+");
  7.         for (Iterator iterator = mq2.iterator();iterator.hasNext();){
  8.             String str = (String) iterator.next();
  9.             System.err.println(str);
  10.         }
  11.         List<String> list = new ArrayList<>();
  12.         System.err.println("批量获取："+mq2.drainTo(list,3));
  13.         for (String s:list){
  14.


    5.ArrayBlockingQueue：基于数组的实现阻塞形式队列、自定义长度、生产、消费不能并行执行，可先进先出、先进后出。

    

  1. BlockingQueue<String> mq1 = new ArrayBlockingQueue<String>(3);
  2.         mq1.add("a");
  3.         mq1.put("b");
  4.         mq1.add("e");


    6.PriorityBlockingQueue：基于优先级的阻塞形式，使用公平分离锁实现，优先级有comparable对象决定，实现此接口，没取一个队列排列一次队列

  1. package demo6;
  2. public class Task implements Comparable<Task> {
  4.     private int id;
  5.     private String name;
  7.     public int getId() {
  8.         return id;
  9.     }
  11.     public void setId(int id) {
  12.         this.id = id;
  13.     }
  15.     public String getName() {
  16.         return name;
  17.     }
  19.     public void setName(String name) {
  20.         this.name = name;
  21.     }
  24.     @Override
  25.     public int compareTo(Task o) {
  26.         return this.id > o.id ? 1 : (this.id < o.id ? -1 : 0);
  27.     }
  29.     @Override
  30.     public String toString() {
  31.         return "Task{" +
  32.                 "id=" + id +
  33.                 ", name=‘" + name + ‘\‘‘ +
  34.                 ‘}‘;
  35.     }
  36. }
  1. package demo6;
  3. import java.util.concurrent.PriorityBlockingQueue;
  4. public class DemoPriorityBlockingQueue {
  6.     public static  void main(String[] args) throws InterruptedException {
  7.         PriorityBlockingQueue<Task> priorityBlockingQueue = new PriorityBlockingQueue<Task>();
  9.         /**
  10.          * 每取一个队列，就排一次序
  11.          */
  13.         Task t1 = new Task();
  14.         t1.setId(3);
  15.         t1.setName(" 任务1");
  17.         Task t2 = new Task();
  18.         t2.setId(6);
  19.         t2.setName(" 任务2");
  21.         Task t3 = new Task();
  22.         t3.setId(1);
  23.         t3.setName(" 任务3");
  24.         priorityBlockingQueue.add(t1);
  25.         priorityBlockingQueue.add(t2);
  26.         priorityBlockingQueue.add(t3);
  29. /*        for (Task q:priorityBlockingQueue){
  30.             System.err.println(q.getId()+"\t"+q.getName());
  31.         }*/
  32.         System.err.println("容器："+priorityBlockingQueue);
  33.         System.err.println(priorityBlockingQueue.poll().getId());
  34.         System.err.println("容器："+priorityBlockingQueue);
  35.         System.err.println(priorityBlockingQueue.poll().getId());
  36.         System.err.println("容器："+priorityBlockingQueue);
  37.         System.err.println(priorityBlockingQueue.poll().getId());
  38.     }
  39. }
  40. 输出
  41. 容器：[Task{id=1, name=‘ 任务3‘}, Task{id=6, name=‘ 任务2‘}, Task{id=3, name=‘ 任务1‘}]
1
容器：[Task{id=3, name=‘ 任务1‘}, Task{id=6, name=‘ 任务2‘}]
3
容器：[Task{id=6, name=‘ 任务2‘}]
6

    7.synchronizeQueue：无缓冲的队列，生产消费的数据直接被消费者offer，不可直接使用add方法，必须先take方法在前，调用处于阻塞状态，等待add()方法的通知。

  1. package demo6;
  3. import java.util.concurrent.SynchronousQueue;
  4. public class DemoSynchronousQueue {
  7.     public static void main(String[] args) {
  8.         final SynchronousQueue<String> synchronousQueue = new SynchronousQueue<String>();
  9.         /**
  10.          * 不可直接使用add()方法，必须先有take方法在前调用(处于阻塞状态，等待add的通知激活)，add()方法在后,
  11.          */
  12.         Thread t1 = new Thread(new Runnable() {
  13.             @Override
  14.             public void run() {
  15.                 try {
  16.                     System.err.println(synchronousQueue.take());
  17.                 } catch (InterruptedException e) {
  18.                     e.printStackTrace();
  19.                 }
  20.             }
  21.         }, "T1");
  22.         t1.start();
  24.         Thread t2 = new Thread(new Runnable() {
  25.             @Override
  26.             public void run() {
  27.                 synchronousQueue.add("哈哈哈哈");
  28.             }
  29.         }, "T2");
  30.         t2.start();
  31.     }
  32. }

    8.delayQueue：带有延迟时间的队列。

  1. package demo6;
  3. import java.util.concurrent.Delayed;
  4. import java.util.concurrent.TimeUnit;
  5. public class WangMin implements Delayed {
  7.     private String name;
  8.     private String id;
  9.     /*结束时间*/
  10.     private long endTime;
  13.     public WangMin(String name, String id, long endTime) {
  14.         this.name = name;
  15.         this.id = id;
  16.         this.endTime = endTime;
  17.     }
  19.     public String getName() {
  20.         return name;
  21.     }
  23.     public void setName(String name) {
  24.         this.name = name;
  25.     }
  27.     public String getId() {
  28.         return id;
  29.     }
  31.     public void setId(String id) {
  32.         this.id = id;
  33.     }
  35.     public long getEndTime() {
  36.         return endTime;
  37.     }
  39.     public void setEndTime(long endTime) {
  40.         this.endTime = endTime;
  41.     }
  43.     private TimeUnit timeUnit = TimeUnit.SECONDS;
  46.     /*用来判断是否时间已经到期*/
  47.     @Override
  48.     public long getDelay(TimeUnit unit) {
  49.         return this.endTime - System.currentTimeMillis();
  50.     }
  52.     /*相互排序比较时间*/
  53.     @Override
  54.     public int compareTo(Delayed o) {
  55.         WangMin wangMin = (WangMin) o;
  56.         return this.getDelay(this.timeUnit) - wangMin.getDelay(this.timeUnit) > 0 ? 1 : 0;
  57.     }
  58. }
  1. package demo6;
  3. import java.util.concurrent.DelayQueue;
  4. public class WangBa implements Runnable{
  6.     private DelayQueue<WangMin> queue = new DelayQueue<WangMin>();
  8.     public boolean yinye = true;
  9.     public void startShangWang(String name, String id, int money) {
  10.         WangMin wangMin = new WangMin(name,id,1000*money+System.currentTimeMillis());
  11.         System.err.println("网络机器："+wangMin.getName()+"\t身份证："+wangMin.getId()+"\t金额："+money+"\t开始上网。。。");
  12.         queue.add(wangMin);
  13.     }
  15.     public void endShangWang(WangMin wangMin) {
  16.         System.err.println("网络机器："+wangMin.getName()+"\t身份证："+wangMin.getId()+"\t下机了....");
  17.     }
  19.     @Override
  20.     public void run() {
  21.         while (yinye){
  22.             try {
  23.                 WangMin min = queue.take();
  24.                 endShangWang(min);
  25.             } catch (InterruptedException e) {
  26.                 e.printStackTrace();
  27.             }
  28.         }
  29.     }
  31.     public static  void main(String[] args){
  32.         System.err.println("网吧开始...");
  33.         WangBa wangBa = new WangBa();
  34.         Thread sw = new Thread(wangBa);
  35.         sw.start();
  36.         wangBa.startShangWang("亚瑟1","xdwdasdasda12324323",5);
  37.         wangBa.startShangWang("亚瑟2","xdwdasdasda12324323",20);
  38.         wangBa.startShangWang("亚瑟3","xdwdasdasda12324323",15);
  39.     }
  40. }
  41. 输出：
  42. 网吧开始...
网络机器：亚瑟1	身份证：xdwdasdasda12324323	金额：5	开始上网。。。
网络机器：亚瑟2	身份证：xdwdasdasda12324323	金额：20	开始上网。。。
网络机器：亚瑟3	身份证：xdwdasdasda12324323	金额：15	开始上网。。。
网络机器：亚瑟1	身份证：xdwdasdasda12324323	下机了....
网络机器：亚瑟3	身份证：xdwdasdasda12324323	下机了....
网络机器：亚瑟2	身份证：xdwdasdasda12324323	下机了....

常用方法

    1.add() 添加一个元素。

    2.offer(队列,时间秒,时间范围单位)

    3.poll()取出元素并删除，否返回null。

    4.peck()取出，否返回false。

    5.tace()取出，如果队列无数据直接释放cpu。


并发类：

    1. ConcurrentLikedQueue         高性能、无阻塞、无界

    2.BlockingQueue    阻塞形式队列





以上是关于17.并发类容器设计的主要内容，如果未能解决你的问题，请参考以下文章

[Java 并发编程实战] 集合框架之 同步容器类 & 并发容器类

Java 多线程 8同步容器与并发容器

同步类容器并发修改的问题

Java高并发程序设计--JDK的并发容器

[Java 并发编程实战] 同步容器类潜在的问题（含实例代码）

[Java 并发编程实战] 同步容器类潜在的问题（含实例代码）