码迷,mamicode.com
首页 > 其他好文 > 详细

第五章 基础构建模块

时间:2018-08-09 01:13:32      阅读:157      评论:0      收藏:0      [点我收藏+]

标签:数组   priority   pack   使用场景   解决   获取   ash   使用   i++   

5.1 同步容器类

  实现方式 : 将他们的状态封装起来,并对每个公有方法都进行同步, 使得每次只有一个线程可以访问. 

5.1.1 存在的问题

  复合操作 并非线程安全. 比如 迭代, 条件运算等.

  在对同步容器类的复合操作加锁时一定要以容器对象为锁对象, 保证复合操作的锁对象容器使用的锁对象一致.才能实现一个线程安全的复合操作

 public static void getLast(Vector<?> list) {
        // 此处的锁对象必须和 Vector 内部的锁对象一致
        synchronized (list) {
            int lastIndex = list.size()-1;
            list.remove(lastIndex);
        }
    }

 

5.1.2 迭代器与ConcurrentModificationException

  在容器的迭代过程中被修改(结构上被改变)时会抛出  ConcurrentModificationException

  解决方法 : 在迭代过程中持有容器的锁. 并在所有对共享容器进行迭代的地方加锁

5.1.3 隐藏迭代器

  以下操作也会间接的进行容器的迭代操作

  toString() , hashCode() , equals() 等很多方法都出触发容器的迭代操作. 

5.2 并发容器

5.2.1 ConcurrentHashMap

  •   加锁策略 : 分段锁(粒度更细的加锁机制), 同步方法块
  •   支持多个线程并发的访问  ConcurrentHashMap , 实现更高的吞吐量 .  
  •   ConcurrentHashMap 返回的迭代器具有弱一致性 , 可以(但是不保证)将修改操作立即反映给容器
  •   迭代过程中不会加锁 , 也不会抛出 ConcurrentModificationException , 
  •   将一些复合操作(putIfAbsent() 若没有则添加) 实现为原子操作

5.2.3 CopyOnWriteArrayList

  • 每次修改容器时先复制数组, 引用依旧指向原来的数组 , 然后修改新的数组, 最后将引用指向新的数组.
  • 写入时复制 也可理解为 修改时复制
  • 返回的迭代器不会抛出  ConcurrentModificationException 
  • 使用场景 : 迭代操作远远多于修改操作. 复制数组的操作有一定的开销.
  • 修改操作使用 ReentrantLock 进行加锁

5.3 阻塞队列

  • 提供阻塞的 put 和 take 方法
  • put 方法将阻塞到直到有空间可用 , take 方法将阻塞到直到有元素可用
  • 队列可以有界, 也可以无界
  • 修改容器时统一使用创建队列实例时创建的 ReentrantLock 对象

  BlockingQueue(阻塞队列接口)

    LinkedBlockingQueue 类似与 LinkedList

    ArrayBlockingQueue 类似与 ArrayList

    PriorityBlockingQueue   按优先级排序的队列

    SynchronousQueue 每个插入操作必须等待另一个线程的对应移除操作 ,反之亦然。非常适合做交换工作,生产者的线程和消费者的线程同步以传递某些信息、事件或者任务。

 5.3.1 生产者与消费者的特点

  • 生产者和消费者只需要完成各自的任务
  • 阻塞队列将负责所有的控制流
  • 每个功能的代码和逻辑更加清楚
  • 他们整体的并行度取决于两者中较低的并行度

  生产者和消费者设计也可以使用 Executor 任务执行框架来实现, 其本身也使用 生产者--消费者模式

5.3.2 串行线程封闭 

  • 线程封闭对象只能由单个对象拥有,可以通过安全的发该对象来转移所有权.并且发布对象的线程不会再访问它
  • 转移所有权之后,只有另外一个线程获得这个对象的访问权限, 它可以对它做任意的修改,因为它有独占访问权.

5.3.3 双端队列和工作密取

  双端队列

  • ArrayDeque 和 LinkedBlockingDeque
  • 实现在队列头和队列尾的高效插入和移除.

  工作密取

  • 每个消费者有自己的双端队列 , 当一个消费者完成自己队列的所有任务后 , 那么它可以从其他消费者的双端队列秘密的获取任务 .

5.4 阻塞方法和中断方法

  处理 InterruptedException 

  • 将 InterruptedException  传递给方法的调用者
  • 捕获这个异常, 并恢复中断状态

5.5 同步工具类

5.5.1 闭锁(CountDownLatch)

  作用 : 用来确保某些活动直到其他活动都完成后才执行.

  • 计数器 : 表示需要等待的事件数量
  • await() : 阻塞调用此方法的线程直到计数器为0
  • countDown() : 计数器递减

用法一 : 创建一定数量的线程,多个线程并发的执行任务

  • 使用两个闭锁, 分别表示起始门和结束门. 起始门初始值为1 , 结束门初始值为等待的事件数量. 
  • 每个工作线程在起始门等待,
  • 所有线程就绪后起始门调用 countDown()直到起始门的值为0, 之后所有线程同时开始执行(因为线程之前都在起始门等待)
  • 主线程调用结束门的await(), 主线程阻塞直到所有工作线程结束
  • 工作线程全部执行完毕, 主线程运行
