@[toc]

不安全示例

package cn.wyj.learn.juc;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

@Slf4j
public class CollectionListExample2ThreadSafe {

    private static Collection list = new ArrayList();
    //运行次数
    private static final int RUN_TOTAL = 5000;
    // 并发控制
    private static final int THREAD_TOTAL = 200;


    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Semaphore semaphore = new Semaphore(THREAD_TOTAL);
        for (int i = 0; i < RUN_TOTAL; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    list.add(1);
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
        // 需要等待成执行完
        executorService.awaitTermination(10, TimeUnit.SECONDS);
        log.info("list Size:{}", list.size());
    }


}

运行结果 1

21:06:36.240 [main] INFO cn.wyj.learn.juc.CollectionListExample1NoThreadSafe - list Size:4929

运行结果 2

Exception in thread "pool-1-thread-9" java.lang.ArrayIndexOutOfBoundsException: 366
    at java.util.ArrayList.add(ArrayList.java:459)
    at cn.wyj.learn.juc.CollectionListExample1NoThreadSafe.lambda$main$0(CollectionListExample1NoThreadSafe.java:28)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

安全写法1

    private static Collection list = CollectionUtils.synchronizedCollection(new ArrayList());

原因 .对原有api做了包裹处理

  /**
     * Constructor that wraps (not copies).
     * 
     * @param collection  the collection to decorate, must not be null
     * @throws IllegalArgumentException if the collection is null
     */
    protected SynchronizedCollection(Collection collection) {
        if (collection == null) {
            throw new IllegalArgumentException("Collection must not be null");
        }
        this.collection = collection;
        this.lock = this;
    }
    

    public boolean add(Object object) {
    synchronized (lock) {
        return collection.add(object);
    }
}

安全写法2

    private static Collection list = new Vector();

原因,Vector 容器本身就是安全的

 /**
     * Appends the specified element to the end of this Vector.
     *
     * @param e element to be appended to this Vector
     * @return {@code true} (as specified by {@link Collection#add})
     * @since 1.2
     */
    public synchronized boolean add(E e) {
        modCount++;
        ensureCapacityHelper(elementCount + 1);
        elementData[elementCount++] = e;
        return true;
    }

安全写法3

    private static Collection list = new CopyOnWriteArrayList();

@since 1.5

Copy-On-Write简称COW，是一种用于程序设计中的优化策略。其基本思路是，从一开始大家都在共享同一个内容，当某个人想要修改这个内容的时候，才会真正把内容Copy出去形成一个新的内容然后再改，这是一种延时懒惰策略。

容器声明

    private transient volatile Object[] array;

写

 /**
    * Appends the specified element to the end of this list.
    *
    * @param e element to be appended to this list
    * @return {@code true} (as specified by {@link Collection#add})
    */
   public boolean add(E e) {
       final ReentrantLock lock = this.lock;
       lock.lock();
       try {
           Object[] elements = getArray();
           int len = elements.length;
           Object[] newElements = Arrays.copyOf(elements, len + 1);
           newElements[len] = e;
           setArray(newElements);
           return true;
       } finally {
           lock.unlock();
       }
   }

读

 @SuppressWarnings("unchecked")
   private E get(Object[] a, int index) {
       return (E) a[index];
   }

   /**
    * {@inheritDoc}
    *
    * @throws IndexOutOfBoundsException {@inheritDoc}
    */
   public E get(int index) {
       return get(getArray(), index);
   }

image.png

安全写法4

自己实现并发控制,synchronized

package cn.wyj.learn.juc;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.*;

@Slf4j
public class CollectionListExample5ThreadSafe {

   private static Collection list = new ArrayList();
   //运行次数
   private static final int RUN_TOTAL = 5000;
   // 并发控制
   private static final int THREAD_TOTAL = 200;


   public static void main(String[] args) throws InterruptedException {
       ExecutorService executorService = Executors.newCachedThreadPool();
       Semaphore semaphore = new Semaphore(THREAD_TOTAL);
       for (int i = 0; i < RUN_TOTAL; i++) {
           executorService.execute(() -> {
               try {
                   semaphore.acquire();
                   add();
                   semaphore.release();
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           });
       }
       executorService.shutdown();
       // 需要等待成执行完
       executorService.awaitTermination(10, TimeUnit.SECONDS);
       log.info("list Size:{}", list.size());
   }

   private static synchronized void add() {
       list.add(1);
   }


}

安全写法5

自己实现并发控制, 重入锁

package cn.wyj.learn.juc;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class CollectionListExample6ThreadSafe {

    private static Collection list = new ArrayList();
    //运行次数
    private static final int RUN_TOTAL = 5000;
    // 并发控制
    private static final int THREAD_TOTAL = 200;

    static Lock lock = new ReentrantLock();

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Semaphore semaphore = new Semaphore(THREAD_TOTAL);
        for (int i = 0; i < RUN_TOTAL; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    add();
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
        // 需要等待成执行完
        executorService.awaitTermination(10, TimeUnit.SECONDS);
        log.info("list Size:{}", list.size());
    }

    private static void add() {
        lock.lock();
        try {
            list.add(1);
        } finally {
            lock.unlock();
        }
    }


}