《Java7併發編程實戰手冊》學習筆記（九）——測試併發應用程序

此篇博客爲我的學習筆記，也是Java併發編程學習第一部分的最後一篇，若有錯誤歡迎你們指正

本次內容

1. 監控Lock接口
2. 監控Phaser類
3. 監控執行器框架
4. 監控Fork/Join池
5. 輸出高效的日誌信息

1.監控Lock接口

咱們在併發程序中使用鎖進行同步時，能夠直接調用鎖對象的一些方法來判斷鎖當前的狀態。咱們也能夠經過繼承ReentrantLock類並在定製類的公開方法中調用父類中被protected關鍵字修飾的方法以此來達到某些需求。下面是ReentrantLock類實現的用於監控鎖的狀態的經常使用方法：

• getOwner()：返回當前持有鎖的線程對象，此方法被protected關鍵字修飾
• getQueuedThreads()：返回等待獲取鎖的線程集合，此方法一樣被protected關鍵字修飾
• hasQueuedThreads()：返回一個布爾值表示當前是否有線程在等待獲取鎖，true表示有
• getQueueLength()：返回當前等待獲取鎖的線程的數量
• isLocked()：返回一個布爾值表示這個鎖當前是否被一個線程持有，true表示被持有
• isFair()：返回一個布爾值表示這個鎖是否時公平鎖，true表示公平
• getHoldCount()：返回當前線程獲取該鎖的次數，這裏的次數實際是指鎖內部計數器的當前值
• isHeldByCurrentThread()：返回一個布爾值表示當前線程是否正持有該鎖

範例實現

在這個範例中，咱們調用了一些方法去監控鎖的狀態
MyLock（定製鎖，用於方便咱們調用ReentrantLock中的部分被保護方法）：

package day09.code_1;

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

public class MyLock extends ReentrantLock {

    //獲取當前持有鎖的線程的名稱
    public String getOwnerName() {
        //判斷當前是否有線程持有鎖
        if (getOwner() == null) {
            return "None";
        }
        //得到當前持有鎖的線程的名稱
        return getOwner().getName();
    }

    public Collection<Thread> getThreads() {
        //返回當前等待獲取鎖的線程集合
        return getQueuedThreads();
    }
}
複製代碼

Task（任務類，在此範例中非重點）：

package day09.code_1;

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

public class Task implements Runnable {

    private Lock lock;

    //經過構造方法初始化鎖
    public Task(Lock lock) {
        this.lock = lock;
    }

    @Override
    public void run() {
        //循環5次
        for (int i = 0; i < 5; i++) {
            //獲取鎖
            lock.lock();
            //打印當前線程持有鎖的信息
            System.out.printf("%s: Get the Lock\n",
                    Thread.currentThread().getName());
            try {
                //休眠500毫秒
                Thread.sleep(500);
                //打印當前線程釋放鎖的信息
                System.out.printf("%s: Free the Lock\n",
                        Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                //釋放鎖
                lock.unlock();
            }
        }
    }
}
複製代碼

main方法：

package day09.code_1;

import java.util.Collection;
import java.util.concurrent.TimeUnit;

public class Main {

