线程与并发编程

MaYunLong

线程与并发编程

线程基础知识
线程的概念和生命周期
线程的等待方式
线程方法间的区别
并发编程
并发编程概念
阻塞队列
同步方法
原子操作
并发collection
线程池
多线程的风险

线程基础知识

线程概念：操作系统分配cpu时间片用来执行程序代码的最小单位.
线程生命周期示意图：

线程的等待(1)

Thread.sleep();

Thread.wait();

Thread.yield();

Thread.join();

线程的等待(2)

sleep使用场景：
程序需要暂停操作：如：监听队列时的时间间隔；测试cache时等待过期等。
解释：使当前线程暂停millis所指定的毫秒，开始重新抢占CPU。

线程的等待(3)

wait使用场景：  
生产者和消费者模式 

wait() 必须在synchronized 函数或者代码块里面  

wait()会让已经获得synchronized 函数或者代码块控制权的Thread暂时休息，并且丧失控制权  

这个时候，由于该线程丧失控制权并且进入等待，其他线程就能取得控制权，并且在适当情况下调用notifyAll()来唤醒wait()的线程。
  
需要注意的是，被唤醒的线程由于已经丧失了控制权，所以需要等待唤醒它的线程结束操作，从而才能重新获得控制权。  

所以wait()的确是马上让当前线程丧失控制权，其他的线程可以乘虚而入。  

所以wait()的使用，必须存在2个以上线程，而且必须在不同的条件下唤醒wait()中的线程。 

线程的等待(4)

Thread.yield():
    api中解释： 暂停当前正在执行的线程对象，并执行其他线程。
    注意：这里的其他也包含当前线程，所以会出现以下结果。
   1. public class Test extends Thread {   
   2.   public static void main(String[] args) {   
   3.     for (int i = 1; i <= 2; i++) {   
   4.       new Test().start();   
   5.     }   
   6.   }   
   7.   
   8.   public void run() {   
   9.     System.out.print("1");   
  10.     yield();   
  11.     System.out.print("2");   
  12.   }   
  13. }  
    输出结果： 1122 或者 1212 
所以，yield是让出CPU，马上重新排队竞争CPU时间片。

线程的等待(5-1)

Thread.Join() 用法的理解 
指在一线程里面调用另一线程join方法时，表示将本线程阻塞直至另一线程终止时再执行   
我们首先来看个例子：代码：public class ThreadTest implements Runnable {public static int a = 0;public void run() {for (int k = 0; k < 5; k++) {a = a + 1;}}public static void main(String[] args) throws Exception {Runnable r = new ThreadTest();Thread t = new Thread(r);t.start();System.out.println(a);}}

线程的等待(5-2)

请问程序的输出结果是5吗？答案是：有可能。其实你很难遇到输出5的时候，通常情况下都不是5。当然这也和机器有严重的关系。为什么呢？我的解释是当主线程main方法执行System.out.println(a);这条语句时，线程还没有真正开始运行，或许正在为它分配资源准备运行吧。因为为线程分配资源需要时间，而main方法执行完t.start()方法后继续往下执行System.out.println(a);,这个时候得到的结果是a还没有被改变的值0。怎样才能让输出结果为5！其实很简单，join()方法提供了这种功能。join()方法，它能够使调用该方法的线程在此之前执行完毕。 
那么怎么样才能输出5呢？

线程的等待(5-3)

public class ThreadTest implements Runnable {public static int a = 0;public void run() {for (int k = 0; k < 5; k++) {a = a + 1;}}public static void main(String[] args) throws Exception {Runnable r = new ThreadTest();Thread t = new Thread(r);t.start();t.join();System.out.println(a);}}这个时候，程序输入结果始终为5。

notify与notifyAll区别 

调用notifyAll通知所有线程继续执行，只能有一个线程在执行其余的线程在等待(因为在所有线程被唤醒的时候在synchornized块中)。这时的等待和调用notifyAll前的等待是不一样的。

notifyAll前：在对象上休息区内休息
notifyAll后：在排队等待获得对象锁。

notify和notifyAll都是把某个对象上休息区内的线程唤醒,notify只能唤醒一个,但究竟是哪一个不能确定,而notifyAll则唤醒这个对象上的休息室中所有的线程.

  一般有为了安全性,我们在绝对多数时候应该使用notifiAll(),除非你明确知道只唤醒其中的一个线程.

至于有些书上说“notify：唤醒同一对象监视器中调用wait的第一个线程”我认为是没有根据的因为sun公司是这样说的“The choice is arbitrary and occurs at the discretion of the implementation.” 

