java并发数据结构

Wesley13
• 阅读 363

一.BlockingDeque阻塞双端队列(线程安全):

注意ArrayDeque和LinkedList仅仅扩展了Deque,是非阻塞类型的双端队列。

BlockingQueue单向队列,其内部基于ReentrantLock + Condition来控制同步和"阻塞"/"唤醒"的时机;有如下几个实现类:

  1. ArrayBlockingQueue: “浮动相对游标”的数组,来实现有界的阻塞队列。
  2. DelayQueue:支持“可延迟”的队列,DelayQueue还只接受Delayed类型的元素,Delayed接口继承自Compare接口并提供了一个long getDelay(TimeUnit),来获取指定时间到now的时间剩余量。DelayQueue底层就是使用PriorityQueue作为支撑的。
  3. PriorityBlockingQueue:有权重的队列,此队列时可以根据指定的comparator进行排序的。
  4. SynchronousQueue://
  5. LinkedBlockingDeque:有界或者无界阻塞队列

PriorityQueue为非线程安全非阻塞,有权重的队列,其权重需要根据特定的compartor来产生。

**二.ConcurrentMap(接口)**:

支持并发的map,支持多线程环境中安全的访问。

其提供了几个独特的方法:

  • V putIfAbsent(K,V):如果map中不存在此key,则put,否则返回现有的此key关联的值。此过程有Lock同步:

Java代码  

  1. //等价于:  
  2. if (!map.containsKey(key))   
  3.       return map.put(key, value);  
  4.   else  
  5. return map.get(key);   
  6. Java代码  java并发数据结构
  7. Map<String,Object> map = new ConcurrentHashMap<String, Object>();  
  8. if(map.containsKey("key")){  
  9.     map.put("key", new Object());  
  10. }  
  11. //注意,concurrentHashMap并不保证contains方法和put方法直接保持"原子性",即有可能contains方法返回false之后,在put之前,可能其他线程已经put成功,即在当前线程put时,此时数据已经不一致了.建议采用putIfAbsent

  

  • boolean remove(Object key,Object value):比较并删除指定的key和value。
  • boolean replace(K,V oldValue,V newValue):比较并替换。

目前实现ConcurrentMap的类有ConcurrentHashMap,一种基于锁分段技术实现的并发hashMap,锁采取了ReentrantLock。

三.ConcurrentLinkedQueue:

基于单向链表实现的,线程安全的并发队列,无界非阻塞队列,当队列需要在多线程环境中被使用,可以考虑使用它。记住,这是个非阻塞队列不过支持阻塞的队列,貌似都是线程安全的

此队列的size不是时间固定的,它的iterator也会被不断调整。ConcurrentLinkedQueue并没有使用Lock,而是采用了CAS的方式,对tail.next进行赋值操作。因为tail.next永远是null,且队列不接受null的元素。

注意,非并发集合(list,queue,set)的iterator以及forEach循环在并发环境中是不能正常工作的,如果原始集合被外部修改(其他线程的add,remove),将会导致异常。对于并发集合的iterator,没有做相关的size校验。

Lock(锁)是控制操作(action)的,可以让一个操作或者一个子操作被串行的处理。。。CAS其实只是对内存数据的变更时使用,如果想知道数据变更在并发环境中是否符合预期,才会使用到CAS。

四.ConcurrentSkipListMap/ConcurrentSkipListSet

两个基于SkipList(跳跃表)方式实现的、支持并发访问的数据结构。依据跳跃表的思想,可以提高数据访问的效率。其中ConcurrentSkipListSet底层使用ConcurrentSkipListMap支撑。

ConcurrentSkipListMap也是ConcurrentNavigableMap的实现类,对于SkipList,其内部元素,必须是可排序的。

跳跃表是一个很简单的表,(参见跳跃表概念),对底层的线性存储结构,加入类似“多级索引”的概念,“索引”的添加时基于随即化。一个跳跃表,整体设计上(设计思路很多)分为表左端head索引,右端tail索引(边界),底端存储层(排序的线性链表),和一个随机化、散列化的不同高度的多级索引“指针”。head索引是高度最高的索引,它是整个链表中值最小的元素锁产生的索引;右端为边界索引,索引指向null或者任意设计的边界值(bound).

跳跃表的底端是一个和普通的链表没啥区别,单向或者双向的均可,前提是必须是排序的。索引节点,就是一个有向路径的标,每个索引节点,都分别有right、down指向,对于双向跳跃表,就具有left、right、up、down四个方向指针;指针就是为了方便寻路。每个新增元素时,首先会导致底层链表的改动;根据自定义的随即算法,来决定此元素的索引高度,如果高度不为0,则依次建立相应层高的索引,并调整各个层高的所以指向。

