Cách tối ưu hóa migrate dữ liệu lớn

// Kích thước hàng đợi
private static final int QUEUE_CAPACITY = 1000;

public static void main(String[] args) {
    // Hàng đợi dùng chung giữa Producer và Consumer
    BlockingQueue<String> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);

    // Thread Pool cho Producer và Consumer
    ExecutorService producerPool = Executors.newFixedThreadPool(2); // 2 Producers
    ExecutorService consumerPool = Executors.newFixedThreadPool(5); // 5 Consumers

    // Khởi chạy Producer Threads
    for (int i = 0; i < 2; i++) {
        producerPool.submit(new Producer(queue, i));
    }

    // Khởi chạy Consumer Threads
    for (int i = 0; i < 5; i++) {
        consumerPool.submit(new Consumer(queue, i));
    }

    // Đóng Thread Pools
    producerPool.shutdown();
    consumerPool.shutdown();

    try {
        producerPool.awaitTermination(1, TimeUnit.HOURS);
        consumerPool.awaitTermination(1, TimeUnit.HOURS);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }

    System.out.println("Data migration completed!");
}

static class Producer implements Runnable {
    private final BlockingQueue<String> queue;
    private final int producerId;

    public Producer(BlockingQueue<String> queue, int producerId) {
        this.queue = queue;
        this.producerId = producerId;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 100; i++) { // Giả lập đọc dữ liệu
                String data = "Merchant-" + producerId + "-" + i;
                queue.put(data); // Đẩy vào hàng đợi
                System.out.println("Producer " + producerId + " added: " + data);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

static class Consumer implements Runnable {
    private final BlockingQueue<String> queue;
    private final int consumerId;

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

    @Override
    public void run() {
        try {
            while (true) { // Lặp vô hạn (sẽ dừng khi producer kết thúc và queue trống)
                String data = queue.take(); // Lấy dữ liệu từ hàng đợi
                process(data); // Xử lý dữ liệu
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private void process(String data) {
        System.out.println("Consumer " + consumerId + " processing: " + data);
        try {
            Thread.sleep(100); // Giả lập thời gian xử lý dữ liệu
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}