多线程理解（九）同步容器

一、线程不安全的类：如果一个类的对象可以同时被多个线程访问，并且不做特殊的同步或并发处理，那么它就很容易表现出不安全的现象。比如抛出异常，逻辑处理错误。

1.StringBuilder与StringBuffer

StringBuilder是线程不安全的，StringBuffer是线程安全的。

StringBuffer的方法使用了synchronized关键字修饰。

2.ArrayList，HashMap，HashSet等Collection类。

二、同步容器：同步容器有两类

一类是Vector，Stack，HashTable；另一类是Collections类中提供的静态工厂方法创建的类。

Vector实现了List接口，Vector实际上就是一个数组，和ArrayList非常的类似，但是内部的方法都是使用synchronized修饰过的方法。
Stack它的方法也是使用synchronized修饰了，继承了Vector，实际上就是栈。

1.同步容器真的是安全的吗

（1）删除与获取并发操作

@NotThreadSafe

public class Test {
    static Vector<Integer> vector = new Vector<Integer>();
    public static void main(String[] args){
        while(true) {
            for (int i = 0; i < 10; i++) {
                vector.add(i);
            }
            Thread thread1 = new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < vector.size(); i++) {
                        vector.remove(i);
                    }
                }
            });
            Thread thread2 = new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < vector.size(); i++) {
                        vector.get(i);
                    }
                }
            });
            thread1.start();
            thread2.start();
        }
    }
}

运行结果：报错java.lang.ArrayIndexOutOfBoundsException: Array index out of range

原因分析：同时发生获取与删除的操作。当两个线程在同一时间都判断了vector的size，假设都判断为9，而下一刻线程1执行了remove操作，随后线程2才去get，所以就出现了错误。synchronized关键字可以保证同一时间只有一个线程执行该方法，但是多个线程同时分别执行remove、add、get操作的时候就无法控制了。

（2）使用foreach\iterator遍历Vector的时候进行增删操作

public class VectorExample3 {
    // java.util.ConcurrentModificationException
    private static void test1(Vector<Integer> v1) { // foreach
        for(Integer i : v1) {
            if (i.equals(3)) {
                v1.remove(i);
            }
        }
    }
    // java.util.ConcurrentModificationException
    private static void test2(Vector<Integer> v1) { // iterator
        Iterator<Integer> iterator = v1.iterator();
        while (iterator.hasNext()) {
            Integer i = iterator.next();
            if (i.equals(3)) {
                v1.remove(i);
            }
        }
    }
    // success
    private static void test3(Vector<Integer> v1) { // for
        for (int i = 0; i < v1.size(); i++) {
            if (v1.get(i).equals(3)) {
                v1.remove(i);
            }
        }
    }
    public static void main(String[] args) {
        Vector<Integer> vector = new Vector<>();
        vector.add(1);
        vector.add(2);
        vector.add(3);
        test3(vector);
    }
}

2.HashMap的线程安全类：HashTable

@ThreadSafe

public class HashTableExample {
    // 请求总数
    public static int clientTotal = 5000;
    // 同时并发执行的线程数
    public static int threadTotal = 200;
    private static Map<Integer, Integer> map = new Hashtable<>();
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal; i++) {
            final int count = i;
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    update(count);
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("size:{}", map.size());
    }
    private static void update(int i) {
        map.put(i, i);
    }
}

保证安全性：使用了synchronized修饰

不允许空值（在代码中特殊做了判断）

HashMap和HashTable都使用哈希表来存储键值对。在数据结构上是基本相同的，都创建了一个继承自Map.Entry的私有的内部类Entry，每一个Entry对象表示存储在哈希表中的一个键值对。

3.Collections类中的相关同步方法

Collections类中提供了一系列的线程安全方法用于处理ArrayList等线程不安全的Collection类

（1）List:

@ThreadSafe
public class CollectionsExample1 {
    // 请求总数
    public static int clientTotal = 5000;
    // 同时并发执行的线程数
    public static int threadTotal = 200;
    private static List<Integer> list = Collections.synchronizedList(Lists.newArrayList());
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal; i++) {
            final int count = i;
              executorService.execute(() -> {
              try {
                    semaphore.acquire();
                    update(count);
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("size:{}", list.size());
    }
    private static void update(int i) {
        list.add(i);
    }
}

（2）Map:

@ThreadSafe
public class CollectionsExample3 {
    // 请求总数
    public static int clientTotal = 5000;
    // 同时并发执行的线程数
    public static int threadTotal = 200;
    private static Map<Integer, Integer> map = Collections.synchronizedMap(new HashMap<>());
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal; i++) {
            final int count = i;
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    update(count);
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("size:{}", map.size());
    }
    private static void update(int i) {
        map.put(i, i);
    }
}

（3）Set

@ThreadSafe
public class CollectionsExample2 {
    // 请求总数
    public static int clientTotal = 5000;
    // 同时并发执行的线程数
    public static int threadTotal = 200;
    private static Set<Integer> set = Collections.synchronizedSet(Sets.newHashSet());
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal; i++) {
            final int count = i;
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    update(count);
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("size:{}", set.size());
    }
    private static void update(int i) {
        set.add(i);
    }
}

多线程理解（九） 同步容器

猜你喜欢

多线程理解（九）同步容器