跳跃表之所以这么设计,实事上就是在做一件事情:基于简单的设计思路和算法,来实现较为高效的查询策略。相对于二分查找有一定的优势.

五.CopyOnWriteArrayList/CopyOnWriteArraySet:

均是CopyOnWrite思想,在数据修改时(happen-before),对数据进行Copy(),read操作可以在原数据结构上继续进行,待write操作结束后,调整数据结构指针。基于这种设计思路的数据结构,通常是read操作频率远大于write操作,可以在并发环境中,支撑较高的吞吐量;避免了因为同步而带来的瓶颈,同时也能确保数据安全操作。同时需要注意,copy操作将会带来多余的空间消耗。注意,此API时支持并发的,多个线程add操作(即CopyOnWrite)将会被队列化,内部采取了ReentrantLock机制来控制.

  • CopyOnWriteArrayList底层基于数组实现,在进行write操作时(add,remove),将会导致Arrays.copy操作,创建一个新的数组;待write操作成功后,将原数组的指针替换成新数组指针.
  • CopyOnWriteArraySet底层直接使用CopyOnWriteArrayList作为支撑,只不过在add操作时会遍历整个数组结构并进行equals比较(确保具有Set的特性),只有发现此新元素不存在时才会"替换指针".

    java中这两个API,支持并发操作时,仍然可以进行遍历而无需额外的同步;即不会抛出ConcurrentModificationException。事实上,迭代器所持有的数组只是一个"创建iterator时底层数组的引用",所以在遍历期间,即使CopyOnWriteArrayList已经新增或者删除了某些元素,仍不会发生冲突,因为iterator持有的是旧数组的引用,而CopyOnWriteArrayList持有的是Copy操作时创建的新数组引用..由此可见,iterator也无法反应实时的数组变化(遍历期间,实际数组的添加、删除),但是原始数组中对象内容发生改变还是可以在迭代器中反应出来。CopyOnWrite的遍历器的remove/add/set操作不被支持,这区别于ArrayList.

    CopyOnWriteArrayList、CopyOnWriteArraySet,底层基于数组实现,采取ReentrantLock来同步add/remove/clear等操作。并采取了snapshot的简单手段:

Java代码  java并发数据结构

  1. //例如add:  

  2. public boolean add(E e) {  

  3.     final ReentrantLock lock = this.lock;  

  4.     lock.lock();  

  5.     try {  

  6.        Object[] elements = getArray();  

  7.        int len = elements.length;  

  8.         //数组copy  

  9.        Object[] newElements = Arrays.copyOf(elements, len + 1);  

  10.        newElements[len] = e;  

  11.         //修改结束后,指针转换  

  12.        setArray(newElements);  

  13.        return true;  

  14.     } finally {  

  15.        lock.unlock();  

  16.     }  

  17. }

  

六.CountDownLatch:

同步类,用于多个线程协调工作。共享锁,当锁计数器较少到0时,将释放等待的线程。使用场景是,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。当CountDownLatch的锁计数器为1时,可以作为一种“开关”来使用。计数器无法被重置,如果需要重复计数,可以使用CyclicBarrier。

CountDownLatch内部基于AQS来控制线程访问。这个API很简单,只有2个核心方法:

  • void await():如果计数器不为0,则获取锁失败,加入同步队列;即线程阻塞。

  • void countDown():释放锁,导致计数器递减,如果此时计数器为0,则表示锁释放成功,AQS会帮助“发射”因为await阻塞的线程(组)。

Java代码  java并发数据结构

  1. public class CountDownLatchTestMain {  

  2.     /** 

  3.     * @param args 

  4.     */  

  5.     public static void main(String[] args) throws Exception{  

  6.         System.out.println("Begin");  

  7.         CountDownLatch latch = new CountDownLatch(2);  

  8.         for(int i=0;i<4;i++){  

  9.             CThread c = new CThread(i,latch);  

  10.             c.start();  

  11.             //Thread.sleep(500);  

  12.         }  

  13.         Thread.sleep(1000);  

  14.         System.out.println("End");  

  15.     }  

  16.     static class CThread extends Thread{  

  17.         CountDownLatch latch;  

  18.         int count;  

  19.         CThread(int count,CountDownLatch latch){  

  20.             this.count = count;  

  21.             this.latch = latch;  

  22.         }  

  23.         @Override  

  24.         public void run(){  

  25.             try{  

  26.                 System.out.println("---"+count);  

  27.             if(count % 2 == 0){  

  28.                 latch.await();  

  29.                 System.out.println("//" + count + "await--!");  

  30.             }else{  

  31.                 latch.countDown();  

  32.                 System.out.println("//" + count + "down!");  

  33.             }  

  34.             }catch(Exception e){  

  35.                 e.printStackTrace();  

  36.             }  

  37.         }  

  38.     }  

  39. }

  

