Những điều cần biết về java.util.concurrent trong lập trình đa luồng phần 2

Trong java.util.concurrent cung cấp các tiện ích đồng bộ hoá giúp cho việc lập trình multithread dễ dàng hơn. Bài viết này sẽ giới thiệu một số ý tưởng đồng bộ hóa thread ở mức cao hơn so với thread đơn thuần, trong bài viết này sẽ hướng dẫn cơ bản về sử dụng và quản lý multithread tư tưởng giống như

- Ổ khoá: Trong một thời điểm chỉ có 1 thread được chạy
- Cửa chặn: Thread phải chờ đến khi thoả mãn điều kiện nào đó thì mới được chạy

1.Semaphore

Trong một hệ thống, không phải là hiếm các trường hợp khi các nhà phát triển cần phải điều tiết số lượng các yêu cầu đang được mở đối với một tài nguyên cụ thể. Thực tế, đôi khi việc điều tiết có thể cải thiện thông qua một hệ thống bằng cách giảm số lượng tranh chấp giành tài nguyên cụ thể đó. Mặc dù chắc chắn có thể viết mã bằng tay, thì việc sử dụng lớp semaphore còn dễ dàng hơn, do semaphore sẽ lo việc điều tiết cho bạn

import java.util.Random;
import java.util.concurrent.Semaphore;

public class SemaphoreDemo {
    public static void main(String[] args) {
        Runnable r = new Runnable() {
            final Random rand = new Random();
            final Semaphore available = new Semaphore(3);
            int count = 0;

            public void run() {
                int time = rand.nextInt(15);
                int num = count++;

                try {
                    available.acquire();

                    System.out.println("Executing " + "long-running action for " + time + " seconds... #" + num);

                    Thread.sleep(time * 1000);

                    System.out.println("Done with #" + num + "!");

                    available.release();
                } catch (InterruptedException intEx) {
                    intEx.printStackTrace();
                }
            }
        };

        for (int i = 0; i < 10; i++)
            new Thread(r).start();
    }
}

Kết quả thực hiện:

Executing long-running action for 7 seconds... #0
Executing long-running action for 14 seconds... #1
Executing long-running action for 7 seconds... #2
Done with #0!
Executing long-running action for 12 seconds... #3
Done with #2!
Executing long-running action for 9 seconds... #4
Done with #1!
Executing long-running action for 0 seconds... #5
Done with #5!
Executing long-running action for 14 seconds... #6
Done with #4!
Executing long-running action for 14 seconds... #7
Done with #3!
Executing long-running action for 6 seconds... #8
Done with #8!
Executing long-running action for 12 seconds... #9
Done with #6!
Done with #7!
Done with #9!

Mặc dù 10 Thread trong ví dụ này đang chạy(bạn có thể kiểm tra bằng lệnh jstack dựa vào tiến trình Java đang chạy SemaphoreDemo nhưng chỉ có 3 luồng đang hoạt động. Bảy thread khác đang được giữ trong ngăn cho đến khi một trong các bộ đếm của semaphore được giải phóng

2.CountDownLatch

Nếu semaphore là lớp được thiết kế để cho phép các luồng vào, mỗi cái một lần thì countdownlatch là cổng xuất phát của cuộc đua ngựa. Lớp này nắm giữ tất cả các luồng ở trong ngăn của nó cho đến khi có một điều kiện cụ thể được đáp ứng thì lúc đó nó sẽ giải phóng tất cả cùng một lúc

import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {
    public static void main(String args[]) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(4);
        Worker first = new Worker(1000, latch, "WORKER-1");
        Worker second = new Worker(2000, latch, "WORKER-2");
        Worker third = new Worker(3000, latch, "WORKER-3");
        Worker fourth = new Worker(4000, latch, "WORKER-4");

        first.start();
        second.start();
        third.start();
        fourth.start();

        // Main thread will wait until all thread finished
        latch.await();

        System.out.println(Thread.currentThread().getName() + " has finished");
    }
}

class Worker extends Thread {
    private int delay;
    private CountDownLatch latch;

    public Worker(int delay, CountDownLatch latch, String name) {
        super(name);
        this.delay = delay;
        this.latch = latch;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(delay);
            latch.countDown();
            System.out.println(Thread.currentThread().getName() + " has finished");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Kết quả thực hiện:

WORKER-1 has finished
WORKER-2 has finished
WORKER-3 has finished
WORKER-4 has finished
main has finished

Trong ví dụ trên CountDownLatch có nhiệm vụ đảm bảo cho toàn bộ các worker làm việc rồi thì mới cho bắt đầu chương trình. Giống như trong tạo 1 chương trình có nhiều service hoặc nhiều thread. Để đảm bảo toàn bộ thread đã khởi tạo xong thì chương trình mới khởi động. Nếu không dùng CountDownLatch thì kết quả sẽ như sau

main has finished
WORKER-1 has finished
WORKER-2 has finished
WORKER-3 has finished
WORKER-4 has finished

3.Executor

Trong các ví dụ trên có một thiếu sót, đó là chúng buộc bạn phải trực tiếp tạo các đối tượng Thread. Đây là cách làm mang lại rắc rối và tăng bộ nhớ. Nên việc sử dụng lại các Thread hiện có tốt hơn rất nhiều so với tạo ra 1 thread mới. Java cung cấp ExecutorService mô hình hoá việc tạo thread như là 1 dịch vụ có thể được điều khiển chung.

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

public class ExecutorServiceDemo {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        executorService.execute(new Runnable() {
            public void run() {
                System.out.println("Asynchronous task");
            }
        });

        executorService.shutdown();
    }
}

4.ScheduledExecutorServices

Dùng ExecutorService thật là tuyệt vời, nhưng một số thread cần phải thực hiện có lịch trình, chẳng hạn như việc thi hành 1 thread ở trong một thời gian xác định hoặc tại 1 thời điểm cụ thể thì chúng ta sử dụng ScheduledExecutorServices nó là mở rộng của ExecutorServices

Ví dụ nếu mục tiêu của bạn là tạo ra một lệnh heartbeat để ping 5s/lần thì ScheduledExecutorServices sẽ làm cho nó đơn giản như sau:

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

public class Heartbeat {
    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(1);
        Runnable pinger = new Runnable() {
            public void run() {
                System.out.println("PING!");
            }
        };
        ScheduledFuture<?> scheduledFuture = ses.scheduleAtFixedRate(pinger, 5, 5, TimeUnit.SECONDS);

        Thread.sleep(20000);

        scheduledFuture.cancel(true);
    }
}

Không còn phiền phức với Thread, cũng không phiền phức với những gì phải làm nếu người dùng muốn huỷ nhịp tim đó, không cần phải quan tâm đến thread chạy khi nào, bạn chỉ cần để lại chi tiết lịch trình cho ScheduledExecutorServices. Bỗng nhiên bạn muốn huỷ nó đi cũng rất đơn giản bạn chỉ cần gọi phương thức cancel

Kết luận:

Gói java.util.concurrent chứa nhiều tiện ích tốt hơn là mở rộng ra ngoài các bộ sưu tập, đặc biệt là trong các gói .locks và .atomic. Hãy đi sâu vào và bạn sẽ tìm thấy các cấu trúc điều khiển có ích như CyclicBarier và nhiều hơn nữa ...

Bất cứ khi nào bạn viết về đa luồng, hãy nhớ về các tiện ích đã đề cập trong bài viết này và bài viết trước

Thời gian tới, tôi sẽ chuyển sang chủ đề những nâng cấp java8 về concurrent.