首页 > 编程语言> > JAVA高并发集合详解

JAVA高并发集合详解

2020-04-21 21:05:29 作者：互联网

Queue（队列）
主要是为了高并发准备的容器
Deque：双端队列，可以反方向装或者取

最开始jdk1.0只有Vector和hashtable 默认所有方法都实现了synchronized锁,线程安全但性能比较差，因此后续SUN意识到这个问题之后加了完全没加锁的hashmap，但是由于Hashmap完全没锁，SUN又想到能不能让Hashmap在有锁的时候用呢，此时添加了Collection，里面有一个Collection.synchronizedMap(new HashMap()),将Hashmap变成了加锁的版本，里面锁的粒度变小了，能部分提高性能，在JUC之后，新出来了ConcurrentHashtable由于Hashtable和Vectoy是古老的版本带过来的，所以现在基本不用，知道就行。
1、HashTable性能测试

public class Constants {
    public static final int COUNT = 1000000;
    public static final int THREAD_COUNT = 100;
}

public class T01_TestHashtable {
    static Hashtable<UUID, UUID> m = new Hashtable<>();
    static int count = Constants.COUNT;
    static UUID[] keys = new UUID[count];
    static UUID[] values = new UUID[count];
    static final int THREAD_COUNT = Constants.THREAD_COUNT;
    static {
        for (int i = 0; i < count; i++) {
            keys[i] = UUID.randomUUID();
            values[i] = UUID.randomUUID();
        }
    }
    static class MyThread extends Thread {
        int start;
        int gap = count/THREAD_COUNT;
        public MyThread(int start) {
            this.start = start;
        }
        @Override
        public void run() {
            for(int i=start; i<start+gap; i++) {
                m.put(keys[i], values[i]);
            }
        }
    }
    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        Thread[] threads = new Thread[THREAD_COUNT];
        for(int i=0; i<threads.length; i++) {
            threads[i] =
            new MyThread(i * (count/THREAD_COUNT));
        }
        for(Thread t : threads) {
            t.start();
        }
        for(Thread t : threads) {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        long end = System.currentTimeMillis();
        System.out.println(end - start);
        System.out.println(m.size());
        //-----------------------------------
        start = System.currentTimeMillis();
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(()->{
                for (int j = 0; j < 10000000; j++) {
                    m.get(keys[10]);
                }
            });
        }
        for(Thread t : threads) {
            t.start();
        }
        for(Thread t : threads) {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        end = System.currentTimeMillis();
        System.out.println(end - start);
    }
}

相对来说，插入的时候ConcurrentHashMap相对于HashMap效率没有明显提高，但是读取的时候效率高很多。

ConcurrentSkipListMap 跳表结构，由于CAS在Tree这种结构上操作太复杂，然后冒出了跳表这种结构，底层时链表，为了查找效率，提取关键元素，制作新的链表，找的时候先找11->78,发现元素比78小，则与11比较，如果大，则在11-78中找，明显效率提高了。
跳表：
多线程尽量使用Queue，少用List、Set
CopyOnWrite,写时复制，读的时候不加锁，写的时候加锁，原理是：会在原来基础上Copy一个出来数组，在新的数组上写，写完后将引用指向新数组，过程中读的时候读的是原来数组，当写的过程结束之后读得是新数组，写时复制

CopyOnWriteArrayList，写（添加）的时候是加了锁的，里面先获取原数组的长度，创建新数组长度为原数组+1，读（get）的时候是加了锁的

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();
        }
    }

    public E get(int index) {
        return get(getArray(), index);
    }

BlockingQueue 为线程池做准备，重点在blocking上，阻塞队列，有很多对多线程友好的方法。
1、常用的一共三种，分别是无界的LinkedBlockingQueue（表示用Linked实现是，无界，除非内存溢出，添加方法put的区别是，put方法非要添加，如果满了就阻塞住，取方法 take（），非要取，如果没有了就阻塞，阻塞的原理就是实现了park（），线程阻塞住，进入wait状态）、
2、有界的ArrayBlockingQueue（表示用Array实现是，有界，除非内存溢出，满了之后程序会阻塞住，等消费者）、
3、DelayQueue（实现时间上的排序）、
4、synchronusQueue（实现线程中间传输内容、任务）、
5、TransferQueue（是前面几种的组合，也可以传输任务，并且不是传递一个，可以传递多个）
ConcurrentLinkedQueue
里面一些友好的方法添加 offer（）（返回值是Boolean类型，true表示成功，）、peak（）（取）
Queue 与 List的区别，主要在于Queue添加了很多对线程友好的API，比如offer peek poll
这其中一个子类BlockingQueue在上面（offer peek poll）的基础上添加了put take ->主要是实现了阻塞，可以自然而然的实现任务队列，也就是生产者、消费者模型，这是多线程中最重要的模型，是MQ的基础（必问内容）

DelayQueue，可以实现等待时间，类实现Delayed接口，同时设置运行时间runningtime，时间等待越短的优先运行，实现Compare接口，重写方法来确定任务执行队列
按时间进行任务调度。本质是priorityQueue。

    static BlockingQueue<MyTask> tasks = new DelayQueue<>();

    static Random r = new Random();
    
    static class MyTask implements Delayed {
        String name;
        long runningTime;
        
        MyTask(String name, long rt) {
            this.name = name;
            this.runningTime = rt;
        }

        @Override
        public int compareTo(Delayed o) {
            if(this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS))
                return -1;
            else if(this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) 
                return 1;
            else 
                return 0;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            
            return unit.convert(runningTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        
        
        @Override
        public String toString() {
            return name + " " + runningTime;
        }
    }

main方法：

public static void main(String[] args) throws InterruptedException {
        long now = System.currentTimeMillis();
        MyTask t1 = new MyTask("t1", now + 1000);
        MyTask t2 = new MyTask("t2", now + 2000);
        MyTask t3 = new MyTask("t3", now + 1500);
        MyTask t4 = new MyTask("t4", now + 2500);
        MyTask t5 = new MyTask("t5", now + 500);
        
        tasks.put(t1);
        tasks.put(t2);
        tasks.put(t3);
        tasks.put(t4);
        tasks.put(t5);
        
        System.out.println(tasks);
        
        for(int i=0; i<5; i++) {
            System.out.println(tasks.take());
        }
    }

结果：

t5 1587472415617
t1 1587472416117
t3 1587472416617
t2 1587472417117
t4 1587472417617

PriorityQueue,会默认对任务排序，最小最优先，里面是小顶堆
Demo：

public class T07_01_PriorityQueque {
    public static void main(String[] args) {
        PriorityQueue<String> q = new PriorityQueue<>();
        q.add("c");
        q.add("e");
        q.add("a");
        q.add("d");
        q.add("z");
        for (int i = 0; i < 5; i++) {
            System.out.println(q.poll());
        }
    }
}
//打印结果:a、c、d、e、z

synchronusQueue，容量为0，不是用来装东西，主要是用来一个线程给另一个线程下达任务，与之前的exchanger容器类似，不可以用add方法，里面容器为0，不可以装东西。应用场景是：做某件事需要等待结果完成才继续执行接下来的任务。

标签：JAVA,int,并发,详解,static,put,new,MyTask,public
来源： https://www.cnblogs.com/Courage129/p/12746406.html