package com.pinnet.test;

import java.util.concurrent.CountDownLatch;

public class CountLatchTest {

    public void timeTask(int threadNumbers, Runnable task) throws InterruptedException {
        CountDownLatch start = new CountDownLatch(1);
        CountDownLatch end = new CountDownLatch(threadNumbers);

        for (int i = 0; i < threadNumbers; i++) {
            new Thread() {
                public void run() {
                    try {
                        // 所有线程在起始门等待
                        start.await();
                        // 执行任务
                        task.run();
                        // 结束门递减
                        end.countDown();
                    } catch (InterruptedException e) {

                    }
                }
            }.start();
        }
        // 所有工作线程开始执行
        start.countDown();
        // 所有工作线程启动后主线程立即登待
        end.await();
        System.out.println("开始主线程");

    }

}

用法二: 创建一定数量的线程,多个线程依次的执行任务

  • 使用一个闭锁
  • 线程依次启动, 执行完成后countDown() 
  • 主线程 await() , 直到计数器为0 ,主线程执行
public void timeTask2(int threadNumbers, Runnable task) throws InterruptedException {
        CountDownLatch start = new CountDownLatch(threadNumbers);
        // 任务一次执行
        for (int i = 0; i < threadNumbers; i++) {
            new Thread() {
                public void run() {
                        // 任务开始
                        task.run();
                        // 递减
                        start.countDown();
                }
            }.start();
        }
        // 主线程阻塞直到计数器为0
        start.await();
        System.out.println("开始主线程");

    }

5.5.2 FutureTask

  一种可生成结果,可异步取消的计算.

  • 计算通过 Callable 实现
  • 可以将计算结果从执行任务的线程 传递 到获取这个结果的线程 , 并保证其安全性
  • get() 获取结果可阻塞到任务完成
public class FutureTaskTest {
    // 创建任务
    private final FutureTask<Integer> future = new FutureTask<>(new Callable<Integer>() {
        public Integer call() {
            return 123;
        }
    });
    // 创建线程
    private final Thread thread = new Thread(future);
    // 对外提供方法启动线程
    public void start() {
        thread.start();
    }
    // 获取计算结果
    public Integer get() {
        try {
            return future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            return null;
        }
    }
}

5.5.3 信号量(Semaphore)

  • 控制同时访问某个资源或执行某个操作的数量.
  • Semaphore 管理一定数量的许可, 执行操作时先获取许可,执行完操作后释放许可.
  • 获取许可时可阻塞.

5.5.4 栅栏

  闭锁和栅栏的区别 :

  • 闭锁是一次性对象 , 一旦进入终止状态 , 就不能被重置
  • 栅栏是所有线程必须同时到达栅栏位置 , 才能继续执行.
  • 闭锁用于等待事件 , 栅栏用于等待线程.

  基本使用 : 

  • 指定数量线程到达栅栏位置后,所有线程被释放 . 栅栏将被重置 
  • 成功通过栅栏,await()将会返回一个到达索引号
  • await() 超时 或 await()阻塞的线程被中断 , 则栅栏被打破.所有阻塞的await() 调用被终止 , 并抛出 BrokenBarrierException.

  用法一 : CyclicBarrier(int number)   

      await() 调用 number 次后所有调用 await() 的线程继续执行 , 否则线程在await() 阻塞

public void barrier() {
        int number = 6;
        // 参数表示屏障拦截的线程数量 
        // barrier.await() 调用 number 次后所有线程的阻塞状态解除
        CyclicBarrier barrier = new CyclicBarrier(number);
        
        for (int i = 0; i < number; i++) {
            new Thread(new Runnable() {
                
                @Override
                public void run() {
                    System.out.println("此线程任务已经完成");
                    try {
                        // 调用await方法告诉CyclicBarrier我已经到达了屏障,然后当前线程被阻塞。
                        barrier.await();
                        System.out.println("所有线程执行完成");
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

用法二 : CyclicBarrier(int parties, Runnable runnable) 指定数量的线程到达屏障点后 执行  runnable

 public void barrier2() {
        int number = 6;
        // 参数表示屏障拦截的线程数量 
        // barrier.await() 调用 number 次后所有线程的阻塞状态解除
        CyclicBarrier barrier = new CyclicBarrier(number, new Runnable() {
            
            @Override
            public void run() {
                //  指定数量的线程到达屏障点后执行  barrier()
                barrier();
            }
        });
        
        for (int i = 0; i < number; i++) {
            new Thread(new Runnable() {
                
                @Override
                public void run() {
                    System.out.println("此线程任务已经完成");
                    try {
                        // 调用await方法告诉CyclicBarrier我已经到达了屏障,然后当前线程被阻塞。
                        barrier.await();
                        System.out.println("所有线程执行完成");
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

双方形式的栅栏 Exchanger 

 

第五章 基础构建模块

标签:数组   priority   pack   使用场景   解决   获取   ash   使用   i++   

原文地址:https://www.cnblogs.com/virgosnail/p/9446516.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!