美文网首页
java并发包（第四天）

java并发包（第四天）

作者: hzhang94 | 来源:发表于2018-09-27 15:25 被阅读0次

同步容器类

Vector与ArrayList区别

  1. ArrayList是最常用的List实现类，内部是通过数组实现的，它允许对元素进行快速随机访问。数组的缺点是每个元素之间不能有间隔，当数组大小不满足时需要增加存储能力，就要讲已经有数组的数据复制到新的存储空间中。当从ArrayList的中间位置插入或者删除元素时，需要对数组进行复制、移动、代价比较高。因此，它适合随机查找和遍历，不适合插入和删除。
  2. Vector与ArrayList一样，也是通过数组实现的，不同的是它支持线程的同步，即某一时刻只有一个线程能够写Vector，避免多线程同时写而引起的不一致性，但实现同步需要很高的花费，因此，访问它比访问ArrayList慢。

Vector源码

public synchronized boolean add(E e) {
    modCount++;
    ensureCapacityHelper(elementCount + 1);
    elementData[elementCount++] = e;
    return true;
}

ArrayList源码

public boolean add(E e) {
    ensureCapacityInternal(size + 1);  // Increments modCount!!
    elementData[size++] = e;
    return true;
}

HasTable与HasMap的区别

  1. HashMap不是线程安全的，HastMap是一个接口 是map接口的子接口，是将键映射到值的对象，其中键和值都是对象，并且不能包含重复键，但可以包含重复值。HashMap允许null key和null value，而hashtable不允许。
  2. HashTable是线程安全的一个Collection。
  3. HashMap是Hashtable的轻量级实现（非线程安全的实现），他们都完成了Map接口，主要区别在于HashMap允许空（null）键值（key）,由于非线程安全，效率上可能高于Hashtable。
  4. HashMap允许将null作为一个entry的key或者value，而Hashtable不允许。
  5. HashMap把Hashtable的contains方法去掉了，改成containsvalue和containsKey。

Collections工具类

  1. 使用Collections.synchronizedMap(map);等方法将Collection里面的非线程安全的类转换为线程安全的。里面其实是使用synchronized包装起来。
  2. 源代码
        public int size() {
            synchronized (mutex) {return m.size();}
        }
        public boolean isEmpty() {
            synchronized (mutex) {return m.isEmpty();}
        }
        public boolean containsKey(Object key) {
            synchronized (mutex) {return m.containsKey(key);}
        }
        public boolean containsValue(Object value) {
            synchronized (mutex) {return m.containsValue(value);}
        }
        public V get(Object key) {
            synchronized (mutex) {return m.get(key);}
        }

        public V put(K key, V value) {
            synchronized (mutex) {return m.put(key, value);}
        }
        public V remove(Object key) {
            synchronized (mutex) {return m.remove(key);}
        }
        public void putAll(Map<? extends K, ? extends V> map) {
            synchronized (mutex) {m.putAll(map);}
        }
        public void clear() {
            synchronized (mutex) {m.clear();}
        }

ConcurrentHashMap

ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分，每个段其实就是一个小的HashTable,它们有自己的锁。只要多个修改操作发生在不同的段上，它们就可以并发进行。把一个整体分成了16个段(Segment)也就是最高支持16个线程的并发修改操作。
这也是在多线程场景时减小锁的粒度从而降低锁竞争的一种方案。并且代码中大多共享变量使用volatile关键字声明，目的是第一时间获取修改的内容，性能非常好。

CountDownLatch

CountDownLatch类位于java.util.concurrent包下，利用它可以实现类似计数器的功能。比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了。

package top.nightliar.study.day04;

import java.util.concurrent.CountDownLatch;

/**
 * Created by Nightliar
 * 2018-09-17 13:49
 */
public class ThreadDemo02 {

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(2);
        System.out.println("主线程开始");
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("子线程1开始");
                countDownLatch.countDown();  // 每次减去1
                System.out.println("子线程1结束");
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("子线程2开始");
                countDownLatch.countDown();  // 每次减去1
                System.out.println("子线程2结束");
            }
        }).start();
        countDownLatch.await();  // 调用当前方法主线程阻塞  countDown结果为0, 阻塞变为运行状态
        System.out.println("两个子线程执行完毕");
        System.out.println("主线程结束");
    }

}

结果：
主线程开始
子线程2开始
子线程1开始
子线程2结束
子线程1结束
两个子线程执行完毕
主线程结束

CyclicBarrier

CyclicBarrier初始化时规定一个数目，然后计算调用了CyclicBarrier.await()进入等待的线程数。当线程数达到了这个数目时，所有进入等待状态的线程被唤醒并继续。
CyclicBarrier就象它名字的意思一样，可看成是个障碍， 所有的线程必须到齐后才能一起通过这个障碍。
CyclicBarrier初始时还可带一个Runnable的参数， 此Runnable任务在CyclicBarrier的数目达到后，所有其它线程被唤醒前被执行。

package top.nightliar.study.day04;

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

/**
 * Created by Nightliar
 * 2018-09-17 13:59
 */
