<jdk7學習筆記>讀書筆記-并行api

bovenson 發布于2019-08-14 12:47 / 621人閱讀

摘要：然而，這兩個方法都只是讀取對象狀態，如果只是讀取操作，就可以允許線程并行，這樣讀取效率將會提高。分配線程執行子任務執行子任務獲得子任務進行完成的結果

Lock

Lock接口主要操作類是ReentrantLock,可以起到synchronized的作用，另外也提供額外的功能。
用Lock重寫上一篇中的死鎖例子

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Resource {
    Lock lock=new ReentrantLock();
    int num=0;
    void doSome(){
        
    }
    public void deal(Resource res){
        while(true){
            boolean mylock=this.lock.tryLock();//嘗試獲得當前Resource的鎖定
            boolean resLock=res.lock.tryLock();//嘗試獲得傳入的Resource的鎖定
            try{
                if(mylock&&resLock){
                    res.doSome();
                    System.out.println(res+":"+this.num);
                    break;//退出循環
                }
            }finally{
                if(mylock)
                    this.lock.unlock();
                if(resLock)
                    res.lock.unlock();
            }
        }
    }
}

重寫后不會出現死鎖的原因在于，當無法同時獲得兩個鎖定時，干脆釋放已獲得的鎖定。
上面代碼使用當前Resource的Lock的tryLock()方法嘗試獲得鎖定，以及傳入Resource的Lock的tryLock()方法嘗試獲得鎖定。只有當可以獲得兩個Resource的鎖定，才能執行res.doSome().最后無論什么情況，都要finally解除鎖定。

ReadWriteLock

ReadWriteLock接口定義了讀取鎖定和寫入鎖定的行為。可以使用readLock(),writeLock()方法返回Lock操作對象。
ReentrantReadWriteLock是ReadWriteLock接口的主要操作類.
ReentrantReadWriteLock.readLock操作Lock接口，調用其lock()方法時，若沒有任何ReentrantReadWriteLock.writeLock調用過lock()方法,也就是沒有任何寫入鎖定時，才可以取得讀取鎖定。
下面用ReadWriteLock試著寫一個ArrayList

import java.util.Arrays;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class MyArrayList {
    private ReadWriteLock lock=new ReentrantReadWriteLock();
    private Object[] list;
    private int next=0;
    public MyArrayList(){
        list=new Object[16];
    }
    public void add(T obj){
        try{
            lock.writeLock().lock();//獲取寫入鎖定
            if(next==list.length)
                list=Arrays.copyOf(list, list.length*2);
            list[next++]=obj;
        }finally{
            lock.writeLock().unlock();//解除寫入鎖定
        }
    }
    @SuppressWarnings("unchecked")
    public T get(int index){
        try{
            lock.readLock().lock();//獲取讀取鎖定
            return (T) list[index];
        }finally{
            lock.readLock().unlock();//解除讀取鎖定
        }
    }
    public int size(){
        try{
            lock.readLock().lock();
            return next;
        }finally{
            lock.readLock().unlock();
        }
    }
}

重寫后的效果是

若有線程調用add()方法進行寫入操作，先獲得寫入鎖定。這時如果有其他線程準備獲得寫入鎖定或讀取鎖定，都必須等待前面的寫入鎖定解除。

若有線程調用get()方法進行讀取操作，先獲得讀取鎖定。這時如果有其他線程準備獲得讀取鎖定，則可以獲得；但如果是準備獲得寫入鎖定，仍然要等待所有讀取鎖定解除。

使用ReadWriteLock的好處在于，如果有兩個線程都想調用get()和size()方法，由于鎖定的關系，其中一個線程只能等到另一個線程解除鎖定。然而，這兩個方法都只是讀取對象狀態，如果只是讀取操作，就可以允許線程并行，這樣讀取效率將會提高。

Condition

Condition接口用來搭配Lock,最基本的用法就是達到Object的wait(),notify(),notifyAll()方法的作用。
下面用wait(),notify(),notifyAll()實現生產者與消費者.

店員從生產者獲得生產出的商品，消費者從店員取走商品

若生產者生產速度較快，店員那可能有很多商品，店員會叫生產者停一下。過一段時間，店員那商品不多了，再通知生產者繼續生產

若消費者取走速度過快，店員那可能沒有商品可供取走，店員會叫消費者停一下。過一段時間，店員那有商品了，再通知消費者過來取

這里假定店員那最多只能放一件商品