    public static void main(String[] args) throws InterruptedException {
        //建立鎖對象
        MyLock lock = new MyLock();
        //建立線程數組
        Thread[] threads = new Thread[5];
        for (int i = 0; i < 5; i++) {
            //建立任務並將鎖傳入
            Task task = new Task(lock);
            //建立線程
            threads[i] = new Thread(task);
            //開啓線程
            threads[i].start();
        }
        //循環15次
        for (int i = 0; i < 15; i++) {
            //打印提示信息
            System.out.printf("Main: Logging the Lock\n");
            System.out.printf("**************************\n");
            //打印當前持有鎖的線程名稱
            System.out.printf("Lock: Owner : %s\n", lock.getOwnerName());
            //打印當前是否有線程正等待獲取鎖
            System.out.printf("Lock: Queued Threads: %s\n",
                    lock.hasQueuedThreads());
            //若是存在線程等待獲取鎖
            if (lock.hasQueuedThreads()) {
                //打印等待獲取鎖的線程的數量
                System.out.printf("Lock: Queue Length: %d\n",
                        lock.getQueueLength());
                //提示信息前綴
                System.out.printf("Lock: Queued Threads: ");
                //獲取等待獲取鎖的線程集合
                Collection<Thread> lockedThreads = lock.getThreads();
                //遍歷集合打印線程的名稱
                for (Thread lockedThread : lockedThreads) {
                    System.out.printf("%s ", lockedThread.getName());
                }
                //換行
                System.out.printf("\n");
            }
            //打印鎖的公平性
            System.out.printf("Lock: Fairness: %s\n", lock.isFair());
            //打印鎖是否被某個線程持有
            System.out.printf("Lock: Locked: %s\n", lock.isLocked());
            System.out.printf("**************************\n");
            //休眠1秒
            TimeUnit.SECONDS.sleep(1);
        }
    }
}
複製代碼

2.監控Phaser類

Phaser類是一個功能十分強大的線程同步輔助類，它不只能夠將併發任務統一地分階段執行，還能夠在程序中動態的更改phaser對象的任務註冊數。Phaser類也提供了許多方法來幫助開發者監控phaser對象的狀態，以下：

• getPhase()：返回phaser對象的當前階段（階段是從0開始的）
• getRegisteredParties()：返回phaser對象上所同步（註冊）的任務數
• getArrivedParties()：返回已結束當前階段並進行等待的任務數
• getUnarrivedParties()：返回未結束當前階段的任務數

範例實現

Task（任務類）：

package day09.code_2;

import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;

public class Task implements Runnable {

    //休眠的時間
    private int time;

    //phaser對象
    private Phaser phaser;

    //經過構造函數賦值
    public Task(int time, Phaser phaser) {
        this.time = time;
        this.phaser = phaser;
    }

