并发容器与并发工具类的使用

92 阅读 0 评论 61 点赞

我是靠谱客的博主秀丽身影，最近开发中收集的这篇文章主要介绍并发容器与并发工具类的使用，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

同步容器

Vector，HashTable – jdk提供的同步容器类

Collections.synchronizedXXX 本质是对响应的容器进行包装
同步容器类的缺点

在单独使用里面的方法的时候，可以保证线程安全，但是符合操作需要额外加锁来保证线程安全，使用Interator迭代容器或使用for–each遍历容器，在迭代过程中修改容器会抛出CurrentModificationException异常。要想避免CurrentModificationException，就必须在迭代的过程中持有容器的锁。但是若容器交大的话，迭代的时间会较长，那么需要访问该容器的线程就会进入长时间的等待，从而极大的降低性能。

若不希望在迭代的过程中对容器加锁，可以使用克隆容器的方式，使用线程封闭，由于其他线程不会对容器进行修改，可以避免CurrentModificationException，但是在创建副本的时候，存在较大性能开销。
并发容器

CopyOnWrite、Concurrent、BlockingQueue+

根据具体的场景进行设计，尽量避免使用锁，提高容器的并发访问性。

ConcurrentBlockingQueue 基于queue实现的FIFO先进先出的队列，队列为空，取操作会被阻塞

ConcurrentLinkedQueue 队列为空，取的时候直接返回空

同步容器Vector的坑：

import java.util.Iterator;
import java.util.Vector;

public class VectorDemo {

    public static void main(String[] args) {
        Vector<String> strings = new Vector<>();
        for(int i = 0; i < 1000; i++){
            strings.add("demo" + i);
        }

        //错误迭代
        /*strings.forEach(e -> {
            if(e.equals("demo3")){
                strings.remove(e);   //ConcurrentModificationException
            }
            System.out.println(e);
        });*/

        //正确的迭代
        Iterator<String> iterator = strings.iterator();
        /*单线程下正确的迭代方式
        while (iterator.hasNext()){
            String next = iterator.next();
            if("demo3".equals(next)){
                iterator.remove();
            }
        }*/

        for(int i = 0; i < 4; i++){
            new Thread(() -> {
                synchronized (iterator) {
                    while (iterator.hasNext()) {
                        String next = iterator.next();
                        if ("demo3".equals(next)) {
                            iterator.remove();    //不加锁的话，在多线程下回报错NoSuchElementException
                        }
                    }
                }
            }).start();
        }
    }
}

并发容器使用代码示例：

import java.util.concurrent.CopyOnWriteArrayList;

public class Demo {

    public static void main(String[] args) {
        CopyOnWriteArrayList<String> stringCopyOnWriteArrayList = new CopyOnWriteArrayList<>();
        for(int i = 0; i < 1000; i++){
            stringCopyOnWriteArrayList.add("demo"+i);
        }

        //可以正常访问
       /* stringCopyOnWriteArrayList.forEach(e -> {
            if("demo2".equals(e)){
                stringCopyOnWriteArrayList.remove(e);
            }
        });*/

      /*  Iterator<String> iterator = stringCopyOnWriteArrayList.iterator();
        while (iterator.hasNext()){
            String next = iterator.next();
            if("demo2".equals(next)){
                //CopyOnWriteArrayList不支持在迭代器里面移除元素
                iterator.remove();     //UnsupportedOperationException
            }
        }*/

        for (int i = 0; i < 4; i++){
            new Thread(() -> {
                stringCopyOnWriteArrayList.forEach(e -> {
                    if("demo2".equals(e)){
                        stringCopyOnWriteArrayList.remove(e);
                    }
                });
            }).start();
        }
    }
}

LinkedBlockingQueue,在并发编程中使用非常频繁，因其可以作为生产者消费者的中间商，没有元素的时候会阻塞

LinkedBlockingQueue的使用代码示例

import java.util.concurrent.LinkedBlockingDeque;

