多線程及線程並發庫

1. 創建和啟動線程的兩種傳統方式

概念：java 實現一條線索
有兩種方法：
1、繼承thread 類
例子：
// 繼承Thread類

 new Thread(){
        publicvoid run(){
          while(true){
             try {
                Thread.sleep(2000);
             }catch (InterruptedException e) {
                e.printStackTrace();
             }
   System.out.println(Thread.currentThread().getName());
          }
        }
      }.start();  

2、實現runnable接口
例子：

//實現Runnable接口
      new Thread(new Runnable(){
        publicvoid run(){
          while(true){
             try {
                Thread.sleep(2000);
             }catch (InterruptedException e) {
                e.printStackTrace();
             }
          System.out.println(Thread.currentThread().getName());
          }
        }
      }).start();

2.傳統的定時器實現方式
代碼

importjava.util.concurrent.Executors;
importjava.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
publicclass TimerTest {
   /**
    * 定時器
    */
   publicstaticvoid main(String[] args) {
      ScheduledExecutorServiceservice =  Executors.newScheduledThreadPool(3);
      service.scheduleAtFixedRate(
          new Runnable(){
             publicvoid run() {
                System.out.println(Thread.currentThread().getName());
             }

          },
          10,
          1,
          TimeUnit.SECONDS);

      while(true){
        try {
          Thread.sleep(2000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName());
      }
   }

}

3.傳統的線程同步互斥

 線程的同步和互斥
   Synchronized
線程創建過程：同步：子線程循環10次，接著主線程循環100次，接著的又回到子線程循環10次，接著又回到主線程循環10次。如此，循環50次。

packageorg.nagi.Thread;
publicclass TeaTSynchronized {
   static Businessbusiness =new Business();
   publicstaticvoid main(String[] args)throws InterruptedException {
      //子線程
      TeaTSynchronized.init();
      //主線程
      for(int i=0;i<50;i++){
        business.main();
      }
   }

   //子線程
   publicstaticvoid init(){
        new Thread(new Runnable() {
          @Override
          publicvoid run() {
             for(int i=0;i<50;i++){
                 try {
                   business.sub();
                }catch (InterruptedException e) {
                   e.printStackTrace();
                }
             }
          }
        }).start();
   }
}

 class Business{
    booleanshouldSub =true;

    //子線程方法
    publicsynchronizedvoid sub()throws InterruptedException{
      //使用我當前線程的，等待
       while(!shouldSub){//這裡使用while而不用iF防止假喚醒！詳細信息可參考JavaApi
         this.wait();
            }
       for(int i=0;i<10;i++){
         System.out.println(Thread.currentThread().getName()+"count" + i);
       }
       shouldSub = false;
       this.notify();

    }
    //主線程的操作方法
    publicsynchronizedvoid main()throws InterruptedException{
       while(shouldSub){//這裡使用while而不用iF防止假喚醒！詳細信息可參考JavaApi
         //使用我當前線程的，等待
         this.wait();
       }
       for(int i=0;i<100;i++){
         System.out.println(Thread.currentThread().getName()+"count" + i);
       }
       shouldSub = true;
       this.notify();
    }
}

4.傳統的線程間通訊

 ThreadLoacal 線程間的通信：就是線程內的數據共享，兩個或多個代碼塊共享的一個數據， 在代碼塊之間數據是獨立，在代碼塊之內是數據共享的。

5.ThreadLocal與線程級變量共享

   ThreadLoacal就相當Map; 一個線程TheadLoacal 只能放一個數據；

6.多線程訪問共享數據的經典總結

   1、可以把需要共享的數據轉給runnable 接口；
   2、將共享數據的封裝為一個對象， 在多個線程裡面runnable 接口run方法
   ３、可以使用ＴｈｒｅａｄＬｏｃａｌ類

如果想要操作一個原子性的數據類型可以用原子方式更新的 int 值。
在操作某一個數據原子性操作的時候，可以使用AtomicInteger　AtomicIntegerArray

 定義一個線程池：

packageorg.nagi.Thread;

importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;

/**
 *線程池
 *@author dell
 *
 */
publicclass ThreadPool {
   publicstaticvoid main(String[] args) {
      /*緩存線程池*/
      //ExecutorService  threadcached=Executors.newCachedThreadPool();
      /*單個線程池，當線程死了，自動會有線程啟動*/
      //ExecutorService threadSingle = Executors.newSingleThreadScheduledExecutor();
      //線程池
      ExecutorService  threadPool = Executors.newFixedThreadPool(4);
        threadPool.execute(new Runnable() {
          @Override
          publicvoid run() {
             for(int i=0;i<5;i++){
                System.out.println(Thread.currentThread().getName()+"任務："+ i);
             }
          }
        });
        //當任務執行完成之後就可以關閉掉
        threadPool.shutdown();
   }
}

8.Java5中的定時器

方法參考
ScheduledFuture
schedule(Callable callable, long delay, TimeUnit unit)
創建並執行在給定延遲後啟用的 ScheduledFuture。
ScheduledFuture

   Executors.newScheduledThreadPool(4).scheduleAtFixedRate(
             new Runnable() {
                @Override
                publicvoid run() {
                   System.out.println("Bombing");
                }
             },
             4,//多少秒之後執行run方法
             2,//間隔多少時間執行
              TimeUnit.SECONDS);//時間單位
Callable和Futrue使用:

        /*使用Callable調用回返一個數據，使用futrue去取並且可以指定時間*/
        ExecutorService  threadSingle = Executors.newSingleThreadExecutor();
        Future futrue =threadSingle.submit(newCallable() {

          @Override
          public String call()throws Exception {
             Thread.sleep(1000);
             return"hello world";
          }
        });
        try {
          //futrue取出數據
          System.out.println("result:" + futrue.get(3,TimeUnit.SECONDS));
        } catch (Exceptione) {
          e.printStackTrace();
        }

9.java5提供的可返回結果的線程模式

10.java5中的鎖與讀寫鎖

Lock：java 中鎖Lock 比Synchronized更加的面對對象，與生活中的對象一樣，鎖業是一個對象。
例子代碼：

importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;
importjava.util.concurrent.locks.Lock;
importjava.util.concurrent.locks.ReentrantLock;

publicclass LockTest {

   /**
    * @param args
    */
   publicstaticvoid main(String[] args) {
      final Business business =new Business();
      ExecutorService service = Executors.newFixedThreadPool(3);
      for(int i=0;i<3;i++){
        service.execute(new Runnable(){
          publicvoid run(){
             business.service();
          }
        });
      }
      service.shutdown();
   }
}
class Business{
   privateintcount = 0;
   Lock lock = new ReentrantLock();
   publicvoid service(){
      lock.lock();
      count++;
      try {
        Thread.sleep(10);
        System.out.println(count);    
      } catch (InterruptedException e) {
        e.printStackTrace();
      }finally{
        lock.unlock();
      }

   }
}

Lock : 還可以實現一個讀寫鎖，
/**
    * 讀的的時候不互斥
    * 寫的時候互斥
    * @param key
    * @return
    */
   public Object getEntityFromCache(Stringkey){

      Object object = cache.get(key);
      if(object ==null){
        lock.readLock().lock();//鎖得read操作
        try{
          object ="btbu";//去數據庫查找
          cache.put(key, object);
        }finally{
          lock.readLock().unlock();
        }
      }
      return object;
   }

11.LockCondition同步工具

Condition 將 Object 監視器方法（wait、notify和notifyAll）分解成截然不同的對象，以便通過將這些對象與任意Lock 實現組合使用，為每個對象提供多個等待 set（wait-set）。其中，Lock 替代了 synchronized 方法和語句的使用，Condition 替代了 Object 監視器方法的使用。

例子：

packageorg.nagi.Thread;

import java.util.concurrent.locks.Condition;
importjava.util.concurrent.locks.ReentrantLock;


publicclass TestCondition {
   static Businessbusiness =new Business();
   publicstaticvoid main(String[] args)throws InterruptedException {
      //子線程
      TestCondition.init();
      //主線程
      for(int i=0;i<50;i++){
        business.main();
      }
   }

   //子線程
   publicstaticvoid init(){
        new Thread(new Runnable() {
          @Override
          publicvoid run() {
             for(int i=0;i<50;i++){
                 try {
                   business.sub();
                }catch (InterruptedException e) {
                   e.printStackTrace();
                }
             }
          }
        }).start();

   }

}

 class Business{
    booleanshouldSub =true;

    ReentrantLock lock = new ReentrantLock();
    Condition condition =lock.newCondition();

    //子線程方法
    public void sub()throws InterruptedException{
       lock.lock();
       try{
          //使用我當前線程的，等待
           while(!shouldSub){//這裡使用while而不用iF防止假喚醒！詳細信息可參考JavaApi
               condition.await();
                }
           for(int i=0;i<10;i++){
              System.out.println(Thread.currentThread().getName()+"count" + i);
           }
           shouldSub =false;
       }finally{
         condition.signal();
       }

    }
    //主線程的操作方法
    publicsynchronizedvoid main()throws InterruptedException{
       lock.lock();
       try{
         while(shouldSub){//這裡使用while而不用iF防止假喚醒！詳細信息可參考JavaApi
           //使用我當前線程的，等待
           condition.await();
         }
         for(int i=0;i<100;i++){
           System.out.println(Thread.currentThread().getName()+"count" + i);
         }
         shouldSub = true;
         this.notify();
       }finally{
         condition.signal();
       }
    }
}

緩沖隊列的實現

import java.util.concurrent.locks.Condition;
importjava.util.concurrent.locks.Lock;
importjava.util.concurrent.locks.ReentrantLock;
/**
 *緩沖隊列
 *@author dell
 *
 */
class BoundedBuffer{
   final Locklock =new ReentrantLock();
   final ConditionnotFull =lock.newCondition();
   final ConditionnotEmpty =lock.newCondition();

   //一個數組的緩存區
   final Object[]items =new Object[100];
   intputptr,takeptr,count;

   //存放的操作
   publicvoid put(Object x)throws InterruptedException {
     lock.lock();
     try {
       while (count ==items.length)
         notFull.await();
       items[putptr] = x;
       if (++putptr ==items.length)putptr = 0;
       ++count;
       notEmpty.signal();
     } finally {
       lock.unlock();
     }
   }

   //取出數據的操作
   public Object take()throwsInterruptedException {
     lock.lock();
     try {
       while (count == 0)
         notEmpty.await();
       Object x = items[takeptr];
       if (++takeptr ==items.length)takeptr = 0;
       --count;
       notFull.signal();
       return x;
     } finally {
       lock.unlock();
     }
   }
 }

12.Semaphore同步工具

代碼:

importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;
importjava.util.concurrent.Semaphore;

/**
 *信號燈
 *@author dell
 *
 */
publicclass SemaphoreTest {
   publicstaticvoid main(String[] args) {
      //緩存連接池
      ExecutorService service = Executors.newCachedThreadPool();
      final Semaphore sp =new Semaphore(3);
      for(int i=0;i<10;i++){
        Runnable runnable = new Runnable(){
             publicvoid run(){
             try {
                sp.acquire();//獲得信號量許可
             }catch (InterruptedException e1) {
                e1.printStackTrace();
             }
             System.out.println("線程" + Thread.currentThread().getName() +
                   "進入，當前已有" + (3-sp.availablePermits()) +"個並發");
             try {
                Thread.sleep((long)(Math.random()*10000));
             }catch (InterruptedException e) {
                e.printStackTrace();
             }
             System.out.println("線程" + Thread.currentThread().getName() +
                   "即將離開");             
              sp.release();//釋放信號量
             //下面代碼有時候執行不准確，因為其沒有和上面的代碼合成原子單元
             System.out.println("線程" + Thread.currentThread().getName() +
                   "已離開，當前已有" + (3-sp.availablePermits()) +"個並發");             
          }
        };
        //執行已提交的 Runnable任務的對象。
        service.execute(runnable);    
      }
   }

}

運行結果：
線程pool-1-thread-1進入，當前已有1個並發
線程pool-1-thread-1即將離開
線程pool-1-thread-1已離開，當前已有0個並發
線程pool-1-thread-2進入，當前已有1個並發
線程pool-1-thread-2即將離開
線程pool-1-thread-2已離開，當前已有0個並發
線程pool-1-thread-3進入，當前已有1個並發
線程pool-1-thread-3即將離開

13.CyclicBarrier同步工具

一個同步輔助類，它允許一組線程互相等待，直到到達某個公共屏障點 (common barrier point)。在涉及一組固定大小的線程的程序中，這些線程必須不時地互相等待，此時 CyclicBarrier 很有用。因為該 barrier 在釋放等待線程後可以重用，所以稱它為循環的 barrier。

代碼：

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 *並發工具輔佐類
 *
 *之循環障礙點
 *@author dell
 *
 */
publicclass CyclicBarrierTest {

   publicstaticvoid main(String[] args) {
      ExecutorService service = Executors.newCachedThreadPool();
      final CyclicBarrier cb =newCyclicBarrier(3);
      for(int i=0;i<3;i++){
        Runnable runnable = new Runnable(){
             publicvoid run(){
             try {
                Thread.sleep((long)(Math.random()*10000));  
               System.out.println("線程" + Thread.currentThread().getName() +
                     "即將到達集合地點1，當前已有" + cb.getNumberWaiting() +"個已經到達，正在等候");               
                cb.await();

                Thread.sleep((long)(Math.random()*10000));  
               System.out.println("線程" + Thread.currentThread().getName() +
                     "即將到達集合地點2，當前已有" + cb.getNumberWaiting() +"個已經到達，正在等候");               
                cb.await();
                Thread.sleep((long)(Math.random()*10000));  
               System.out.println("線程" + Thread.currentThread().getName() +
                     "即將到達集合地點3，當前已有" + cb.getNumberWaiting() +"個已經到達，正在等候");               
                cb.await();             
             }catch (Exception e) {
                e.printStackTrace();
             }         
          }
        };
        service.execute(runnable);

      }
      service.shutdown();
   }

}

14.CountDownLatch同步工具

一個同步輔助類，在完成一組正在其他線程中執行的操作之前，它允許一個或多個線程一直等待。首先先是點時器，
用給定的計數 初始化 CountDownLatch。由於調用了countDown()方法，所以在當前計數到達零之前，await方法會一直受阻塞。之後，會釋放所有等待的線程，await的所有後續調用都將立即返回。這種現象只出現一次——計數無法被重置。

importjava.util.concurrent.CountDownLatch;
importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;

/**
 *程序計數器
 *@author dell
 *
 */
publicclass CountdownLatchTest {

   publicstaticvoid main(String[] args) {
      ExecutorService service = Executors.newCachedThreadPool();
      //命令
      final CountDownLatch cdOrder =new CountDownLatch(1);
      //需要3個都為了時候再執行
      final CountDownLatch cdAnswer =new CountDownLatch(3);    
      for(int i=0;i<3;i++){
        Runnable runnable = new Runnable(){
             publicvoid run(){
              try {
               System.out.println("線程" + Thread.currentThread().getName() +
                     "正准備接受命令");               
                cdOrder.await();
               System.out.println("線程" + Thread.currentThread().getName() +
                "已接受命令");                   
                Thread.sleep((long)(Math.random()*10000));  
               System.out.println("線程" + Thread.currentThread().getName() +
                     "回應命令處理結果");
                //減1，當計數器為0的時候，再可以往下執行
                cdAnswer.countDown();              
             }catch (Exception e) {
                e.printStackTrace();
             }         
          }
        };
        service.execute(runnable);
      }    
      try {
        Thread.sleep((long)(Math.random()*10000));

        System.out.println("線程" + Thread.currentThread().getName() +
             "即將發布命令");              
        cdOrder.countDown();
        System.out.println("線程" + Thread.currentThread().getName() +
        "已發送命令，正在等待結果");
        cdAnswer.await();
        System.out.println("線程" + Thread.currentThread().getName() +
        "已收到所有響應結果");
      } catch (Exception e) {
        e.printStackTrace();
      }         
      service.shutdown();
   }
}

15.Exchanger同步與數據交換工具

可以在對中對元素進行配對和交換的線程的同步點。每個線程將條目上的某個方法呈現給exchange方法，與伙伴線程進行匹配，並且在返回時接收其伙伴的對象。Exchanger 可能被視為SynchronousQueue 的雙向形式。Exchanger 可能在應用程序（比如遺傳算法和管道設計）中很有用。

importjava.util.concurrent.Exchanger;
importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;

/**
 *當兩個現場都到達的時候，再可以實現線程的數據交換
 *@author dell
 *
 */
publicclass ExchangerTest {

   publicstaticvoid main(String[] args) {
      ExecutorService service = Executors.newCachedThreadPool();
      finalExchanger exchanger =newExchanger();
      service.execute(new Runnable(){
        publicvoid run() {
          try {        
             Thread.sleep((long)(Math.random()*10000));
             String data1 ="zxx";
             System.out.println("線程" + Thread.currentThread().getName() +
             "正在把數據" + data1 +"換出去");
             String data2 = (String)exchanger.exchange(data1);
             System.out.println("線程" + Thread.currentThread().getName() +
             "換回的數據為" + data2);
          }catch(Exception e){

          }
        } 
      });
      service.execute(new Runnable(){
        publicvoid run() {
          try {        
             Thread.sleep((long)(Math.random()*10000));
             String data1 ="lhm";
             System.out.println("線程" + Thread.currentThread().getName() +
             "正在把數據" + data1 +"換出去");
             String data2 = (String)exchanger.exchange(data1);
             System.out.println("線程" + Thread.currentThread().getName() +
             "換回的數據為" + data2);
          }catch(Exception e){

          }         
        } 
      });    
   }
}

16.Java5中的可阻塞隊列
隊列：先進先出

   import java.util.concurrent.ArrayBlockingQueue;
importjava.util.concurrent.BlockingQueue;
importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;

/**
 *阻塞隊列
 *@author dell
 *
 */
publicclass BlockingQueueCondition {

   publicstaticvoid main(String[] args) {
      ExecutorService service = Executors.newSingleThreadExecutor();
      final Business3 business =new Business3();
      service.execute(new Runnable(){

        publicvoid run() {
          for(int i=0;i<50;i++){
             business.sub();
          }
        }

      });

      for(int i=0;i<50;i++){
        business.main();
      }
   }

}

class Business3{
   //隊列
   BlockingQueue subQueue =newArrayBlockingQueue(1);
   BlockingQueue mainQueue =newArrayBlockingQueue(1);
   {//匿名構造方法，有多少個對象就有多少匿名構造方法
      try {
        mainQueue.put(1);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
   }
   publicvoid sub(){
      try
      {
        mainQueue.take();
        for(int i=0;i<10;i++){
          System.out.println(Thread.currentThread().getName()+" :" + i);
        }
        subQueue.put(1);
      }catch(Exception e){

      }
   }

   publicvoid main(){

      try
      {
        subQueue.take();
        for(int i=0;i<5;i++){
          System.out.println(Thread.currentThread().getName()+" :" + i);
        }
        mainQueue.put(1);
      }catch(Exception e){
      }    
   }
}

17.傳統與java5中的同步集合

1、在java5 之後並入了一個並發事務庫
java.util.concurrent
在並發編程中很常用的實用工具類。

2、HashSet 是單列表的只有的一個Key,而HashMap 雙列表一個key 和一個value,HashSet 是底層是使用HashMap ,只是不用Value的那一部分。

3、在迭代集合集合的時候不能對集合進行需要。