Java并发编程实战第二章 线程安全

编写线程安全代码的核心是管理对状态的访问，尤其共享和可变状态的访问。通过同步协调对对象可变状态的访问可以实现一个对象的线程安全。同步最基本的机制是synchronized关键词，但是同步(synchronization)这个属于还包括:volatile变量、显式(explicit)锁和原子(atomic)变量。

最佳实践

1. 在设计时就考虑线程安全
2. 利用封装(encapsulation)，不可变性(immutablility)，清晰指定不可变量(invariants)是比较好的实践。

2.1 什么是线程安全

1. 所有线程安全合理定义的核心都是正确性。正确性 意味着类符合其规范，一个良好的规范定义了约束对象状态的不变量和描述操作结果的后置条件。

   线程安全的定义: 如果一个类在多个线程访问时正确运行，则它是线程安全的，不管运行时环境如何调度或者交错这些线程的执行，并且在调用代码部分没有多余的同步和协调。

2. 线程安全类会封装任何需要的同步，从而使客户不必在自己实现。
3. 无状态(stateless)的类永远是线程安全的

2.2 原子性

2.2.1 Race Condition

一个竞争条件会在计算正确性依赖于运行时中多线程相对时间或者交错出现。换句话说，正确答案取决于幸运时机。大部分的竞争条件的类型为check-then-act型，此时可能会有陈旧观察用于决定下一步做什么。

例子：

1. 自增操作是一个read-modify-write操作，分三步完成而非一步，所以是非原子操作。
2. 星巴克约会的例子
3. lazy initialization的例子

原子操作

如果从一个执行A复合操作的线程的角度来看，执行B复合操作的线程要么将B所有操作执行完，要么完全不执行B系列的任何操作，那么，A和B是对彼此来说是原子的。一个原子操作是指与包括自己在内的所有在同一状态上的操作集都是互为原子操作的操作集。

原子变量

在java.util.concurrent.atomic包下有很多原子变量，可以执行原子操作。如AtomicLong的实例方法incrementAndGet()。对于无状态的类添加一个线程安全的属性，则这个类仍然是线程安全的，但是如果添加的多余一个，则不保证这个类依然是线程安全的类。

在实践中，要尽量使用线程安全对象来管理类的状态。

2.3 加锁

Intrisic Lock(固有锁)

synchronized是Java内置固有锁，即可以修饰方法，又可以修饰代码块。修饰方法时默认由方法所属对象作为intrisic locks或者monitor locks

可重入性(Reentrancy)

1. 可重入意味着锁被以线程而非以调用为单位获取。
2. 可重入通过把锁与获取计数器和其所属线程关联来实现:计数器为0时锁是unheld状态的，当一个线程获取一个unheld状态的锁后，计数器加1，如果此线程的再一次(如调用了另一个相同锁的方法)调用，则计数器加1，而非被阻塞；当线程离开synchronized块之后,计数器减1.

2.4 使用锁保护状态

1. 每个共享和可变的状态变量都应该被一个且仅一个锁保护，从而使维护人员知道是哪一个锁。
2. 给一个对象内的所有可变量加锁是一个惯例。但是这个加锁协议很容易因为添加一个新的方法或者code path被破坏。比如一个单线程程序引入一个异步的TimerTask
3. 对每个包含多个变量的不变性条件(invariants)，其中涉及的所有变量都需要由同一个锁保护。
4. 即使对象中的所有方法都是原子的，也不能保证复合操作是原子的，还需要额外的加锁机制。此外每个方法都作为同步方法还容易引起活跃性(Liveness) 问题和 性能问题。

2.5 活跃性和性能问题

1. 应该在保证原子性的情况下缩小同步的范围。
2. 占有锁的时间过长可能会导致活跃性问题和性能问题
3. 避免在长时计算或者不能快速结束如网络或者控制台IO等操作时保持锁。

Item 7: 消除过期对象引用

产生过期(obsolete)对象的主要原因:

1. 类自己管理内存
2. 缓存处理不当
3. 监听和其他回调

类自己管理内存

场景

例如实现一个Stack，使用数组维护栈内内容，当栈增加元素时是没问题的，但是当栈执行pop()时，如果只返回当前元素，而不将数组当前元素指向null，则此对象因一直被引用而不能被回收。