public class Demo {
    public static void main(String[] args) throws InterruptedException {
        LinkedBlockingDeque<String> strings = new LinkedBlockingDeque<>();

        //向队列中放入元素
        strings.add("");   //实际调用的是offer，在队列满的时候，会抛出异常
        strings.offer(""); //如果队列满了，会返回如对失败
        strings.put("");   //作为生产者，如果队列满了，会进入阻塞状态

        //从队列中去元素
        String remove = strings.remove(); //返回移除的元素，会抛出异常
        strings.poll();   //在队列为空的时候，返回为null
        strings.take();   //作为消费者,在队列为空的时候会进入等待状态
    }
}

add 调用的是offer

offer 队列如果满了直接入队失败

CountDownLatch

应用场景：启动三个线程计算，需要对结果进行累加
```
await();      //进入等待状态
countDown();  //计数器减一
```

CountDownLatch使用代码示例

import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {

    public static void main(String[] args) {

        CountDownLatch countDownLatch = new CountDownLatch(8);

        new Thread(() -> {
            try {
                //进入等待状态，等countDownLatch减为0时才执行下面的操作
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("800密比赛结束，准备清空跑道并进行跨栏比赛");
        }).start();

        for (int i = 0; i < 8; i++){
            int finalI = i;
            new Thread(() -> {
                try {
                    Thread.sleep(finalI * 1000L);
                    System.out.println(Thread.currentThread().getName()+"----比赛结束");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    //在线程执行结束后，减1
                    countDownLatch.countDown();
                }
            }).start();
        }
    }
}
/*
    执行结果：
    Thread-1----比赛结束
    Thread-2----比赛结束
    Thread-3----比赛结束
    Thread-4----比赛结束
    Thread-5----比赛结束
    Thread-6----比赛结束
    Thread-7----比赛结束
    Thread-8----比赛结束
    800密比赛结束，准备清空跑道并进行跨栏比赛
*/

CyclicBarrier–栅栏

允许一组线程达到一个公共的障碍点，之后再继续执行
CyclicBarrier与CountDownLatch的区别
- CountDownLatch一般用于一个线程等待若干个其他线程执行完成任务后，它才执行，不可重复使用
- CyclicBarrier一般用于一组线程相互等待至某个状态，然后这一组线程再同时执行，可重用的

CyclicBarrier代码示例

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierDemo {

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(8);

        for (int i = 0; i < 8; i++) {
            int finalI = i;
            new Thread(() -> {
                try {
                    Thread.sleep(finalI * 1000L);
                    System.out.println(Thread.currentThread().getName() + "----准备就绪");
                    //执行到就绪状态的时候，进入等待状态
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }

                //等所有的线程都进入到就绪状态，再同时执行下面的操作
                System.out.println(Thread.currentThread().getName() + "----开始比赛");
            }).start();
        }
    }

}

Semaphore – 信号量

控制线程的并发数量
Semaphore 的使用场景：接口限流

代码示例

import java.util.concurrent.Semaphore;

public class SemaphoreDemo {

    public static void main(String[] args) {

        Semaphore semaphore = new Semaphore(8);
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    semaphore.acquire();  //获得信号,当信号量不足时进入等待状态
                    System.out.println(Thread.currentThread().getName() + "----开始执行");
                    Thread.sleep(5 * 1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release(); //释放信号
                }
            }).start();
        }
    }
}

Exchanger：用来两个线程之间交换数据

它提供了一个同步点，在这个同步点两个线程可以交换数据。这两个线程通过exchanger交换数据，如果第一个线程先执行exchange方法，它会一直等待第二个线程执行exchange，当两个线程都达到同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。所以使用exchanger的重点是成对的线程使用exchange()方法，当有一堆线程达到了同步点，就会进行交换数据，因此该工具类的线程对象时"成对"的。

Exchanger使用代码示例

import java.util.concurrent.Exchanger;

public class ExchangerDemo {

    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();
        String str1 = "str1";
        String str2 = "str2";

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "===初始值1===" + str1);
            try {
                String exchange = exchanger.exchange(str1);
                System.out.println(Thread.currentThread().getName() + "===交换值1===" + exchange);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "线程1").start();

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "===初始值2===" + str2);
            try {
                String exchange = exchanger.exchange(str2);
                System.out.println(Thread.currentThread().getName() + "===交换值2===" + exchange);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "线程2").start();

    }

}

/*
	执行结果：
        线程1===初始值1===str1
        线程2===初始值2===str2
        线程2===交换值2===str1
        线程1===交换值1===str2
*/