Java線程同步方法實(shí)例總結(jié)

2024-07-14 08:42:10

字體：大中小

供稿：網(wǎng)友

本文實(shí)例講述了Java線程同步方法。分享給大家供大家參考，具體如下：

1. Semaphore

1.1 二進(jìn)制Semaphore

Semaphore算是比較高級(jí)點(diǎn)的線程同步工具了，在許多其他語(yǔ)言里也有類(lèi)似的實(shí)現(xiàn)。Semaphore有一個(gè)最大的好處就是在初始化時(shí)，可以顯式的控制并發(fā)數(shù)。其內(nèi)部維護(hù)這一個(gè)c計(jì)數(shù)器，當(dāng)計(jì)數(shù)器小于等于0時(shí)，是不允許其他線程訪問(wèn)并發(fā)區(qū)域的，反之則可以，因此，若將并發(fā)數(shù)設(shè)置為1，則可以確保單一線程同步。下面的例子模擬多線程打印，每個(gè)線程提交打印申請(qǐng)，然后執(zhí)行打印，最后宣布打印結(jié)束，代碼如下：

import java.util.concurrent.Semaphore;public class Program{    public static void main(String[] agrs){        PrintQueue p=new PrintQueue();        Thread[] ths=new Thread[10];        for(int i=0;i<10;i++){            ths[i]=new Thread(new Job(p),"Thread"+i);        }        for(int i=0;i<10;i++){            ths[i].start();        }    }}class PrintQueue{    private Semaphore s;    public PrintQueue(){        s=new Semaphore(1);//二進(jìn)制信號(hào)量    }    public void printJob(Object document){        try{            s.acquire();            long duration=(long)(Math.random()*100);            System.out.printf("線程名：%s 睡眠：%d",Thread.currentThread().getName(),duration);            Thread.sleep(duration);        }        catch(InterruptedException e){            e.printStackTrace();        }        finally{            s.release();        }    }}class Job implements Runnable{    private PrintQueue p;    public Job(PrintQueue p){        this.p=p;    }    @Override    public void run(){        System.out.printf("%s:正在打印一個(gè)任務(wù)/n ",Thread.currentThread().getName());        this.p.printJob(new Object());        System.out.printf("%s:文件已打印完畢/n ",Thread.currentThread().getName());    }}

執(zhí)行結(jié)果如下：

Thread0:正在打印一個(gè)任務(wù)
Thread9:正在打印一個(gè)任務(wù)
Thread8:正在打印一個(gè)任務(wù)
Thread7:正在打印一個(gè)任務(wù)
Thread6:正在打印一個(gè)任務(wù)
Thread5:正在打印一個(gè)任務(wù)
Thread4:正在打印一個(gè)任務(wù)
Thread3:正在打印一個(gè)任務(wù)
Thread2:正在打印一個(gè)任務(wù)
Thread1:正在打印一個(gè)任務(wù)
線程名：Thread0 睡眠：32 Thread0:文件已打印完畢
線程名：Thread9 睡眠：44 Thread9:文件已打印完畢
線程名：Thread8 睡眠：45 Thread8:文件已打印完畢
線程名：Thread7 睡眠：65 Thread7:文件已打印完畢
線程名：Thread6 睡眠：12 Thread6:文件已打印完畢
線程名：Thread5 睡眠：72 Thread5:文件已打印完畢
線程名：Thread4 睡眠：98 Thread4:文件已打印完畢
線程名：Thread3 睡眠：58 Thread3:文件已打印完畢
線程名：Thread2 睡眠：24 Thread2:文件已打印完畢
線程名：Thread1 睡眠：93 Thread1:文件已打印完畢

可以看到，所有線程提交打印申請(qǐng)后，按照并發(fā)順序一次執(zhí)行，沒(méi)有任何并發(fā)沖突，誰(shuí)先獲得信號(hào)量，誰(shuí)就先執(zhí)行，其他剩余線程均等待。這里面還有一個(gè)公平信號(hào)與非公平信號(hào)之說(shuō)：基本上java所有的多線程工具都支持初始化的時(shí)候指定一個(gè)布爾變量，true時(shí)表明公平，即所有處于等待的線程被篩選的條件為“誰(shuí)等的時(shí)間長(zhǎng)就選誰(shuí)進(jìn)行執(zhí)行”，有點(diǎn)first in first out的感覺(jué)，而false時(shí)則表明不公平（默認(rèn)是不non-fairness），即所有處于等待的線程被篩選執(zhí)行是隨機(jī)的。這也就是為什么多線程往往執(zhí)行順序比較混亂的原因。

1.2 多重并發(fā)控制

若將上面的代碼改為s=new Semaphore(3);//即讓其每次可以并發(fā)3條線程，則輸出如下：

Thread0:正在打印一個(gè)任務(wù)
Thread9:正在打印一個(gè)任務(wù)
Thread8:正在打印一個(gè)任務(wù)
Thread7:正在打印一個(gè)任務(wù)
Thread6:正在打印一個(gè)任務(wù)
Thread5:正在打印一個(gè)任務(wù)
Thread3:正在打印一個(gè)任務(wù)
Thread4:正在打印一個(gè)任務(wù)
Thread2:正在打印一個(gè)任務(wù)
Thread1:正在打印一個(gè)任務(wù)
線程名：Thread9 睡眠：26線程名：Thread8 睡眠：46線程名：Thread0 睡眠：79 Thread9:文件已打印完畢
線程名：Thread7 睡眠：35 Thread8:文件已打印完畢
線程名：Thread6 睡眠：90 Thread7:文件已打印完畢
線程名：Thread5 睡眠：40 Thread0:文件已打印完畢
線程名：Thread3 睡眠：84 Thread5:文件已打印完畢
線程名：Thread4 睡眠：13 Thread4:文件已打印完畢
線程名：Thread2 睡眠：77 Thread6:文件已打印完畢
線程名：Thread1 睡眠：12 Thread1:文件已打印完畢
Thread3:文件已打印完畢
Thread2:文件已打印完畢

很明顯已經(jīng)并發(fā)沖突了。若要實(shí)現(xiàn)分組（每組3個(gè)）并發(fā)嗎，則每一組也要進(jìn)行同步，代碼修改如下：

import java.util.concurrent.Semaphore;import java.util.concurrent.locks.Lock;import java.util.concurrent.locks.ReentrantLock;public class Program{    public static void main(String[] agrs){        PrintQueue p=new PrintQueue();        Thread[] ths=new Thread[10];        for(int i=0;i<10;i++){            ths[i]=new Thread(new Job(p),"Thread"+i);        }        for(int i=0;i<10;i++){            ths[i].start();        }    }}class PrintQueue{    private Semaphore s;    private boolean[] freePrinters;    private Lock lock;    public PrintQueue(){        s=new Semaphore(3);//二進(jìn)制信號(hào)量        freePrinters=new boolean[3];        for(int i=0;i<3;i++){            freePrinters[i]=true;        }        lock=new ReentrantLock();    }    public void printJob(Object document){        try{            s.acquire();            int printerIndex=getIndex();                long duration=(long)(Math.random()*100);                System.out.printf("線程名：%s 睡眠：%d/n",Thread.currentThread().getName(),duration);                Thread.sleep(duration);                freePrinters[printerIndex]=true;//恢復(fù)信號(hào)，供下次使用        }        catch(InterruptedException e){            e.printStackTrace();        }        finally{            s.release();        }    }    //返回一個(gè)內(nèi)部分組后的同步索引    public int getIndex(){        int index=-1;        try{            lock.lock();            for(int i=0;i<freePrinters.length;i++){                if(freePrinters[i]){                    freePrinters[i]=false;                    index=i;                    break;                }            }     }     catch(Exception e){         e.printStackTrace();     }     finally{         lock.unlock();     }     return index;    }}class Job implements Runnable{    private PrintQueue p;    public Job(PrintQueue p){        this.p=p;    }    @Override    public void run(){        System.out.printf("%s:正在打印一個(gè)任務(wù)/n ",Thread.currentThread().getName());        this.p.printJob(new Object());        System.out.printf(" %s:文件已打印完畢/n ",Thread.currentThread().getName());    }}

其中getIndex()方法主要為了維護(hù)內(nèi)部分組后（支持并發(fā)3個(gè)）組內(nèi)數(shù)據(jù)的同步（用lock來(lái)同步）。

輸出如下：

Thread0:正在打印一個(gè)任務(wù)
Thread9:正在打印一個(gè)任務(wù)
Thread8:正在打印一個(gè)任務(wù)
Thread7:正在打印一個(gè)任務(wù)
Thread6:正在打印一個(gè)任務(wù)
Thread5:正在打印一個(gè)任務(wù)
Thread4:正在打印一個(gè)任務(wù)
Thread3:正在打印一個(gè)任務(wù)
Thread2:正在打印一個(gè)任務(wù)
Thread1:正在打印一個(gè)任務(wù)
線程名：Thread0 睡眠：82 打印機(jī)：0號(hào)
線程名：Thread8 睡眠：61 打印機(jī)：2號(hào)
線程名：Thread9 睡眠：19 打印機(jī)：1號(hào)
Thread9:文件已打印完畢
線程名：Thread7 睡眠：82 打印機(jī)：1號(hào)
Thread8:文件已打印完畢
線程名：Thread6 睡眠：26 打印機(jī)：2號(hào)
Thread0:文件已打印完畢
線程名：Thread5 睡眠：31 打印機(jī)：0號(hào)
Thread6:文件已打印完畢
線程名：Thread4 睡眠：44 打印機(jī)：2號(hào)
Thread7:文件已打印完畢
線程名：Thread3 睡眠：54 打印機(jī)：1號(hào)
Thread5:文件已打印完畢
線程名：Thread2 睡眠：48 打印機(jī)：0號(hào)
Thread4:文件已打印完畢
線程名：Thread1 睡眠：34 打印機(jī)：2號(hào)
Thread3:文件已打印完畢
Thread2:文件已打印完畢
Thread1:文件已打印完畢

2. CountDownLatch

CountDownLatch同樣也是支持多任務(wù)并發(fā)的一個(gè)工具。它主要用于“等待多個(gè)并發(fā)事件”，它內(nèi)部也有一個(gè)計(jì)數(shù)器，當(dāng)調(diào)用await()方法時(shí)，線程處于等待狀態(tài)，只有當(dāng)內(nèi)部計(jì)數(shù)器為0時(shí)才繼續(xù)(countDown()方法來(lái)減少計(jì)數(shù))，也就說(shuō)，假若有一個(gè)需求是這樣的：主線程等待所有子線程都到達(dá)某一條件時(shí)才執(zhí)行，那么只需要主線程await，然后在啟動(dòng)每個(gè)子線程的時(shí)候進(jìn)行countDown操作。下面模擬了一個(gè)開(kāi)會(huì)的例子，只有當(dāng)所有人員都到齊了，會(huì)議才能開(kāi)始。

import java.util.concurrent.CountDownLatch;public class Program{    public static void main(String[] agrs){        //開(kāi)啟可容納10人的會(huì)議室        VideoConference v=new VideoConference(10);        new Thread(v).start();        //參與人員陸續(xù)進(jìn)場(chǎng)        for(int i=0;i<10;i++){            Participant p=new Participant(i+"號(hào)人員",v);            new Thread(p).start();        }    }}class VideoConference implements Runnable{    private CountDownLatch controller;    public VideoConference(int num){        controller=new CountDownLatch(num);    }    public void arrive(String name){        System.out.printf("%s 已經(jīng)到達(dá)！/n",name);        controller.countDown();        System.out.printf("還需要等 %d 個(gè)成員！/n",controller.getCount());    }    @Override    public void run(){        try{            System.out.printf("會(huì)議正在初始化...!/n");            controller.await();            System.out.printf("所有人都到齊了，開(kāi)會(huì)吧!/n");        }        catch(InterruptedException e){            e.printStackTrace();        }    }}class Participant implements Runnable{    private VideoConference conference;    private String name;    public Participant(String name,VideoConference conference){        this.name=name;        this.conference=conference;    }    @Override    public void run(){        long duration=(long)(Math.random()*100);        try{            Thread.sleep(duration);            conference.arrive(this.name);     }     catch(InterruptedException e){     }    }}

輸出：

會(huì)議正在初始化...!
0號(hào)人員已經(jīng)到達(dá)！
還需要等 9 個(gè)成員！
1號(hào)人員已經(jīng)到達(dá)！
還需要等 8 個(gè)成員！
9號(hào)人員已經(jīng)到達(dá)！
還需要等 7 個(gè)成員！
4號(hào)人員已經(jīng)到達(dá)！
還需要等 6 個(gè)成員！
8號(hào)人員已經(jīng)到達(dá)！
還需要等 5 個(gè)成員！
5號(hào)人員已經(jīng)到達(dá)！
還需要等 4 個(gè)成員！
6號(hào)人員已經(jīng)到達(dá)！
還需要等 3 個(gè)成員！
3號(hào)人員已經(jīng)到達(dá)！
還需要等 2 個(gè)成員！
7號(hào)人員已經(jīng)到達(dá)！
還需要等 1 個(gè)成員！
2號(hào)人員已經(jīng)到達(dá)！
還需要等 0 個(gè)成員！
所有人都到齊了，開(kāi)會(huì)吧!

3. Phaser

import java.util.concurrent.Phaser;import java.util.concurrent.TimeUnit;import java.util.List;import java.util.ArrayList;import java.io.File;import java.util.Date;public class Program{    public static void main(String[] agrs){        Phaser phaser=new Phaser(3);        FileSearch system=new FileSearch("C://Windows", "log",phaser);        FileSearch apps=new FileSearch("C://Program Files","log",phaser);        FileSearch documents=new FileSearch("C://Documents And Settings","log",phaser);        Thread systemThread=new Thread(system,"System");        systemThread.start();        Thread appsThread=new Thread(apps,"Apps");        appsThread.start();        Thread documentsThread=new Thread(documents, "Documents");        documentsThread.start();        try {            systemThread.join();            appsThread.join();            documentsThread.join();            } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("Terminated: "+ phaser.isTerminated());    }}class FileSearch implements Runnable{    private String initPath;    private String end;    private List<String> results;    private Phaser phaser;    public FileSearch(String initPath,String end,Phaser phaser){        this.initPath=initPath;        this.end=end;        this.results=new ArrayList<String>();        this.phaser=phaser;    }    private void directoryProcess(File file){        File[] files=file.listFiles();        if(files!=null){            for(int i=0;i<files.length;i++){                if(files[i].isDirectory()){                    directoryProcess(files[i]);                }                else{                    fileProcess(files[i]);                }            }        }    }    private void fileProcess(File file){        if(file.getName().endsWith(end)){            results.add(file.getAbsolutePath());        }    }    private void filterResults(){        List<String> newResults=new ArrayList<String>();        long actualDate=new Date().getTime();        for(int i=0;i<results.size();i++){            File file=new File(results.get(i));            long fileDate=file.lastModified();            if(actualDate-fileDate<TimeUnit.MILLISECONDS.convert(1,TimeUnit.DAYS)){                newResults.add(results.get(i));            }        }        results=newResults;    }    private boolean checkResults(){        if(results.isEmpty()){            System.out.printf("%s: Phase %d: 0 results./n",Thread.currentThread().getName(),phaser.getPhase());            System.out.printf("%s: Phase %d: End./n",Thread.currentThread().getName(),phaser.getPhase());            phaser.arriveAndDeregister();        }        else{            System.out.printf("%s: Phase %d: %d results./n",Thread.currentThread().getName(),phaser.getPhase(),results.size());             phaser.arriveAndAwaitAdvance();            return true;        }    }    private void showInfo() {        for (int i=0; i<results.size(); i++){            File file=new File(results.get(i));            System.out.printf("%s: %s/n",Thread.currentThread().getName(),file.getAbsolutePath());        }        phaser.arriveAndAwaitAdvance();    }    @Override    public void run(){        File file=new File(initPath);        if(file.isDirectory()){            directoryProcess(file);        }        if(!checkResults()){            return;        }        filterResults();        if(!checkResults()){            return;        }        showInfo();        phaser.arriveAndDeregister();        System.out.printf("%s: Work completed./n",Thread.currentThread().getName());    }}

運(yùn)行結(jié)果：