public class Producer implements Runnable{
    private Clerk clerk;
    public Producer(Clerk clerk){
        this.clerk=clerk;
    }
    @Override
    public void run() {
        for(int i=0;i<10;i++){
            try {
                Thread.sleep((int)Math.random()*3000);
            } catch (InterruptedException e) {
                
            }
            clerk.setProduct(i);
        }
    }
}

public class Consumer implements Runnable{
    private Clerk clerk;
    public Consumer(Clerk clerk){
        this.clerk=clerk;
    }
    @Override
    public void run() {
        for(int i=0;i<10;i++){
            try {
                Thread.sleep((int)Math.random()*3000);
            } catch (InterruptedException e) {
                
            }
            clerk.getProduct();
        }
    }
}

public class Clerk extends Thread{
    private int product=-1;//沒有商品
    public synchronized void setProduct(int product){
        while(this.product!=-1){
            try {
                wait();//店員那有商品，生產者停一下
            } catch (InterruptedException e) {
                
            }
        }
        this.product=product;
        System.out.println("生產者生產商品"+this.product);
        notify();//通知等待集合(喚醒的可能是消費者，也可能是生產者)
    }
    public synchronized int getProduct(){
        while(this.product==-1){
            try {
                wait();//店員沒有商品，消費者停一下
            } catch (InterruptedException e) {
                
            }
        }
        int p=this.product;
        System.out.println("消費者消費商品"+this.product);
        this.product=-1;//商品已經被取走
        notify();
        return p;
    }
    public static void main(String[] args){
        Clerk clerk=new Clerk();
        new Thread(new Producer(clerk)).start();
        new Thread(new Consumer(clerk)).start();
    }
}

生產者生產商品0
消費者消費商品0
生產者生產商品1
消費者消費商品1
生產者生產商品2
消費者消費商品2
生產者生產商品3
消費者消費商品3
生產者生產商品4
消費者消費商品4
生產者生產商品5
消費者消費商品5
生產者生產商品6
消費者消費商品6
生產者生產商品7
消費者消費商品7
生產者生產商品8
消費者消費商品8
生產者生產商品9
消費者消費商品9

現在用Condition接口重寫

public class Clerk {
    private int product=-1;//沒有商品
    Lock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();
    public void setProduct(int product){
        try{
            lock.lock();
            while(this.product!=-1){
                try {
                    condition.await();//店員那有商品，生產者停一下
                } catch (InterruptedException e) {
                    
                }
            }
            this.product=product;
            System.out.println("生產者生產商品"+this.product);
            condition.signal();//通知等待集合(喚醒的可能是消費者，也可能是生產者)
        }finally{
            lock.unlock();
        }
    }
    public int getProduct(){
        try{
            lock.lock();
            while(this.product==-1){
                try {
                    condition.await();//店員沒有商品，消費者停一下
                } catch (InterruptedException e) {
                    
                }
            }
            int p=this.product;
            System.out.println("消費者消費商品"+this.product);
            this.product=-1;//商品已經被取走
            condition.signal();
            return p;
        }finally{
            lock.unlock();
        }
    }
    public static void main(String[] args){
        Clerk clerk=new Clerk();
        new Thread(new Producer(clerk)).start();
        new Thread(new Consumer(clerk)).start();
    }
}

注意在多個生產者，消費者線程的情況下，等待集合中兩者都會有，而condition.signal()從等待集合中喚醒的具體對象是不確定的。有可能消費者取走商品后，喚醒的還是消費者，這時，消費者又會執行while循環，進入等待集合。
事實上，一個Condition對象可以表示一個等待集合。這樣上面例子，可以有兩個等待集合，一個給消費者用，一個給生產者用。生產者只會通知消費者的等待集合，消費者也只會通知生產者的等待集合。這樣效率會高些。

public class Clerk {
    ...
    private Condition producerCondition=lock.newCondition();//生產者的等待集合
    private Condition consumerCondition=lock.newCondition();//消費者的等待集合
    public void setProduct(int product){
        try{
            lock.lock();
            while(this.product!=-1){
                try {
                    producerCondition.await();//店員那有商品，生產者停一下
                } catch (InterruptedException e) {
                    
                }
            }
            this.product=product;
            System.out.println("生產者生產商品"+this.product);
            consumerCondition.signal();//喚醒消費者等待集合
        }finally{
            lock.unlock();
        }
    }
    public int getProduct(){
        try{
            lock.lock();
            while(this.product==-1){
                try {
                    consumerCondition.await();//店員沒有商品，消費者停一下
                } catch (InterruptedException e) {
                    
                }
            }
            int p=this.product;
            System.out.println("消費者消費商品"+this.product);
            this.product=-1;//商品已經被取走
            producerCondition.signal();//喚醒生產者等待集合
            return p;
        }finally{
            lock.unlock();
        }
    }
    ...
}