解决方案

1. 手动将不用的对象引用指向null。注意:这样做应该是例外而非惯例
2. 让包含对象引用的变量尽快脱离其作用域才是更好的实践，如将变量定义在最小的作用域里。

缓存处理不当

场景

当将对象放到缓存中，随着时间推移，缓存中的对象已经无效后未被正确清除就会造成内存泄漏。

解决方案

1. 使用WeakHashMap存储缓存，当key不再被引用的时候，GC会自动把这个entry清理掉
2. 对于那些使用时间不确定的缓存条目，可以是用ScheduledThreadPoolExecutor在后台定时清理
3. 也可以当做添加新缓存条目时的副作用，比如通过重写LinkedHashMap.removeEldestEntry()方法，在添加新元素时该方法如果返回true清除旧的元素
4. 更复杂的需求则需要是用java.lang.ref了

监听和其他回调

场景

当你的API有回调函数被Client注册之后，Client如果没有显式解除注册就会造成内存泄漏

方案

1. 将这些回调用WeakHashMap的key保存

实践建议

1. 提前预防，编码时就认真审核这些问题。
2. 其次是审核代码和使用heap profiler来分析

1 线程组和未处理异常

较少使用

• 线程组可以批量设置管理线程，如果创建的线程不指定分组则为默认线程组，默认情况下子线程和创建它的父线程属于同一个分组。
• 可以在线程或者线程组上设置Thread.UncaughtExceptionHandler,在线程执行过程中如果抛出一个未处理的异常，JVM在结束该线程前会自动查找是否有对应的实现了Thread.UncaughtExceptionHandler接口的对象，如果找到则执行该对象的UncaughtException方法

2 Callable和Future

特点如下：

1. 实现Callable接口后，可以通过call()抛出异常和返回结果
2. 实现Future接口的FutureTask可以接受Callable的返回值.

3 线程池

ExecutorService pool = Executors.newFixedThreadPool(6); pool.submit(new RunnableThread());//支持Runnable和Future pool.shutdown();

4 线程相关类

4.1 ThreadLocal

1. 要点: 同步机制是为了使所有线程能够共享对象，而ThreadLocal是为了隔离对象，即每个线程都修改该对象的副本而不会影响到其他线程。
2. 使用: private ThreadLocal<String> localVar= new ThreadLocal<String>()

4.2 包装线程不安全的类

Collections提供了以下几个接口来将线程不安全的集合转为线程安全的:

• synchronizedCollection() 返回线程安全的collection
• synchronizedSet() 返回线程安全的Set
• synchronizedMap() 返回线程安全的Map
• synchronizedSortedSet() 返回线程安全的SortedSet
• synchronizedSortedMap() 返回线程安全的SortedMap
• synchronizedList() 返回线程安全的List

4.3 线程安全的集合类

java.util.concurrent包下的:

• ConcurrentHashMap
• ConcurrentLinkedDeque
• ConcurrentLinkedQueue
  以上几个类都支持并发读写，在读时不会锁定。创建迭代器之后，对其修改是不能在遍历中反应出来的，而且不会报错,而java.util包下的集合类是会报ConcurrentModificationException的异常的

进程和线程

一个进程可以有多个线程，线程有自己的堆栈，自己的程序计数器和局部变量，但是他们共享进程的
系统资源。因此要小心线程之间彼此影响。

并发编程三个特点：

• 原子性问题
• 可见性问题
• 有序性问题

多线程优点

• 进程间内存不能共享，而线程间可以共享内存
• 新建进程需要分配系统资源，而新建线程则不需要分配，创建和切换代价小，从而使多线程来实现
  多任务比多进程效率更高。

java线程的创建

1. 使用Thread()创建线程，使用start()方法来启动线程。
2. 也可以通过实现Runnable接口，然后作为Thread的target创建线程。
3. Runnable接口方式的优点是可以继承其他类。

线程的周期

1. 创建。想让新创建的子线程立即执行可以使用Thread.sleep(1)，让主线程暂停一下。
2. 就绪
3. 运行
4. 阻塞
5. 死亡

tip

1. 注意少使用suspend()和stop()，这样会造成死锁。
2. 使用isAlive()判断

线程的控制

1. join线程
   在执行的线程A运行时调用其他线程B的B.join()方法后，A线程将等到B线程结束之后才执行。