七.CyclicBarrier:

循环屏障,它允许一组线程互相等待,直到到达某个公共屏障点;线程(组)数量固定,线程之间需要不时的互相等待,CyclicBarrier和CountDownLatch相比,它可以在释放等待线程后被再次“重用”,所以称为循环屏障。它提供了类似“关卡”的功能。对于失败的同步尝试,CyclicBarrier 使用了一种要么全部要么全不 (all-or-none) 的破坏模式:如果因为中断、失败或者超时等原因,导致线程过早地离开了屏障点,那么在该屏障点等待的其他所有线程也将通过 BrokenBarrierException(如果它们几乎同时被中断,则用 InterruptedException)以反常的方式离开。

  • CyclicBarrier(int parties):指定参与者个数

  • CyclicBarrier(int parties,Runnable barrierAction):指定一个屏障操作,此操作将会有最后一个进入barrier的线程执行。

  • int await():在所有的线程达到barrier之前,一直等待。此方法可以抛出InterrutedExeception(此线程被中断),可以抛出BrokenBarrierExeception(当其他参与者在wait期间中断,导致屏障完整性被破坏),在方法被await时,如果抛出上述异常,需要做补救的相应操作。此方法返回当前线程到达屏障时的索引。(第一个到达的,为0,最后一个为getParties() - 1);根据返回值的不同可以做一些操作,比如最先/最后达到的做一些前置、后置操作等。

  • boolean isBroken():屏障是否处于损坏状态。

  • void reset():重置屏障为其初始状态;如果此时有线程在await,其线程将会抛出BrokenBarrierExeception。对于reset操作,需要线程的执行方法有相应的配合(比如支持操作轮训等),否则重置会带来一些不必要的麻烦。。。如果你需要重置,尚不如重新建一个CyclicBarrier。

底层基于ReentrantLock实现。线程阻塞基于Condition方式(注意Condition底层也是通过AQS框架实现);大概伪代码:

Java代码  java并发数据结构

  1. ReentrantLock lock = new ReentrantLock();  
  2. Condition trip = lock.newCondition();  
  3. ////await方法:  
  4. if(count!=0){  
  5.     trip.await();//AQS:当前线程队列化,LockSupport.park  
  6.     count--;  
  7. }else{  
  8.     trip.signalAll();  
  9. }

  

Java代码  java并发数据结构

  1. //////////////////代码实例  

  2. public class CyclicBarrierTestMain {  

  3.     /** 

  4.     * @param args 

  5.     */  

  6.     public static void main(String[] args) throws Exception{  

  7.         CyclicBarrier barrier = new CyclicBarrier(5, new Runnable() {  

  8.             @Override  

  9.             public void run() {  

  10.                 System.out.println("Barrier action!!");  

  11.             }  

  12.             });  

  13.         for(int i=0;i<5;i++){  

  14.             CThread c = new CThread(barrier);  

  15.             c.start();  

  16.         }  

  17.         Thread.sleep(1000);  

  18.     }  

  19.     static class CThread extends Thread{  

  20.         CyclicBarrier barrier;  

  21.         CThread(CyclicBarrier barrier){  

  22.             this.barrier = barrier;  

  23.         }  

  24.         @Override  

  25.         public void run(){  

  26.             int count = 0;  

  27.             while(true){  

  28.                 try{  

  29.                     System.out.print("---" + count);  

  30.                     int index = barrier.await();  

  31.                     System.out.println("+++" + count);  

  32.                     count++;  

  33.                     if(index == barrier.getParties() - 1){  

  34.                         //barrier.reset();  

  35.                     }  

  36.                 }catch(Exception e){  

  37.                     e.printStackTrace();  

  38.                     break;  

  39.                 }  

  40.             }  

  41.         }  

  42.     }  

  43. }

  

八.Exchanger

Exchanger:同步交换器,2个互相匹配(协调的对象),互相交换数据。2个线程需要把相同类型的数据,以互相等待的方式交换。比如线程1将数据A交换给B,此时线程1等待直到线程B将数据交换出去。Exchanger有一个方法,就是exchange(V x):其作用就是等待另一个线程到达交换点,然后将数据传递给线程。