public class ThreadDemo03 {

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(4);
        for (int i = 0; i < 4; i++) {
            new CreateThread(cyclicBarrier).start();
        }
    }

}

class CreateThread extends Thread{

    CyclicBarrier cyclicBarrier;

    public CreateThread(CyclicBarrier cyclicBarrier) {
        this.cyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        System.out.println(getName() + "写入数据开始...");
        try {
            sleep(40);
            System.out.println(getName() + "写入数据完毕");
            cyclicBarrier.await();
            System.out.println(getName() + "所有线程一起执行完毕");
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

}

结果：
Thread-0写入数据开始...
Thread-3写入数据开始...
Thread-2写入数据开始...
Thread-1写入数据开始...
Thread-2写入数据完毕
Thread-3写入数据完毕
Thread-0写入数据完毕
Thread-1写入数据完毕
Thread-2所有线程一起执行完毕
Thread-3所有线程一起执行完毕
Thread-0所有线程一起执行完毕
Thread-1所有线程一起执行完毕

Semaphore

Semaphore是一种基于计数的信号量。它可以设定一个阈值，基于此，多个线程竞争获取许可信号，做自己的申请后归还，超过阈值后，线程申请许可信号将会被阻塞。Semaphore可以用来构建一些对象池，资源池之类的，比如数据库连接池，我们也可以创建计数为1的Semaphore，将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态。
基础方法：

  1. availablePermits(); 函数用来获取当前可用的资源数量。
  2. acquire(); 申请资源。
  3. release(); 释放资源。
package top.nightliar.study.day04;

import java.util.Random;
import java.util.concurrent.Semaphore;

/**
 * 10个线程同时运行，但只能同时运行其中三个。
 * Created by Nightliar
 * 2018-09-17 14:32
 */
public class ThreadDemo04 {

    public static void main(String[] args) {
        // 设置3个并发量
        Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < 11; i++) {
            new CreateThread02("第" + i + "个人，", semaphore).start();
        }
    }

}

class CreateThread02 extends Thread {

    Semaphore semaphore;
    String name;

    public CreateThread02(String name, Semaphore semaphore) {
        this.semaphore = semaphore;
        this.name = name;
    }

    @Override
    public void run() {
        // 获取当前可用信号量
        int availablePermits = semaphore.availablePermits();
        System.out.println("当前可用信号量为：" + availablePermits);
        try {
            // 申请获取资源 如果资源达到3次，就等待
            semaphore.acquire();
            System.out.println(name + "终于抢到资源了！！");
            Thread.sleep(new Random().nextInt(1000)); // 模拟操作资源所需的时间。
            System.out.println(name + "操作资源结束！！");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 释放资源
            semaphore.release();
        }
    }
}

结果：
当前可用信号量为：3
当前可用信号量为：3
第0个人，终于抢到资源了！！
第1个人，终于抢到资源了！！
当前可用信号量为：1
当前可用信号量为：1
第5个人，终于抢到资源了！！
当前可用信号量为：0
当前可用信号量为：0
当前可用信号量为：0
当前可用信号量为：0
当前可用信号量为：0
当前可用信号量为：0
当前可用信号量为：0
第0个人，操作资源结束！！
第4个人，终于抢到资源了！！
第1个人，操作资源结束！！
第7个人，终于抢到资源了！！
第7个人，操作资源结束！！
第3个人，终于抢到资源了！！
第5个人，操作资源结束！！
第2个人，终于抢到资源了！！
第2个人，操作资源结束！！
第8个人，终于抢到资源了！！
第3个人，操作资源结束！！
第10个人，终于抢到资源了！！
第4个人，操作资源结束！！
第9个人，终于抢到资源了！！
第8个人，操作资源结束！！
第6个人，终于抢到资源了！！
第10个人，操作资源结束！！
第6个人，操作资源结束！！
第9个人，操作资源结束！！

并发队列

ConcurrentLinkedDeque（非阻塞）

ConcurrentLinkedQueue : 是一个适用于高并发场景下的队列，通过无锁的方式，实现了高并发状态下的高性能，通常ConcurrentLinkedQueue性能好BlockingQueue。它是一个非阻塞，无锁，无界，线程安全双端操作的队列 。该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，该队列不允许null元素。
简单说就是ConcurrentLinkedQueue的升级版，在JDK7之后才提供。该队列也不允许空元素，而且size方法并不是常量时间，其需要遍历链表，此时并发修改链表会造成统计size不正确。同样，bulk操作和equal以及toArray方法不保证原子性。

ConcurrentLinkedDeque<String> deque = new ConcurrentLinkedDeque<>();
deque.offer("张三");
deque.offer("李四");
deque.offer("王五");
//从头获取元素,删除该元素
System.out.println(deque.poll());
//从头获取元素,不刪除该元素
System.out.println(deque.peek());
//获取总长度
System.out.println(deque.size());

BlockingQueue（阻塞）

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：

