Trong quá trình phát triển ứng dụng, chúng ta thường gặp các yêu cầu về xử lý tác vụ trễ như: tự động hủy đơn hàng nếu không thanh toán sau 30 phút, gửi thông báo nhắc lịch sau một khoảng thời gian nhất định, hoặc thay đổi trạng thái sự kiện khi đến giờ bắt đầu. Đối với các ứng dụng chạy đơn lẻ (monolithic), DelayQueue là một giải pháp hiệu quả được tích hợp sẵn trong bộ thư viện java.util.concurrent.

1. Định nghĩa đối tượng chứa dữ liệu (Payload)

Trước tiên, chúng ta cần một lớp để chứa thông tin về tác vụ cần xử lý. Ví dụ ở đây là thông tin về một sự kiện cần thay đổi trạng thái.

public class EventPayload {
    private int actionCode; // 1: Bắt đầu, 2: Kết thúc
    private String resourceId;

    public EventPayload(int actionCode, String resourceId) {
        this.actionCode = actionCode;
        this.resourceId = resourceId;
    }

    public int getActionCode() {
        return actionCode;
    }

    public String getResourceId() {
        return resourceId;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof EventPayload)) return false;
        EventPayload that = (EventPayload) o;
        return actionCode == that.actionCode && resourceId.equals(that.resourceId);
    }

    @Override
    public int hashCode() {
        return Objects.hash(actionCode, resourceId);
    }
}

2. Triển khai Interface Delayed

Để đưa một đối tượng vào DelayQueue, đối tượng đó phải triển khai interface Delayed. Lớp này sẽ chịu trách nhiệm tính toán thời gian chờ còn lại và xác định thứ tự ưu tiên trong hàng đợi.

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class ScheduledJob<T> implements Delayed {
    private final T data;
    private final long executionTime; // Thời điểm thực thi tính bằng milis

    public ScheduledJob(T data, long delayInMilliseconds) {
        this.data = data;
        this.executionTime = System.currentTimeMillis() + delayInMilliseconds;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long diff = executionTime - System.currentTimeMillis();
        return unit.convert(diff, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed other) {
        if (this.executionTime < ((ScheduledJob<?>) other).executionTime) {
            return -1;
        }
        if (this.executionTime > ((ScheduledJob<?>) other).executionTime) {
            return 1;
        }
        return 0;
    }

    public T getData() {
        return data;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) return true;
        if (obj instanceof ScheduledJob) {
            return this.data.equals(((ScheduledJob<?>) obj).getData());
        }
        return false;
    }
}

3. Quản lý và xử lý hàng đợi tập trung

Sử dụng một thành phần quản lý để duy trì hàng đợi và một luồng chạy ngầm để lấy các tác vụ đã hết thời gian chờ và xử lý chúng. Trong ví dụ này, chúng ta sử dụng Spring @Component để quản lý vòng đời của bộ lập lịch.

import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class JobSchedulerManager {

    private final DelayQueue<ScheduledJob<EventPayload>> masterQueue = new DelayQueue<>();
    private final ExecutorService workerPool = Executors.newSingleThreadExecutor();
    
    // Giả sử đây là service xử lý logic nghiệp vụ
    private final BusinessService businessService;

    public JobSchedulerManager(BusinessService businessService) {
        this.businessService = businessService;
    }

    public void schedule(ScheduledJob<EventPayload> job) {
        masterQueue.add(job);
    }

    public void cancel(ScheduledJob<EventPayload> job) {
        masterQueue.remove(job);
    }

    @PostConstruct
    public void startListening() {
        workerPool.execute(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // Phương thức take() sẽ chặn cho đến khi có phần tử hết hạn
                    ScheduledJob<EventPayload> expiredJob = masterQueue.take();
                    processTask(expiredJob.getData());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    private void processTask(EventPayload payload) {
        // Thực hiện logic nghiệp vụ (ví dụ: cập nhật database qua service)
        businessService.executeAction(payload.getResourceId(), payload.getActionCode());
    }
}

Lưu ý khi sử dụng

• Tính ổn định: Dữ liệu trong DelayQueue nằm trên RAM. Nếu ứng dụng bị restart, các tác vụ chưa xử lý sẽ bị mất. Do đó, khi khởi động ứng dụng, cần có cơ chế quét Database để nạp lại các tác vụ vào hàng đợi.
• Khả năng mở rộng: DelayQueue chỉ hoạt động trong phạm vi một máy chủ. Nếu hệ thống triển khai theo cụm (cluster), bạn nên cân nhắc sử dụng Redis (Sorted Set) hoặc các Message Queue hỗ trợ Delay (như RabbitMQ, RocketMQ) để đảm bảo tính đồng nhất.