2. 后台进程
   在start()之前，调用setDaemon()。可以使用isDaemon()判断是否是后台线程，后台线程
   创建的线程默认是后台线程。程序中的所有前台线程运行结束后，后台线程会被通知死亡，不过这会
   有一定的时间。
3. yield()静态方法
   与sleep()类似，但是不会让位于优先级比自己低的线程，所以如果其优先级最高且没有同级的
   线程的话，调用此方法后该线程依然继续执行。
4. 设置线程优先级
   setPriority(int)接受1～10的整数，或者三个静态常量: MAX_PRIORITY,MIN_PRIORITY,NORM_PRIORITY,对应的数值依次为10,1,5.
   虽然Java提供了1～10共10个级别，但是有的系统并不支持，所以为了移植性，最佳实践是
   使用常量而为直接指定数值

线程同步

同步代码块

eg. synchronized(obj),其中obj为同步监视器，Java允许使用任何对象作为同步监视器，
但是由于我们的目的是阻止两条线程并发访问一个共享资源，所以最佳实践是使用可能被并发访问
的资源作为同步监视器。

同步方法

此时不需要指定同步监视器，方法会自动把this当成同步监视器。通过同步方法，一般可以使类
成为线程安全的类，这样的类有如下特点：

1. 该类的对象可以被多个线程同时访问
2. 访问该对象的任意方法都能得到正确的结果
3. 访问该对象的任意方法，该对象都能依然保持合理状态

可变类的线程安全是以降低程序效率为代价的，为了降低对性能的影响，需要采取以下策略:

1. 减少同步的使用，只对会改变竞争资源的方法同步
2. 如果可变类有两个运行环境:多线程环境和单线程环境，则可以开发线程安全和非线程安全两个
   版本

释放同步监视器的锁定

程序无法显示释放对同步监视器的锁定，线程释放锁定有以下几种情况:

1. 同步方法或者代码块执行完毕
2. 同步方法执行到break,return
3. 同步方法执行到wait()方法
4. 同步方法遇到Error和Exception等中止程序继续运行的情况
   线程不会释放锁定的情况:
5. 执行sleep(),yield()时
6. 执行suspend()时

同步锁Lock

类似synchronized的机制，Lock常用的是ReentrantLock。同步锁需要显示的加锁和释放。
通常在finally里释放。ReentrantLock具有重入性，线程可以对它已经加锁的ReentrantLock
再次加锁，所以一段被锁保护的方法可以调用另一个被相同锁保护的方法。

死锁

少使用suspend()方法，会造成死锁。

线程通信

1. synchronized方式
   • wait() ：导致该线程等待并放弃同步监视器，知道该同步监视器的notify()或者notifyAll()方法来唤醒改线程
   • notify() ：选择任意一个在该同步监视器上的线程，将其唤醒
   • notifyAll() ：唤醒所有等待同步监视器的线程
   • 上述三个方法都属于Object而非Thread,这三个方法都必须由同步监视器来调用。
2. 同步锁方式
   • 如果不使用synchronized而是用Lock的话则不能再使用上述方法，此时我们使用Java提供的Condition类保持协调。Lock代替了同步方法或者同步块，Condition代替了同步监视器的功能。Condition提供了类似的三个方法:
   • await() : await()还有多种变体如: awaitNanos(long), awaitUninterruptibly()等
   • signal()
   • signalAll()
3. 使用管道流
   与普通IO类似，可以使用字节流、字符流和新IO Channel三种形式。通常有共享数据就可以了，不需要使用管道流。

实时分析系统设计要素

1. 数据采集
2. 数据流程
3. 数据处理
4. 数据存储
5. 数据交付

flink是什么

Flink是一个基于流计算的分布式引擎。

flink简要历史

2010年，一项名为”Stratosphere: Information Management on the Cloud”的研究项目由柏林科技大学、柏林洪堡大学、Hasso-Plattner-Institut Potsdam合作发起。Flink从Stratosphere分布式执行引擎的分支开始，并于2014年3月成为Apache孵化器项目。在2014年12月， Flink被纳入Apache顶级项目

flink的特性