如果没有其他线程到达交换点,处于调度的目的,禁用当前线程,直到某个线程到达或者某个线程中断。

伪代码:

Java代码  java并发数据结构

  1. void exchange(V item){  
  2.     //如果有线程已经到达  
  3.     for(;;){  
  4.         Node e = get();  
  5.         if(e != null){  
  6.             V i = e.getItem();  
  7.             CAS(e,i,null);//将等待匹配者移除  
  8.             Thread t = e.waiter;  
  9.             LockSupport.unpark(t);  
  10.             //  
  11.             Node ne = new Node(currentThread,ne);  
  12.             set();//将当前需要交换的数据加入,当其他线程unpart之后,可以get,并获取数据  
  13.             return i;//返回需要交换的数据  
  14.         }else{  
  15.             Node e = new Now(currentThread,item);  
  16.             set(node);  
  17.             LockSupport.park(currentThread);  
  18.         //重新回到顶层for循环,并获取交换数据  
  19.         }  
  20.     }  
  21. }

  

如下的例子是基于一个简单的Productor和Consumer模式,一个线程负责生产数据,当数据满时,交换给consumer消费;当consumer消费完时,也申请交换。

Java代码  java并发数据结构

  1. import java.util.ArrayDeque;  

  2. import java.util.Queue;  

  3. import java.util.Random;  

  4. import java.util.concurrent.Exchanger;  

  5. public class ExchangerTestMain {  

  6.     /** 

  7.     * @param args 

  8.     */  

  9.     public static void main(String[] args) throws Exception{  

  10.         Exchanger<Queue> exchanger = new Exchanger<Queue>();  

  11.         CThread c = new CThread(exchanger);  

  12.         PThread p = new PThread(exchanger);  

  13.         c.start();  

  14.         p.start();  

  15.         Thread.sleep(2000);  

  16.     }  

  17.     static class CThread extends Thread{  

  18.         Exchanger<Queue> exchanger ;  

  19.         Queue current;  

  20.         CThread(Exchanger<Queue> exchanger){  

  21.             this.exchanger = exchanger;  

  22.         }  

  23.         @Override  

  24.         public void run(){  

  25.             if(current == null){  

  26.                 current = new ArrayDeque(10);  

  27.             }  

  28.             try{  

  29.                 while(true){  

  30.                 //productor  

  31.                 if(current.size() == 0){  

  32.                     current = exchanger.exchange(current);//交换出去fullList,希望获得EmptyList  

  33.                 }  

  34.                     System.out.println("C:" + current.poll());  

  35.                 }  

  36.             }catch(Exception e){  

  37.                 e.printStackTrace();  

  38.                 return;  

  39.             }  

  40.         }  

  41.     }  

  42.     static class PThread extends Thread{  

  43.         Exchanger<Queue> exchanger ;  

  44.         Queue current;  

  45.         PThread(Exchanger<Queue> exchanger){  

  46.             this.exchanger = exchanger;  

  47.         }  

  48.         @Override  

  49.         public void run(){  

  50.             Random r = new Random();  

  51.             if(current == null){  

  52.                 current = new ArrayDeque(10);  

  53.             }  

  54.             try{  

  55.                 while(true){  

  56.                     //productor  

  57.                     if(current.size() >= 10){  

  58.                         current = exchanger.exchange(current);//交换出去fullList,希望获得EmptyList  

  59.                     }  

  60.                     Integer i = r.nextInt(20);  

  61.                     System.out.println("P:" + i);  

  62.                     current.add(i);  

  63.                 }  

  64.             }catch(Exception e){  

  65.                 e.printStackTrace();  

  66.                 return;  

  67.             }  

  68.         }  

  69.     }  

  70. }

  

九.Semaphore:信号量

我们需要把semaphore真的看成“信号量”,它是可以被“增减”的锁引用,“0”是判断信号“过剩”的界限。

我们通常使用semaphore来控制资源访问并发量。它底层使用“共享”模式锁实现,提供了“公平”“非公平”2中策略。当“信号量”大于0时,允许获取锁;否则将阻塞直到信号量恢复。

将信号量初始化为 1,使得它在使用时最多只有一个可用的许可,从而可用作一个相互排斥的锁。这通常也称为二进制信号量,因为它只能有两种状态:一个可用的许可,或零个可用的许可。按此方式使用时,二进制信号量具有某种属性(与很多 Lock 实现不同),即可以由线程释放“锁”,而不是由所有者(因为信号量没有所有权的概念)。在某些专门的上下文(如死锁恢复)中这会很有用。

  • Semaphore(int permits, boolean fair):指定信号量,指定公平策略。

  • void acquire():获取一个信号,如果信号量<=0,则阻塞;在非公平模式下,允许闯入。

  • void acquire(int permits).