Executor

定義Executor接口的目的是將Runnable的指定與如何執行分離。它只定義了一個execute()方法。

public class Page{
    private Executor executor;
    public Page(Executor executor){
        this.executor=executor;
    }
    ...
    public void method1(){
        ...
        executor.execute(new Runnable(){
            @Override
            public void run(){
                ...
            }
        });
        ...
    }

}

public class DirectExecutor implements Executor{
    public void execute(Runnable r){
        r.run();
    }
}

調用

new Page(new DirectExecutor()).method1();

Executor api

ThreadPoolExecutor

像線程池這類服務，實際上是定義在Executor接口的子接口ExecutorService中。通用的ExecutorService由抽象類AbstractExecutorService操作，如果需要線程池功能，可以使用其子類ThreadPoolExecutor.
重寫上面executor例子

ExecutorService executorService=Executors.newCachedThreadPool();
new Page(executorService).method1();
executorService.shutdown();//在指定執行的Runnable都完成后，將ExecutorService關閉

Future與Callable

ExecutorService還定義了submit(),invokeAll(),invokeAny()等方法，這些方法出現在java.util.concurrent.Future,java.util.concurrent.Callable接口
Future定義的行為就是讓你在將來取得結果。你可以將想執行的工作交給Future,Future會使用另一個線程處理，你可以先做別的事情。過些時候，再調用Future的get()獲得結果。
如果結果已經產生，get()會直接返回，否則會進入阻塞狀態直到結果返回。get()的另一種重載方法可以指定等待結果的時間，若指定時間內結果還沒產生，則拋出TimeoutException異常。也可以使用Future的isDone()方法看結果是否產生。
Future經常與Callable一起使用，Callable的作用與Runnable相似，都是用來定義執行的流程。

Runnable的run()方法無返回值，也無法拋出異常

Callable的call()方法可以有返回值，也可以拋出異常

FutureTask是Future的操作類，創建時可傳入Callable對象指定執行的流程

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class FutureTaskDemo {
    public static int fib(int n){
        return n<=1?n:fib(n-1)+fib(n-2);
    }
    public static void main(String[] args){
        FutureTask task=new FutureTask(
                new Callable(){
                    @Override
                    public Integer call() throws Exception {
                        return fib(30);
                    }    
                }
        );
        new Thread(task).start();
        try {
            Thread.sleep(3000);
            System.out.println(task.get());
        } catch (InterruptedException|ExecutionException e) {
            
        }
    }
}

FutureTask構造類

FutureTask實現RunnableFuture接口，RunnableFuture接口繼承Runnable,Future接口。所以可以new Thread(task).
ExecutorService的submit()方法也可以接受Callable對象，調用后返回Future對象。

ExecutorService service=Executors.newCachedThreadPool();
Future future=service.submit(new Callable(){
    @Override
    public Integer call() throws Exception {
        return fib(30);
    }    
});

如果有多個Callable,可以先將它們收集到Collection中，然后調用ExecutorService的invokeAll()方法，返回List

如果有多個Callable,要求其中只要有一個執行完成就行了，則可以先將它們收集到Collection中，然后調用ExecutorService的invokeAny()方法

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor用來進行工作排程，其中的schedule()方法用來排定Runnable或Callable實例延遲多久執行一次，并返回Future子接口ScheduledFuture的實例。

import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledExecution {
    public static void main(String[] args){
        ScheduledExecutorService service=Executors.newSingleThreadScheduledExecutor();
        service.scheduleWithFixedDelay(new Runnable(){
            public void run(){
                System.out.println(new Date());
                try {
                    Thread.sleep(2000);//假設工作會執行2s
                } catch (InterruptedException e) {
                    
                }
            }
        }, 2000, 1000, TimeUnit.MILLISECONDS);
    }
}

Sat Oct 24 17:11:59 CST 2015
Sat Oct 24 17:12:02 CST 2015
Sat Oct 24 17:12:05 CST 2015
Sat Oct 24 17:12:08 CST 2015
Sat Oct 24 17:12:11 CST 2015

可以看到，輸出兩兩間相差3s.
scheduleWithFixedDelay()方法參數

如果把方法換成scheduleAtFixedRate()

Sat Oct 24 17:28:28 CST 2015
Sat Oct 24 17:28:30 CST 2015
Sat Oct 24 17:28:32 CST 2015
Sat Oct 24 17:28:34 CST 2015

