程序師世界是廣大編程愛好者互助、分享、學習的平台,程序師世界有你更精彩!
首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫知識
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
 程式師世界 >> 編程語言 >> JAVA編程 >> JAVA綜合教程 >> 【JAVA並發】同步工具類,java並發工具類

【JAVA並發】同步工具類,java並發工具類

編輯:JAVA綜合教程

【JAVA並發】同步工具類,java並發工具類


同步工具類主要包括閉鎖(如CountDownLatch),柵欄(如CyclicBarrier),信號量(如Semaphore)和阻塞隊列(如LinkedBlockingQueue)等;

使用同步工具類可以協調線程的控制流;

同步工具類封裝了一些狀態,這些狀態決定線程是繼續執行還是等待,此外同步工具類還提供了修改狀態的方法;

下面將簡單介紹以上同步工具類;

閉鎖

可以讓一個線程等待一組事件發生後(不一定要線程結束)繼續執行;

以CountDownLatch為例,內部包含一個計數器,一開始初始化為一個整數(事件個數),發生一個事件後,調用countDown方法,計數器減1,await用於等待計數器為0後繼續執行當前線程;

舉個例子如下,main線程等待其它子線程的事件發生後繼續執行main線程:

package concurrency;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

class TaskTest implements Runnable {

    private CountDownLatch latch;
    private int sleepTime;

    /**
     * 
     */
    public TaskTest(int sleepTime, CountDownLatch latch) {
        this.sleepTime = sleepTime;
        this.latch = latch;
    }

    /**
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
        try {
            CountDownLatchTest.print(" is running。");
            TimeUnit.MILLISECONDS.sleep(sleepTime);
            CountDownLatchTest.print(" finished。");
            //計數器減減
            latch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

public class CountDownLatchTest {
    public static void main(String[] args) {
        int count = 10;
        final CountDownLatch latch = new CountDownLatch(count);
        ExecutorService es = Executors.newFixedThreadPool(count);
        for (int i = 0; i < count; i++) {
            es.execute(new TaskTest((i + 1) * 1000, latch));
        }

        try {
            CountDownLatchTest.print(" waiting...");
            //主線程等待其它事件發生
            latch.await();
            //其它事件已發生,繼續執行主線程
            CountDownLatchTest.print(" continue。。。");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            es.shutdown();
        }
    }
    
    public static void print(String str){
        SimpleDateFormat dfdate = new SimpleDateFormat("HH:mm:ss");
        System.out.println("[" + dfdate.format(new Date()) + "]" + Thread.currentThread().getName() + str);
    }
}

結果打印如下:

[09:41:43]pool-1-thread-1 is running。
[09:41:43]pool-1-thread-6 is running。
[09:41:43]main waiting...
[09:41:43]pool-1-thread-10 is running。
[09:41:43]pool-1-thread-4 is running。
[09:41:43]pool-1-thread-5 is running。
[09:41:43]pool-1-thread-2 is running。
[09:41:43]pool-1-thread-3 is running。
[09:41:43]pool-1-thread-7 is running。
[09:41:43]pool-1-thread-8 is running。
[09:41:43]pool-1-thread-9 is running。
[09:41:44]pool-1-thread-1 finished。
[09:41:45]pool-1-thread-2 finished。
[09:41:46]pool-1-thread-3 finished。
[09:41:47]pool-1-thread-4 finished。
[09:41:48]pool-1-thread-5 finished。
[09:41:49]pool-1-thread-6 finished。
[09:41:50]pool-1-thread-7 finished。
[09:41:51]pool-1-thread-8 finished。
[09:41:52]pool-1-thread-9 finished。
[09:41:53]pool-1-thread-10 finished。
[09:41:53]main continue。。。

 此外,FutureTask也可用作閉鎖,其get方法會等待任務完成後返回結果,否則一直阻塞直到任務完成;

信號量

控制同時執行某個指定操作的數量,常用於實現資源池,如數據庫連接池,線程池...
以Semaphore為例,其內部維護一組資源,可以通過構造函數指定數目,其它線程在執行的時候,可以通過acquire方法獲取資源,有的話,繼續執行(使用結束後釋放資源),沒有資源的話將阻塞直到有其它線程調用release方法釋放資源;

舉個例子,如下代碼,十個線程競爭三個資源,一開始有三個線程可以直接運行,剩下的七個線程只能阻塞等到其它線程使用資源完畢才能執行;

package concurrency;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class SemaphoreTest {
    
    public static void print(String str){
        SimpleDateFormat dfdate = new SimpleDateFormat("HH:mm:ss");
        System.out.println("[" + dfdate.format(new Date()) + "]" + Thread.currentThread().getName() + str);
    }
    
    public static void main(String[] args) {
        // 線程數目
        int threadCount = 10;
        // 資源數目
        Semaphore semaphore = new Semaphore(3);
        
        ExecutorService es = Executors.newFixedThreadPool(threadCount);

        // 啟動若干線程
        for (int i = 0; i < threadCount; i++)
            es.execute(new ConsumeResourceTask((i + 1) * 1000, semaphore));
    }
}

class ConsumeResourceTask implements Runnable {
    private Semaphore semaphore;
    private int sleepTime;

    /**
         * 
         */
    public ConsumeResourceTask(int sleepTime, Semaphore semaphore) {
        this.sleepTime = sleepTime;
        this.semaphore = semaphore;
    }