上面2个方法都会抛出InterruptException,即在等待线程被“中断时”,将会抛出异常而返回。底层基于AQS.acquireSharedInterruptibly()

  • void acquireUninterruptibly():获取一个信号,不支持中断,当线程被中断时,此线程将继续等待,当线程确实从此方法返回后,将设置其中断状态。底层基于AQS.acquireShared();
  • void release():释放一个信号,直接导致信号量++。
  • boolean tryAcquire():获取一个信号,如果获取成功,则返回true。

内容整理自:http://blog.csdn.net/androidstudio/article/details/24633495

点赞
收藏
评论区
推荐文章
技术小男生 技术小男生
4个月前
linux环境jdk环境变量配置
1:编辑系统配置文件vi/etc/profile2:按字母键i进入编辑模式,在最底部添加内容:JAVAHOME/opt/jdk1.8.0152CLASSPATH.:$JAVAHOME/lib/dt.jar:$JAVAHOME/lib/tools.jarPATH$JAVAHOME/bin:$PATH3:生效配置
光头强的博客 光头强的博客
4个月前
Java面向对象试题
1、请创建一个Animal动物类,要求有方法eat()方法,方法输出一条语句“吃东西”。创建一个接口A,接口里有一个抽象方法fly()。创建一个Bird类继承Animal类并实现接口A里的方法输出一条有语句“鸟儿飞翔”,重写eat()方法输出一条语句“鸟儿吃虫”。在Test类中向上转型创建b对象,调用eat方法。然后向下转型调用eat()方
刚刚好 刚刚好
4个月前
css问题
1、在IOS中图片不显示(给图片加了圆角或者img没有父级)<div<imgsrc""/</divdiv{width:20px;height:20px;borderradius:20px;overflow:h
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
小森森 小森森
4个月前
校园表白墙微信小程序V1.0 SayLove -基于微信云开发-一键快速搭建,开箱即用
后续会继续更新,敬请期待2.0全新版本欢迎添加左边的微信一起探讨!项目地址:(https://www.aliyun.com/activity/daily/bestoffer?userCodesskuuw5n)\2.Bug修复更新日历2.情侣脸功能大家不要使用了,现在阿里云的接口已经要收费了(土豪请随意),\\和注意
晴空闲云 晴空闲云
4个月前
css中box-sizing解放盒子实际宽高计算
我们知道传统的盒子模型,如果增加内边距padding和边框border,那么会撑大整个盒子,造成盒子的宽度不好计算,在实务中特别不方便。boxsizing可以设置盒模型的方式,可以很好的设置固定宽高的盒模型。盒子宽高计算假如我们设置如下盒子:宽度和高度均为200px,那么这会这个盒子实际的宽高就都是200px。但是当我们设置这个盒子的边框和内间距的时候,那
艾木酱 艾木酱
3个月前
快速入门|使用MemFire Cloud构建React Native应用程序
MemFireCloud是一款提供云数据库,用户可以创建云数据库,并对数据库进行管理,还可以对数据库进行备份操作。它还提供后端即服务,用户可以在1分钟内新建一个应用,使用自动生成的API和SDK,访问云数据库、对象存储、用户认证与授权等功能,可专
NVIDIA安培架构下MIG技术分析
关键词:NVIDIA、MIG、安培一什么是MIG2020年5月,NVIDIA发布了最新的GPU架构:安培,以及基于安培架构的最新的GPU:A100。安培提供了许多新的特性,MIG是其中一项非常重要的新特性。MIG的全名是MultiInstanceGPU。NVIDIA安培架构中的MIG模式可以在A100GPU上并行运行七个作业。多实
helloworld_28799839 helloworld_28799839
4个月前
常用知识整理
Javascript判断对象是否为空jsObject.keys(myObject).length0经常使用的三元运算我们经常遇到处理表格列状态字段如status的时候可以用到vue
密钥管理系统-为你的天翼云资产上把“锁
本文关键词:数据安全,密码机,密钥管理一、你的云上资产真的安全么?1.2021年1月,巴西的一个数据库30TB数据被破坏,泄露的数据包含有1.04亿辆汽车和约4000万家公司的详细信息,受影响的人员数量可能有2.2亿;2.2021年2月,广受欢迎的音频聊天室应用Clubhouse的用户数据被恶意黑客或间谍窃取。据悉,一位身份不明的用户能够将Clubho