1. 一个流式优先的运行时，它同时支持批处理和数据流式程序
2. 同时支持Java和Scala的优雅、流畅的API
3. 一个同时支持高吞吐和低延迟的运行时
4. 支持基于Dataflow模型的事件时间(event time)处理和乱序处理
5. 跨越不同时间语义(event time,processing time)的灵活的时间窗口(time,count,session,custom triggers)
6. 确保Exactly Once的容错性
7. 流式程序中的自然back-pressure
8. 各种类库: 图处理 (batch), Machine Learning (batch), and Complex Event Processing (streaming)
9. 在DataSet(batch)API中内置对迭代程序(BSP)的支持
10. 定制内存管理功能，可在内存和外核数据处理算法之间进行高效稳健的切换
11. Apache Hadoop MapReduce和Apache Storm的兼容层
12. 与YARN，HDFS，HBase以及Apache Hadoop生态系统的其他组件集成

预备概念

两种数据集类型

• 无界数据。持续追加的无限数据集。
• 有界数据。确定的、不会改变的数据

两种执行模型

• Streaming
• Batch

流处理技术的发展

连续数据的生产和批处理数据的消耗之间的脱节，虽然使系统构建者的工作变得容易，但是对应用开发者和DevOps团队等需要使用和管理这些系统的使用者来说，这增加了他们管理的复杂度。
为了解决这个脱节，一个开路先是Apache Strom项目，该项目在被纳入ASF之前，最初由Nathan Marz和一个叫BackType(后来被Twitter收购)的公司的一个团队发起。Storm带来了低延迟的流处理特性，但是它的实时处理带有折衷即：很难达到高吞吐，而且Storm没有提供通常需要的容错性(correctness)水平。换言之，Storm 缺少在维护正确状态时的exactly-once保证，而且Storm能提供的正确性保证也有很大的开销。

Lambda架构概览:优缺点

分布式架构如HDFS和基于批处理的MapReduce实现了可用性的扩展，但是这种方式很难解决低延迟的insight。Storm解决了低延迟的问题，但是还不是一个完整的方案，Storm不能通过exactly-once处理保证状态一致性，而且不太能解决时间处理(event-time processing)。
一种融合了这些方案的数据分析的混合观点提供了解决这些挑战的一种途径。这种混合就叫:Lambda架构,通过批处理MapReduce job提供延迟的但是准确的的结果，通过Storm提供一种及时和初步的新结果的视图.\
但是 Lambda也有它的问题所在，但一个时间窗口内由于可见的错误导致不准确数据产生时，Lambda架构需要为相同的业务逻辑编码两次，一是批处理系统，一是流式处理系统。
Spark Stream使用的是“微批”方法，不能达到完全的实时，不过可以使延迟降到几秒甚至亚秒。但是，作为低延迟/实时的代价，无法将窗口适配自然发生的会话，在表达性方面也遇到一些挑战。

mesos的起源及其所要解决的问题

2010 年， 一个旨在解决扩容问题的项目在美国加州大学伯克利分校诞生。这个软件项目就是现在所知的 Apache Mesos。它是一个开源的集群管理工具。它让运维和开发人员更
多地关注应用本身，而不是其下的服务器资源

mesos的基本运行机制

• 它在某种程度上对 CPU、内存、磁盘资源进行抽象，从而允许整个数据中心如同单台大服务器般运转。无需虚拟机和操作系统， Mesos 创造了 一个单独底层的集群为应用提供所需资源。
• Mesos的资源隔离机制支持多用户，该功能允许多个应用运行在同一个机器上。
• Mesos还从一开始就支持分布式、高可用和容错。通过使用容器技术如linux groups(cgroups)和Docker，Mesos实现了进程间的隔离，从而允许多个应用运行在同一机器上。也就是说不必再为Memecached，Jenkins CI和Ruby分别去搭建集群了。

  如何工作

  
  tip:现在slave已经改为agent

  Resource offers

  与其他集群管理器类似，Mesos集群也由一组称为master和agent的机器组成。agent定期以Resource offers的形式将可用的cpu,memory,storage通知到master。

  两层调度(two-tier scheduling)

  在Mesos集群中，Mesos的分配模块(alloction module)和框架的调度器一起负责对资源的调度，也就是所谓的双层调度.

  资源隔离

  通过Linux cgroups或者Dcoker容器，Mesos支持多租户功能，也就是允许多个进程同时在一台agent上运行。