    public void run() {
        try {
            //獲取資源
            semaphore.acquire();
            SemaphoreTest.print(" 占用一個資源...");
            TimeUnit.MILLISECONDS.sleep(sleepTime);
            SemaphoreTest.print(" 資源使用結束,釋放資源");
            //釋放資源
            semaphore.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
[10:30:11]pool-1-thread-1 占用一個資源...
[10:30:11]pool-1-thread-2 占用一個資源...
[10:30:11]pool-1-thread-3 占用一個資源...
[10:30:12]pool-1-thread-1 資源使用結束,釋放資源
[10:30:12]pool-1-thread-4 占用一個資源...
[10:30:13]pool-1-thread-2 資源使用結束,釋放資源
[10:30:13]pool-1-thread-5 占用一個資源...
[10:30:14]pool-1-thread-3 資源使用結束,釋放資源
[10:30:14]pool-1-thread-8 占用一個資源...
[10:30:16]pool-1-thread-4 資源使用結束,釋放資源
[10:30:16]pool-1-thread-6 占用一個資源...
[10:30:18]pool-1-thread-5 資源使用結束,釋放資源
[10:30:18]pool-1-thread-9 占用一個資源...
[10:30:22]pool-1-thread-8 資源使用結束,釋放資源
[10:30:22]pool-1-thread-7 占用一個資源...
[10:30:22]pool-1-thread-6 資源使用結束,釋放資源
[10:30:22]pool-1-thread-10 占用一個資源...
[10:30:27]pool-1-thread-9 資源使用結束,釋放資源
[10:30:29]pool-1-thread-7 資源使用結束,釋放資源
[10:30:32]pool-1-thread-10 資源使用結束,釋放資源

柵欄

柵欄用於等待其它線程,且會阻塞自己當前線程;

所有線程必須同時到達柵欄位置後,才能繼續執行;

舉個例子如下:

package concurrency;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

class CyclicBarrierTaskTest implements Runnable {
    private CyclicBarrier cyclicBarrier;

    private int timeout;

    public CyclicBarrierTaskTest(CyclicBarrier cyclicBarrier, int timeout) {
        this.cyclicBarrier = cyclicBarrier;
        this.timeout = timeout;
    }

    @Override
    public void run() {
        TestCyclicBarrier.print(" 正在running...");
        try {
            TimeUnit.MILLISECONDS.sleep(timeout);
            TestCyclicBarrier.print(" 到達柵欄處,等待其它線程到達");
            cyclicBarrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }

        TestCyclicBarrier.print(" 所有線程到達柵欄處,繼續執行各自線程任務...");
    }
}

public class TestCyclicBarrier {

    public static void print(String str) {
        SimpleDateFormat dfdate = new SimpleDateFormat("HH:mm:ss");
        System.out.println("[" + dfdate.format(new Date()) + "]"
                + Thread.currentThread().getName() + str);
    }

    public static void main(String[] args) {
        int count = 5;
        
        ExecutorService es = Executors.newFixedThreadPool(count);

        CyclicBarrier barrier = new CyclicBarrier(count, new Runnable() {

            @Override
            public void run() {
                TestCyclicBarrier.print(" 所有線程到達柵欄處,可以在此做一些處理...");
            }
        });
        for (int i = 0; i < count; i++)
            es.execute(new CyclicBarrierTaskTest(barrier, (i + 1) * 1000));
    }

}
[11:07:00]pool-1-thread-2 正在running...
[11:07:00]pool-1-thread-1 正在running...
[11:07:00]pool-1-thread-5 正在running...
[11:07:00]pool-1-thread-3 正在running...
[11:07:00]pool-1-thread-4 正在running...
[11:07:01]pool-1-thread-1 到達柵欄處,等待其它線程到達
[11:07:02]pool-1-thread-2 到達柵欄處,等待其它線程到達
[11:07:03]pool-1-thread-3 到達柵欄處,等待其它線程到達
[11:07:04]pool-1-thread-4 到達柵欄處,等待其它線程到達
[11:07:05]pool-1-thread-5 到達柵欄處,等待其它線程到達
[11:07:05]pool-1-thread-5 所有線程到達柵欄處,可以在此做一些處理...
[11:07:05]pool-1-thread-1 所有線程到達柵欄處,繼續執行各自線程任務...
[11:07:05]pool-1-thread-2 所有線程到達柵欄處,繼續執行各自線程任務...
[11:07:05]pool-1-thread-5 所有線程到達柵欄處,繼續執行各自線程任務...
[11:07:05]pool-1-thread-3 所有線程到達柵欄處,繼續執行各自線程任務...
[11:07:05]pool-1-thread-4 所有線程到達柵欄處,繼續執行各自線程任務...

阻塞隊列

阻塞隊列提供了可阻塞的入隊和出對操作,如果隊列滿了,入隊操作將阻塞直到有空間可用,如果隊列空了,出隊操作將阻塞直到有元素可用;

隊列可以為有界和無界隊列,無界隊列不會滿,因此入隊操作將不會阻塞;

下面將使用阻塞隊列LinkedBlockingQueue舉個生產者-消費者例子,生產者每隔1秒生產1個產品,然後有6個消費者在消費產品,可以發現,每隔1秒,只有一個消費者能夠獲取到產品消費,其它線程只能等待...

如下代碼:

package concurrency;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

//生產者
public class Producer implements Runnable {
    private final BlockingQueue<String> fileQueue;

    public Producer(BlockingQueue<String> queue) {
        this.fileQueue = queue;

    }

    public void run() {
        try {
            while (true) {
                TimeUnit.MILLISECONDS.sleep(1000);
                String produce = this.produce();
                System.out.println(Thread.currentThread() + "生產:" + produce);
                fileQueue.put(produce);
            }

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public String produce() {
        SimpleDateFormat dfdate = new SimpleDateFormat("HH:mm:ss");
        return dfdate.format(new Date());
    }

    public static void main(String[] args) {
        BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);

        for (int i = 0; i < 1; i++) {
            new Thread(new Producer(queue)).start();
        }
        for (int i = 0; i < 6; i++) {
            new Thread(new Consumer(queue)).start();
        }
    }
}

// 消費者
class Consumer implements Runnable {
    private final BlockingQueue<String> queue;

    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    public void run() {
        try {
            while (true) {
                TimeUnit.MILLISECONDS.sleep(1000);
                System.out.println(Thread.currentThread() + "prepare 消費");
                System.out.println(Thread.currentThread() + "starting:"
                        + queue.take());
                System.out.println(Thread.currentThread() + "end 消費");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
Thread[Thread-1,5,main]prepare 消費
Thread[Thread-3,5,main]prepare 消費
Thread[Thread-4,5,main]prepare 消費
Thread[Thread-2,5,main]prepare 消費
Thread[Thread-6,5,main]prepare 消費
Thread[Thread-5,5,main]prepare 消費
Thread[Thread-0,5,main]生產:11:36:36
Thread[Thread-1,5,main]starting:11:36:36
Thread[Thread-1,5,main]end 消費
Thread[Thread-1,5,main]prepare 消費
Thread[Thread-0,5,main]生產:11:36:37
Thread[Thread-4,5,main]starting:11:36:37
Thread[Thread-4,5,main]end 消費
Thread[Thread-4,5,main]prepare 消費
Thread[Thread-0,5,main]生產:11:36:38
Thread[Thread-3,5,main]starting:11:36:38
Thread[Thread-3,5,main]end 消費
...

 參考資料:java並發編程實戰

  1. 上一頁:
  2. 下一頁:
Copyright © 程式師世界 All Rights Reserved