wait 和yield区别

1）
定义上   wait()   的启动办法是notify()   和   notifyAll(）方法；
yield()会自动切换回来。   

2）
wait()方法是object的,所以他只停止了current   Thread   的一个锁，这可能产生的结果就是如果还有锁锁着其他threads那么那些thread可就惨了，所以要慎用； 
yield()   主要用在自行判断优先级的场合，是一个主动的暂停。

3）
wait()将所有资源让出来，等得到通知后在参加资源竞争
yield()将cpu资源让出来但马上就参加cpu资源竞争。

Thread与Runable区别

Runnable是Thread的接口，在大多数情况下“推荐用接口的方式”生成线程，因为接口可以实现多继承，况且Runnable只有一个run方法，很适合继承。

在使用Thread的时候只需要new一个实例出来，调用start()方法即可以启动一个线程。
Thread Test = new Thread();
Test.start();

在使用Runnable的时候需要先new一个继承Runnable的实例，之后用子类Thread调用。
Test impelements Runnable
Test t = new Test();
Thread test = new Thread(t); 

Callable与Runable区别

Callable是类似于Runnable的接口，实现Callable接口的类和实现Runnable的类都是可被其他线程执行的任务。

Callable和Runnable的区别如下：
Callable定义的方法是call，而Runnable定义的方法是run。
Callable的call方法可以有返回值，而 Runnable的run方法不能有返回值。
Callable的call方法可抛出异常，而Runnable 的run方法不能抛出异常。 

什么是并发编程

Concurrent(并发), parallel(并行)：并发从用户角度看，同时发起任务请求，而并行是针对代码的执行顺序来看，相当于同时执行代码，需要有多核或多cpu的系统。而多线程是完成并发任务请求的最佳编程模型，因此大多数时候说的java并发编程都指的多线程编程

从任务角度看，在一台机器有各种各样的任务，有的任务是比较耗cpu的，有的是耗memory的，有的是耗I/O的，如果所有的任务都一件件完成，将导致完成耗I/O的任务过程CPU,memory一直空闲，耗CPU的任务memory和I/O空闲，无法充分利用CPU资源，特别是现在的多核系统将无法充分发挥多核的作用，因此多线程是充分利用系统资源，提高应用性能的有效手段。

阻塞队列

BlockQuene:
特点：
支持两个附加操作的 Queue，这两个操作是：检索元素时等待队列变为非空，以及存储元素时等待空间变得可用。
BlockingQueue 不接受 null 元素。
BlockingQueue 可以是限定容量的。
BlockingQueue 实现是线程安全的
适合场景：
BlockingQueue 实现主要用于生产者-使用者队列
子类：
ArrayBlockingQueue：一个由数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序
LinkedBlockingQueue：一个基于已链接节点的、范围任意的 blocking queue。此队列按 FIFO（先进先出）排序元素
DelayQueue：Delayed 元素的一个无界阻塞队列，只有在延迟期满时才能从中提取元素
SynchronousQueue：一种阻塞队列，其中每个 put 必须等待一个 take，反之亦然 

线程同步(1)

ReentrantLock
一个可重入的互斥锁定 Lock，它具有与使用 synchronized 方法和语句所访问的隐式监视器锁定相同的一些基本行为和语义，但功能更强大。 
 class X {
   private final ReentrantLock lock = new ReentrantLock();
public void m() { 
     lock.lock();  // block until condition holds
     try {
       // ... method body
     } finally {
       lock.unlock();
     }
   }
 }
 lock();//获取锁定
 tryLock(long timeout, TimeUnit unit);//如果锁定在给定等待时间内没有被另一个线程保持，且当前线程未被中断，则获取该锁定。
 unlock();//试图释放此锁定。
 isLocked 和 getLockQueueLength 方法，以及一些相关的 protected 访问方法，这些方法对检测和监视可能很有用。 

线程同步(2)

 ReentrantReadWriteLock.WriteLock
 ReentrantReadWriteLock.ReadLock
 class CachedData {
   Object data;
   volatile boolean cacheValid;
   ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
   void processCachedData() {
     rwl.readLock().lock();
     if (!cacheValid) {
        // upgrade lock manually
        rwl.readLock().unlock();   // must unlock first to obtain writelock
        rwl.writeLock().lock();
        if (!cacheValid) { // recheck
          data = ...
          cacheValid = true;
        }
        // downgrade lock
        rwl.readLock().lock();  // reacquire read without giving up write lock
        rwl.writeLock().unlock(); // unlock write, still hold read
     }
     use(data);
     rwl.readLock().unlock();
   }
 }
 规则：写入者可以获取读取锁定——但反过来则不成立（锁定降级）

线程同步(3-1)

尽可能Nonblocking例子

private static Map<String,Connection> connections=new HashMap<String,Connection>();

public Connection get(String key) throws Exception{
synchronized(connections){
if(connections.containsKey(key)){
return connections.get(key);
}
Connection conn=createConnection(key);
connections.put(key,conn);
return conn;
}
}

线程同步(3-2)

改造想法
connections改为ConcurrentHashMap
借助putIfAbsent来减少判断是否存在的lock

private static ConcurrentHashMap<String,Connection> connections=new ConcurrentHashMap<String,Connection>();

public Connection get(String key) throws Exception{
if(connections.containsKey(key)){
return connections.get(key);
}
Connection conn=createConnection(key);
Connection realConn=connections.putIfAbsent(key,conn);
if(realConn!=null){
conn.close();
}
 return realConn;
}

线程同步(3-3)

改造想法
能否不创建无谓的连接，想到了FutureTask

private static ConcurrentHashMap<String,FutureTask<Connection>> connections=new ConcurrentHashMap<String,FutureTask<Connection>>();

public Connection get(final String key) throws Exception{
if(connections.containsKey(key)){
return connections.get(key).get;
}
FutureTask<Connection> newTask=new FutureTask<Connection>(new Callable<Connection>(){
public Connection call() throws Exception{
return createConnection(key);
}
});
FutureTask<Connection> task=connections.putIfAbsent(key, newTask);
if(task==null){
newTask.run();
return newTask.get();
}
 return task.get();
}

同步注意事项

访问共享数据前问问自己是否需要进行同步
尽量避免一次操作拿多把锁（容易产生死锁）
尽量避免在同步块内调用其他对象的方法
尽可能减少锁持时间
使用显示锁时一定要遵循try..finally模式
尽可能使用现有的1.5同步工具类进行同步
使用volatile只能增加可见性，没有同步语义
不要改变默认线程优先级
尽量减少共享对象的状态
谨慎对待InterruptException，不要轻易忽略

原子操作

原子操作：
AtomicBoolean:可以用原子方式更新的 boolean 值
AtomicInteger:可以用原子方式更新的 int 值
AtomicLong:可以用原子方式更新的 long 值
AtomicIntegerArray:可以用原子方式更新其元素的 int 数组
AtomicReference:可以用原子方式更新的对象引用

原子类型特点：
原子类型没有使用锁, 是无阻塞，通过使用volatile和CPU原子语义CAS实现原子操作

时间间隔(TimeUnit)

时间间隔：(TimeUnit)
MICROSECONDS             
MILLISECONDS             
NANOSECONDS             
SECONDS  

sleep(long timeout);

使用此单元执行 Thread.sleep.这是将时间参数转换为 Thread.sleep 方法所需格式的便捷方法。如：TimeUnit.SECONDS.sleep(3);

并发collection

并发 Collection:

ConcurrentHashMap(支持并发)
WeakHashMap(以弱键 实现的基于哈希表的 Map。在 WeakHashMap 中，当某个键不再正常使用时，将自动移除其条目)
MapMaker(google开发、线程安全，高并发性能，异步超时清理)
ConcurrentMap<Key, Value> map = new MapMaker()
.weakValues() // 指定Map保存的Value为WeakReference机制
.makeMap(); 
ConcurrentMap<Key, Value> map = new MapMaker() // 构建一个 computingMap
.expiration(60, TimeUnit.SECONDS) // 元素60秒过期
. makeMap ();

CopyOnWriteArrayList 
ArrayList 的一个线程安全的变体，其中所有可变操作（添加、设置，等等）都是通过对基础数组进行一次新的复制来实现的
这一般需要很大的开销，但是当遍历操作的数量大大超过可变操作的数量时，这种方法可能比其他替代方法更 有效。

CopyOnWriteArraySet
它最适合于 set 大小通常保持很小、只读操作远多于可变操作以及需要在遍历期间防止线程间冲突的应用程序。 

异步线程结果

Future:
interface ArchiveSearcher { String search(String target); }
 class App {
   ExecutorService executor = Executors.newCachedThreadPool();
   ArchiveSearcher searcher = ...
   void showSearch(final String target) throws InterruptedException {
     Future<String> future = executor.submit(new Callable<String>() {
         public String call() { return searcher.search(target); }
     });
     displayOtherThings(); // do other things while searching
     try {
       displayText(future.get()); // use future
     } catch (ExecutionException ex) { cleanup(); return; }
   }
 }
 FutureTask:
FutureTask 类是 Future 的一个实现，Future 可实现 Runnable，所以可通过 Executor 来执行。例如，可用下列内容替换上面带有 submit 的构造： 
     Future<String> future =
       new FutureTask<String>(new Callable<String>() {
         public String call() {
           return searcher.search(target);
       }});
     executor.execute(future);
特点：可实现必须完成的任务的并行执行。

线程池(1)

ScheduledThreadPoolExecutor:
ThreadPoolExecutor pool=new ThreadPoolExecutor(corePoolSize,
maxinumPoolSize,
keepAliveTime,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new ThreadPoolExecutor.CallerRunsPolicy());
核心和最大池大小 
ThreadPoolExecutor 将根据 corePoolSize（参见 getCorePoolSize()）和 maximumPoolSize（参见 getMaximumPoolSize()）设置的边界自动调整池大小。当新任务在方法 execute(java.lang.Runnable) 中提交时，如果运行的线程少于 corePoolSize，则创建新线程来处理请求，即使其他辅助线程是空闲的。如果运行的线程多于 corePoolSize 而少于 maximumPoolSize，则仅当队列满时才创建新线程。如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务。在大多数情况下，核心和最大池大小仅基于构造来设置，不过也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。 

线程池(2)

保持活动时间 

如果池中当前有多于 corePoolSize 的线程，则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止（参见 getKeepAliveTime(java.util.concurrent.TimeUnit)）。这提供了当池处于非活动状态时减少资源消耗的方法。如果池后来变得更为活动，则可以创建新的线程。也可以使用方法 setKeepAliveTime(long, java.util.concurrent.TimeUnit) 动态地更改此参数。使用 Long.MAX_VALUE TimeUnit.NANOSECONDS 的值在关闭前有效地从以前的终止状态禁用空闲线程。 

线程池(3)

排队 
所有 BlockingQueue 都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互： 
如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。 
如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。 
如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。 
排队有三种通用策略： 
直接提交。工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集合时出现锁定。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。 
无界队列。使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙的情况下将新任务加入队列。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。 
有界队列。当使用有限的 maximumPoolSizes 时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。 

线程池(4)

被拒绝的任务 
当 Executor 已经关闭，并且 Executor 将有限边界用于最大线程和工作队列容量，且已经饱和时，在方法 execute(java.lang.Runnable) 中提交的新任务将被拒绝。在以上两种情况下，execute 方法都将调用其 RejectedExecutionHandler 的 RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。下面提供了四种预定义的处理程序策略： 
在默认的 ThreadPoolExecutor.AbortPolicy 中，处理程序遭到拒绝将抛出运行时 RejectedExecutionException。 
在 ThreadPoolExecutor.CallerRunsPolicy 中，线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。 
在 ThreadPoolExecutor.DiscardPolicy 中，不能执行的任务将被删除。 
在 ThreadPoolExecutor.DiscardOldestPolicy 中，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）。 
定义和使用其他种类的 RejectedExecutionHandler 类也是可能的，但这样做需要非常小心，尤其是当策略仅用于特定容量或排队策略时。

线程池(5)

队列维护 
方法 getQueue() 允许出于监控和调试目的而访问工作队列。强烈反对出于其他任何目的而使用此方法。remove(java.lang.Runnable) 和 purge() 这两种方法可用于在取消大量已排队任务时帮助进行存储回收。

getActiveCount() 
          返回主动执行任务的近似线程数。

shutdown() 
          按过去执行已提交任务的顺序发起一个有序的关闭，但是不接受新任务。

多线程！=高性能

Amdahl定律

。如果F是必须串行化执行的比重，那么Amdahl定律告诉我们，在一个N处理器的机器中
，我们最多可以加速：

多线程的风险和注意事项

多线程的风险：
死锁、资源不足、并发错误、线程泄漏、请求过载 

多线程的风险(Multi-thread Anti-pattern).txt

线程交互
wait/notify(notifyAll)
在测kilim一个版本时，高压力的情况下wait/notify貌似有bug，jdk是1.6.0_07
线程池
ThreadPoolExecutor做的已经不错了，但要注意合理使用
不要使用无限制大小的线程池
最好自行实现ThreadFactory，最少给线程加上个前缀
当超过coreSize后，会扔到指定的BlockingQueue中，因此要注意这个地方 

为什么要多线程

谢谢！