Giao tiếp giữa các luồng trong Java

Giao tiếp giữa các luồng

Trong bài toán sản xuất - tiêu thụ, hai luồng (sản xuất và tiêu thụ) cùng truy cập một tài nguyên chung và phụ thuộc lẫn nhau. Cụ thể:

Luồng sản xuất phải thông báo cho luồng tiêu thụ khi có dữ liệu mới.
Luồng tiêu thụ sau khi xử lý xong cần báo hiệu để luồng sản xuất tiếp tục tạo dữ liệu.
Từ khóa synchronized giúp đồng bộ truy cập tài nguyên nhưng không hỗ trợ cơ chế giao tiếp giữa các luồng.

Java cung cấp hai cách phổ biến để giải quyết vấn đề này:

1. Phương pháp vùng đệm (Monitor/Buffer)

Sử dụng một vùng đệm trung gian để lưu trữ dữ liệu. Luồng sản xuất thêm vào vùng đệm, luồng tiêu thụ lấy ra từ đó. Khi vùng đệm đầy, luồng sản xuất phải chờ; khi rỗng, luồng tiêu thụ phải chờ.

Xem mã ví dụ

public class BufferExample {
    public static void main(String[] args) {
        SharedBuffer buffer = new SharedBuffer();
        new Producer(buffer).start();
        new Consumer(buffer).start();
    }
}

class Item {
    int value;
    public Item(int value) {
        this.value = value;
    }
}

class Producer extends Thread {
    private SharedBuffer buffer;
    public Producer(SharedBuffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        for (int i = 0; i < 50; i++) {
            buffer.put(new Item(i));
            System.out.println("Đã sản xuất: " + i);
        }
    }
}

class Consumer extends Thread {
    private SharedBuffer buffer;
    public Consumer(SharedBuffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        for (int i = 0; i < 50; i++) {
            Item item = buffer.take();
            System.out.println("Đã tiêu thụ: " + item.value);
        }
    }
}

class SharedBuffer {
    private final Item[] items = new Item[10];
    private int size = 0;

    public synchronized void put(Item item) {
        while (size == items.length) {
            try {
                wait(); // Đợi nếu buffer đầy
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        items[size++] = item;
        notifyAll(); // Thông báo có dữ liệu mới
    }

    public synchronized Item take() {
        while (size == 0) {
            try {
                wait(); // Đợi nếu buffer rỗng
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        Item item = items[--size];
        notifyAll(); // Thông báo đã giải phóng chỗ trống
        return item;
    }
}

2. Phương pháp cờ hiệu (Flag-based signaling)

Sử dụng một biến cờ (flag) để điều phối luân phiên giữa hai luồng. Một luồng chỉ được thực hiện khi cờ ở trạng thái phù hợp, sau đó đổi trạng thái cờ và đánh thức luồng kia.

Xem mã ví dụ

public class SignalingExample {
    public static void main(String[] args) {
        PerformanceStage stage = new PerformanceStage();
        new Performer(stage).start();
        new Viewer(stage).start();
    }
}

class Performer extends Thread {
    private PerformanceStage stage;
    public Performer(PerformanceStage stage) {
        this.stage = stage;
    }

    @Override
    public void run() {
        for (int i = 0; i < 5; i++) {
            stage.perform("Chương trình " + i);
        }
    }
}

class Viewer extends Thread {
    private PerformanceStage stage;
    public Viewer(PerformanceStage stage) {
        this.stage = stage;
    }

    @Override
    public void run() {
        for (int i = 0; i < 5; i++) {
            stage.watch();
        }
    }
}

class PerformanceStage {
    private String content;
    private boolean isPerforming = true; // true: đang biểu diễn, false: đang xem

    public synchronized void perform(String program) {
        while (!isPerforming) {
            try {
                wait();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        content = program;
        System.out.println("Biểu diễn: " + program);
        isPerforming = false;
        notifyAll();
    }

    public synchronized void watch() {
        while (isPerforming) {
            try {
                wait();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        System.out.println("Xem: " + content);
        isPerforming = true;
        notifyAll();
    }
}

Thẻ: Java multithreading thread-communication synchronized wait-notify

Đăng vào ngày 30 tháng 5 lúc 10:28

Thành phố Cuồng loạn

Giao tiếp giữa các luồng trong Java