    @Override
    public void run() {
        //通知phaser對象線程以完成當前階段並直接向下執行
        phaser.arrive();
        //打印進入第一階段提示信息
        System.out.printf("%s: Entering phase 1\n",
                Thread.currentThread().getName());
        //休眠
        try {
            TimeUnit.SECONDS.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //打印線程完成第一階段提示信息
        System.out.printf("%s: Finishing phase 1\n",
                Thread.currentThread().getName());
        //通知phaser對象線程以完成當前階段並等待
        phaser.arriveAndAwaitAdvance();
        //打印進入第二階段提示信息
        System.out.printf("%s: Entering phase 2\n",
                Thread.currentThread().getName());
        //休眠
        try {
            TimeUnit.SECONDS.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //打印線程完成第二階段提示信息
        System.out.printf("%s: Finishing phase 2\n",
                Thread.currentThread().getName());
        //通知phaser對象線程以完成當前階段並等待
        phaser.arriveAndAwaitAdvance();
        //打印進入第三階段提示信息
        System.out.printf("%s: Entering phase 3\n",
                Thread.currentThread().getName());
        //休眠
        try {
            TimeUnit.SECONDS.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //打印完成第三階段提示信息
        System.out.printf("%s: Finishing phase 3\n",
                Thread.currentThread().getName());
        //通知phaser對象線程以完成當前階段並取消註冊
        phaser.arriveAndDeregister();
    }
}
複製代碼

main方法：

package day09.code_2;

import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;

public class Main {

    public static void main(String[] args) throws InterruptedException {
        //建立phaser對象
        Phaser phaser = new Phaser(3);
        //循環建立三個任務並開啓線程運行它們
        for (int i = 0; i < 3; i++) {
            //經過循環次數爲其設置不一樣的休眠時間
            Task task = new Task(i + 1, phaser);
            Thread thread = new Thread(task);
            thread.start();
        }
        //循環十次
        for (int i = 0; i < 10; i++) {
            System.out.printf("********************\n");
            //打印提示信息
            System.out.printf("Main: Phaser Log\n");
            //打印phaser的當前階段
            System.out.printf("Main: Phaser: Phase: %d\n",
                    phaser.getPhase());
            //打印在phaser對象上註冊的任務數
            System.out.printf("Main: Phaser: Registered Parties: %d\n",
                    phaser.getRegisteredParties());
            //打印已結束當前階段的任務數
            System.out.printf("Main: Phaser: Arrived Parties: %d\n",
                    phaser.getArrivedParties());
            //打印未結束當前階段的任務數
            System.out.printf("Main: Phaser: Unarrived Parties: %d\n",
                    phaser.getUnarrivedParties());
            //休眠1秒
            TimeUnit.SECONDS.sleep(1);
        }
    }
}
複製代碼

3.監控執行器框架

ThreadPoolExecutor類爲咱們提供了以下方法來獲取執行器的狀態信息：

• getCorePoolSize()：獲取線程池的核心線程數
• getPoolSize()：獲取線程池中的實際線程數
• getActiveCount()：獲取線程池中正在執行任務的線程數
• getTaskCount()：獲取計劃執行的任務數，一般來講等於提交的任務數
• getCompletedTaskCount()：獲取已經執行完成的任務數
• isShutdown()：當調用執行器的shutdown()方法後，此方法的返回值會變爲true
• isTerminating()：當執行器正在關閉但還未完成時，次方法返回true
• isTerminated()：當執行器已經關閉時，此方法返回true

範例實現

Task（任務類）：

package day09.code_3;

import java.util.concurrent.TimeUnit;

public class Task implements Runnable {

    //休眠時間
    private long milliseconds;

    public Task(long milliseconds) {
        //經過構造函數爲休眠時間賦值
        this.milliseconds = milliseconds;
    }

    @Override
    public void run() {
        //打印任務開始執行的提示信息
        System.out.printf("%s: Begin\n",
                Thread.currentThread().getName());
        //休眠
        try {
            TimeUnit.MILLISECONDS.sleep(milliseconds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //打印任務執行結束的提示信息
        System.out.printf("%s: End\n",
                Thread.currentThread().getName());
    }
}
複製代碼

main方法：

package day09.code_3;

import java.util.Random;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Main {

    public static void main(String[] args) throws InterruptedException {
        //建立一個緩存線程池
        ThreadPoolExecutor executor = (ThreadPoolExecutor)
                Executors.newCachedThreadPool();
        //建立隨機數生成器
        Random random = new Random();
        //循環十次
        for (int i = 0; i < 10; i++) {
            //建立任務並將隨機數做爲其休眠時間
            Task task = new Task(random.nextInt(10000));
            //向執行器發送任務
            executor.submit(task);
        }
        //循環5次
        for (int i = 0; i < 5; i++) {
            //調用showLog方法打印線程池的信息
            showLog(executor);
            //休眠1秒
            TimeUnit.SECONDS.sleep(1);
        }
        //關閉線程池
        executor.shutdown();
        //循環5次
        for (int i = 0; i < 5; i++) {
            //調用showLog方法打印線程池的信息
            showLog(executor);
            //休眠1秒
            TimeUnit.SECONDS.sleep(1);
        }
        //等待線程池執行完全部任務後關閉
        executor.awaitTermination(1, TimeUnit.DAYS);
        //打印程序結束提示信息
        System.out.printf("Main: End of the program\n");
    }

    private static void showLog(ThreadPoolExecutor executor) {
        //準備打印線程池相關信息
        System.out.printf("***********************\n");
        System.out.printf("Main: Executor Log\n");
        //打印線程池的核心線程數
        System.out.printf("Main: Executor: Core Pool Size: %d\n",
                executor.getCorePoolSize());
        //打印線程池的實際線程數
        System.out.printf("Main: Executor: Pool Size: %d\n",
                executor.getPoolSize());
        //打印線程池中正在執行任務的線程數
        System.out.printf("Main: Executor: Active Count: %d\n",
                executor.getActiveCount());
        //打印提交的任務數（計劃執行的任務數）
        System.out.printf("Main: Executor: Task Count: %d\n",
                executor.getTaskCount());
        //打印執行器已執行完成的線程數
        System.out.printf("Main: Executor: Completed Task Count: %d\n",
                executor.getCompletedTaskCount());
        //打印執行器是否關閉
        System.out.printf("Main: Executor: Shutdown: %s\n",
                executor.isShutdown());
        //打印執行器是否正在終止
        System.out.printf("Main: Executor: Terminating: %s\n",
                executor.isTerminating());
        //打印執行器是否已經終止
        System.out.printf("Main: Executor: Terminated: %s\n",
                executor.isTerminated());
        System.out.printf("***********************\n");
    }
}
複製代碼

4.監控Fork/Join池

ForkJoinPool類爲咱們提供了以下方法來獲取線程池的狀態信息：

• getParallelism()：獲取線程池的並行級數
• getPoolSize()：獲取線程池內工做者線程的數量
• getActiveThreadCount()：獲取正在執行任務的線程數
• getRunningThreadCount()：獲取當前正在工做且未被任何機制阻塞（包括等待子任務執行完畢）的線程數
• getQueuedSubmissionCount()：獲取已提交但未被執行過的任務數
• getQueuedTaskCount()：獲取已提交且已開始執行的任務數
• hasQueuedSubmissions()：返回一個布爾值表示是否有未開始執行的任務等待
• getStealCount()：獲取竊取的工做數
• isTerminated()：返回一個布爾值表示線程池是否已經關閉

範例實現

Task（任務類）：

package day09.code_4;

import java.util.concurrent.RecursiveAction;

public class Task extends RecursiveAction {

    //數組
    private int[] array;

    //任務搜索的起始、終止 位置
    private int start, end;

    public Task(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected void compute() {
        //若是任務過大
        if (end - start > 100) {
            //進行拆解
            int mid = (start + end) / 2;
            //建立新任務
            Task task1 = new Task(array, start, mid);
            Task task2 = new Task(array, mid, end);
            //異步執行任務
            task1.fork();
            task2.fork();
            //等待任務執行結束
            task1.join();
            task2.join();
        } else {
            //在指定範圍內遍歷數組
            for (int i = start; i < end; i++) {
                //自增
                array[i]++;
                //休眠5毫秒
                try {
                    Thread.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
複製代碼

main方法：

package day09.code_4;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;

public class Main {

    public static void main(String[] args) throws InterruptedException {
        //建立線程池
        ForkJoinPool pool = new ForkJoinPool();
        //建立數組
        int[] array = new int[10000];
        //建立任務
        Task task = new Task(array, 0, 10000);
        //異步執行
        pool.execute(task);
        //在任務執行結束前不斷循環
        while (!task.isDone()) {
            //打印線程池的信息
            showLog(pool);
            //休眠1秒
            TimeUnit.SECONDS.sleep(1);
        }
        //關閉線程池
        pool.shutdown();
        //等待線程池執行完全部任務
        pool.awaitTermination(1, TimeUnit.DAYS);
        //打印線程池信息
        showLog(pool);
        //打印程序結束提示信息
        System.out.printf("Main: End of the program\n");
    }

    private static void showLog(ForkJoinPool pool) {
        //打印線程池提示信息
        System.out.printf("***********************\n");
        System.out.printf("Main: Fork/Join Pool log\n");
        //打印線程池並行級數
        System.out.printf("Main: Fork/Join Pool: Parallelism: %d\n",
                pool.getParallelism());
        //打印線程池內部實際線程數量
        System.out.printf("Main: Fork/Join Pool: Pool Size: %d\n",
                pool.getPoolSize());
        //打印正在執行任務的線程數
        System.out.printf("Main: Fork/Join Pool: Active Thread Count: %d\n",
                pool.getActiveThreadCount());
        //打印正在工做且未被阻塞的線程數
        System.out.printf("Main: Fork/Join Pool: Running Thread Count: %d\n",
                pool.getRunningThreadCount());
        //打印已提交但未被執行過的任務數
        System.out.printf("Main: Fork/Join Pool: Queued Submission: %d\n",
                pool.getQueuedSubmissionCount());
        //打印已提交且已開始執行的任務數
        System.out.printf("Main: Fork/Join Pool: Queued Tasks: %d\n",
                pool.getQueuedTaskCount());
        //打印一個布爾值表示是否有未開始執行的等待任務
        System.out.printf("Main: Fork/Join Pool: Queued Submissions: %s\n",
                pool.hasQueuedSubmissions());
        //打印任務竊取次數
        System.out.printf("Main: Fork/Join Pool: Steal Count: %d\n",
                pool.getStealCount());
        //打印線程池是否已經關閉
        System.out.printf("Main: Fork/Join Pool: Terminated: %s\n",
                pool.isTerminated());
        System.out.printf("***********************\n");
    }
}
複製代碼

5.輸出高效的日誌信息

在程序中，咱們應當輸出高效的日誌信息，而不僅是將信息打印到控制檯。Java爲咱們提供了Logger類來方便的輸出日誌信息。一個日誌器（Logger）主要有下面幾個組件：

1. Handler：處理器，處理器能夠決定日誌的寫入目標和格式
2. Name：名稱，一般狀況下日誌記錄器的名稱是其餘類的包名加類名
3. Level：日誌記錄器和日誌信息均關聯着一個級別，記錄器不會輸出級別更低的日誌信息

範例實現

MyFormatter（格式化類）：

package day09.code_5;

import java.util.Date;
import java.util.logging.Formatter;
import java.util.logging.LogRecord;

public class MyFormatter extends Formatter {
    @Override
    public String format(LogRecord record) {
        //建立字符串構造器
        StringBuilder sb = new StringBuilder();
        //拼接日誌級別
        sb.append("[" + record.getLevel() + "] - ");
        //拼接日誌生成時間
        sb.append(new Date(record.getMillis()) + " ");
        //拼接產生日誌的類名和方法名
        sb.append(record.getSourceClassName() + " . " + record.getSourceMethodName());
        //拼接日誌信息和換行符
        sb.append(" " + record.getMessage() + "\n");
        //返回相應的字符串
        return sb.toString();
    }
}
複製代碼

MyLogger（自定義日誌生成器）：

package day09.code_5;

import java.io.IOException;
import java.util.logging.FileHandler;
import java.util.logging.Handler;
import java.util.logging.Level;
import java.util.logging.Logger;

public class MyLogger {

    //處理器
    private static Handler handler;

    //用於獲取日誌生成器的靜態方法
    public static Logger getLogger(String name) {
        //調用Logger類的靜態方法獲取日誌生成器
        Logger logger = Logger.getLogger(name);
        //設置日誌級別未ALL，輸出一切等級的日誌
        logger.setLevel(Level.ALL);
        try {
            //若是沒有處理器
            if (handler == null) {
                //建立文檔處理器關聯相關文件
                handler = new FileHandler("src/day09/code_5/recipe8.log");
                //建立格式化工具並賦值給處理器
                MyFormatter format = new MyFormatter();
                handler.setFormatter(format);
            }
            //若是日誌生成器沒有處理器，則添加
            if (logger.getHandlers().length == 0) {
                logger.addHandler(handler);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        //返回日誌生成器
        return logger;
    }

}
複製代碼

Task（任務類）：

package day09.code_5;

import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

public class Task implements Runnable {

    @Override
    public void run() {
        //調用靜態方法獲取得日誌生成器，並將當前類名做爲參數傳入
        Logger logger = MyLogger.getLogger(this.getClass().getName());
        //輸出FINER級別的消息表示方法開始執行
        logger.entering(Thread.currentThread().getName(), "run()");
        //休眠兩秒
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //輸出FINER級別的消息表示方法執行結束
        logger.exiting(Thread.currentThread().getName(), "run()",
                Thread.currentThread());
    }
}
複製代碼

main方法：

package day09.code_5;

import java.util.logging.Level;
import java.util.logging.Logger;

public class Main {

    public static void main(String[] args) {
        //經過靜態方法獲取日誌生成器
        Logger logger = MyLogger.getLogger("Core");
        //輸出FINER級別的消息表示方法開始執行
        logger.entering("Core", "main()", args);
        //建立數組
        Thread[] threads = new Thread[5];
        //遍歷數組
        for (int i = 0; i < threads.length; i++) {
            //輸出INFO級別的日誌
            logger.log(Level.INFO, "Launching thread: " + i);
            //建立任務和線程對象
            Task task = new Task();
            threads[i] = new Thread(task);
            //輸出INFO級別的日誌
            logger.log(Level.INFO, "Thread created: " + threads[i].getName());
            //開啓線程
            threads[i].start();
        }
        //輸出日誌
        logger.log(Level.INFO, "Five Threads created.");
        logger.log(Level.INFO, "Waiting for its finalization");
        //等待任務運行結束
        for (int i = 0; i < threads.length; i++) {
            try {
                threads[i].join();
                //打印日誌
                logger.log(Level.INFO, "Thread has finished its execution", threads[i]);
            } catch (InterruptedException e) {
                //打印出現異常的日誌
                logger.log(Level.SEVERE, "Exeception", e);
            }

        }
        //打印方法結束的日誌
        logger.exiting("Core", "main()");
    }

}
複製代碼