每次排定的執行周期是1s,但是工作執行的時間是2s,會超過排定的執行周期，所以輸出兩兩間相差2s。

ForkJoinPool

Future的另一個操作類ForkJoinTask,與ExecutorService的另一個操作類ForkJoinPool有關，它們都是jdk7新增的api,用來解決分而治之的問題。

ForkJoinTask操作Future接口，可以在未來獲得耗時工作的執行結果

ForkJoinPool管理ForkJoinTask，調用fork()方法，可以讓另一個線程執行ForkJoinTask

如果要獲得ForkJoinTask的執行結果，可以調用join()方法。如果執行結果還沒產生，會阻塞直至有執行結果返回

使用ForkJoinTask的子類RecursiveTask,它是個抽象類，使用時必須繼承它，并操作compute()方法。

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class FibDemo extends RecursiveTask{
    final int n;
    FibDemo(int n){
        this.n=n;
    }
    public static int fib(int n){
        return n<=1?n:fib(n-1)+fib(n-2);
    }
    @Override
    protected Integer compute() {
        if(n<=10){
            return fib(n);
        }
        FibDemo f1=new FibDemo(n-1);
        f1.fork();//ForkJoinPool分配線程執行子任務
        FibDemo f2=new FibDemo(n-2);
        return f2.compute()+f1.join();//執行f2子任務+獲得f1子任務進行完成的結果
    }
    public static void main(String[] args){
        FibDemo fib=new FibDemo(40);
        ForkJoinPool pool=new ForkJoinPool();
        System.out.println(pool.invoke(fib));
    }
}

GPU云服務器云服務器讀書筆記編程思想讀書筆記大數據時代讀書筆記學習筆記

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/64701.html

<jdk7學習筆記>讀書筆記-線程

摘要：如果把注釋去掉，則在所以非線程都結束時，自動終止。默認所有從線程產生的線程也是線程。停止線程線程完成方法后，就進入狀態。被標示為的區塊會被監控，任何線程要執行區塊都必須先獲得指定的對象鎖定。 Tread和Runnable 定義線程實現Runnable接口，重寫run()方法繼承Thread類,重寫run()方法啟動線程 Runnable tortoise=ne...

woshicixide 2019-08-14 12:47 評論0 收藏0
《HTML與CSS 第一章認識HTML》讀書筆記

摘要：一讓廣播明星黯然失色要建立頁面，需要創建用超文本標記語言，編寫的文件，把它們放在一個服務器上二服務器能做什么服務器在互聯網上有一份全天候的工作。一、Web讓廣播明星黯然失色　　　　要建立Web頁面，需要創建用超文本標記語言（HyperText Markup Language，HTML）編寫的文件，把它們放在一個Web服務器上二、Web服務器能做什么？　　Web服務器在互聯網上有一份全天候的工...

番茄西紅柿 2019-04-23 13:34 評論0 收藏0
<javascript高級程序設計>第十二章讀書筆記----偏移量

摘要：包括元素的高度上下內邊距上下邊框值，如果元素的的值為那么該值為。該值為元素的包含元素。最后，所有這些偏移量都是只讀的，而且每次訪問他們都需要重新計算。為了避免重復計算，可以將計算的值保存起來，以提高性能。 offsetHeight 包括元素的高度、上下內邊距、上下邊框值，如果元素的style.display的值為none,那么該值為0。offsetWidth 包括元素的寬度、左...

dayday_up 2019-08-21 10:30 評論0 收藏0
<java核心技術>讀書筆記2

摘要：如果需要收集參數化類型對象，只有使用警告這節討論，向參數可變的方法傳遞一個泛型類型的實例。異常不能拋出或捕獲泛型類的實例實際上，泛型類擴展也是不合法的。 Object:所有類的超類 java中每個類都是由它擴展而來，但是并不需要這樣寫：class Employee extends Object.如果沒有明確指出超類，Object類就被認為是這個的超類。可以使用Object類型的變量引用...

jimhs 2019-08-14 12:45 評論0 收藏0
<java核心技術>讀書筆記1

摘要：關鍵字作用調用超類方法調用超類構造器關鍵字作用引用隱式參數如調用該類的其他構造器在覆蓋一個方法時，子類方法可見性不能低于超類方法阻止繼承類和方法目的確保它們不會在子類中改變語義。但是如果將一個類聲明為后面可以改變類變量的值了。數據類型整型 int 存儲要求：4byte 取值范圍：-2147483648 -- 2147483647(超過20億) short 存儲要求：2byte 取...

William_Sang 2019-08-14 12:43 評論0 收藏0