  1. 在队列为空时，获取元素的线程会等待队列变为非空。
  2. 当队列满时，存储元素的线程会等待队列可用。
    java.util.concurrent包提供了几种不同形式的阻塞队列，如数组阻塞队列ArrayBlockingQueue、链表阻塞队列LinkedBlockingQueue、优先级阻塞队列PriorityBlockingQueue和延时队列DelayQueue等。

ArrayBlockingQueue

ArrayBlockingQueue是一个有边界的阻塞队列，它的内部实现是一个数组。有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变。
ArrayBlockingQueue是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

ArrayBlockingQueue<String> arrays = new ArrayBlockingQueue<String>(3);
arrays.add("李四");
arrays.add("张军");
arrays.add("张军");
// 添加阻塞队列
arrays.offer("张三", 1, TimeUnit.SECONDS);

LinkedBlockingQueue

LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX_VALUE的容量 。它的内部实现是一个链表。
和ArrayBlockingQueue一样，LinkedBlockingQueue 也是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(3);
linkedBlockingQueue.add("张三");
linkedBlockingQueue.add("李四");
linkedBlockingQueue.add("李四");
System.out.println(linkedBlockingQueue.size());

PriorityBlockingQueue

PriorityBlockingQueue是一个没有边界的队列，排序规则和java.util.PriorityQueue一样。需要注意，PriorityBlockingQueue中允许插入null对象。所有插入PriorityBlockingQueue的对象必须实现 java.lang.Comparable接口，队列优先级的排序规则就 是按照我们对这个接口的实现来定义的。
另外，我们可以从PriorityBlockingQueue获得一个迭代器Iterator，但这个迭代器并不保证按照优先级顺 序进行迭代。

SynchronousQueue

SynchronousQueue是一个没有数据缓冲的BlockingQueue，生产者线程对其的插入操作put必须等待消费者的移除操作take，反过来也一样。

package top.nightliar.study.day04;

import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.SynchronousQueue;

/**
 * Created by Nightliar
 * 2018-09-17 15:45
 */
public class ThreadDemo05 {

    public static void main(String[] args) throws InterruptedException {
        final SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>();

        Thread product = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("生产者线程start");
                try {
                    queue.put(1);
                } catch (InterruptedException e) {
                }
                System.out.println("生产者线程end");
            }
        });

        Thread customer = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("消费者线程start");
                try {
                    System.out.println("消费的数据: " + queue.take());
                } catch (InterruptedException e) {
                }
                System.out.println("消费者线程end");
            }
        });
        // 先启动消费者，消费者会等待生产者生产数据才会去消费
        customer.start();
        Thread.sleep(1000);
        product.start();
    }
}

使用BlockingQueue模拟生产者与消费者

package top.nightliar.study.day04;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 使用BlockingQueue模拟生产者与消费者
 * Created by Nightliar
 * 2018-09-17 16:26
 */
public class ThreadDemo06 {

    public static void main(String[] args) throws InterruptedException {
        BlockingDeque<String> deque = new LinkedBlockingDeque<>(10);
        Product p1 = new Product(deque);
        Product p2 = new Product(deque);
        Consumer c1 = new Consumer(deque);
        c1.start();
        p1.start();
        p2.start();
        // 执行10s
        Thread.sleep(10 * 1000);
        p1.interrupt();
        p2.interrupt();

    }

}

/**
 * 生产者
 */
class Product extends Thread {

    private BlockingDeque<String> deque;
    private volatile boolean flag = true;
    private static AtomicInteger count = new AtomicInteger();

    public Product(BlockingDeque<String> deque) {
        this.deque = deque;
    }

    @Override
    public void run() {
        System.out.println("生成者开始工作");
        try {
            while (true){
                String data = count.incrementAndGet()+"";
                // 将数据存入队列中
                boolean offer = deque.offer(data, 2, TimeUnit.SECONDS);
                if (offer) {
                    System.out.println("生产者,存入" + data + "到队列中,成功.");
                } else {
                    System.out.println("生产者,存入" + data + "到队列中,失败.");
                }
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            this.flag = false;
        }
    }
}

/**
 * 消费者
 */
class Consumer extends Thread {

    private BlockingDeque<String> deque;
    private volatile boolean flag = true;
    private static AtomicInteger count = new AtomicInteger();

    public Consumer(BlockingDeque<String> deque) {
        this.deque = deque;
    }

    @Override
    public void run() {
        System.out.println("消费线程启动...");
        try {
            while (flag) {
                String data = deque.poll(2, TimeUnit.SECONDS);
                if (data != null) {
                    System.out.println("消费者,拿到队列中的数据data:" + data);
                    Thread.sleep(1000);
                } else {
                    System.out.println("消费者,超过2秒未获取到数据..");
                    flag = false;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者退出线程...");
        }

    }
}

相关文章

网友评论

      本文标题：java并发包（第四天）

      本文链接：https://www.haomeiwen.com/subject/fiveoftx.html

      延伸阅读

      深度阅读

      • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

      栏目导航

      热点阅读

      关于我们|服务条款|联系我们|java并发包（第四天）|投稿指南|网站地图|RSS订阅|排版工具|手机版

      提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

      Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

      备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

      本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

      所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！