Đa luồng và lập trình đồng thời

// ① Kế thừa Thread — ❌ ít dùng, tốn "suất extends" duy nhất
class LuongCuaToi extends Thread {
    @Override
    public void run() { System.out.println("Chạy trong " + getName()); }
}
new LuongCuaToi().start();

// ② Cài đặt Runnable — ✅ tốt hơn, tách tác vụ khỏi cơ chế chạy
Runnable tacVu = () -> System.out.println("Chạy trong "
        + Thread.currentThread().getName());
new Thread(tacVu, "luong-1").start();

// ⚠️ start() vs run() — lỗi kinh điển của người mới
Thread t = new Thread(tacVu);
t.run();      // ❌ chạy TRONG LUỒNG HIỆN TẠI — chỉ là lời gọi method thường!
t.start();    // ✅ tạo luồng MỚI rồi gọi run() trong đó

// ⚠️ Gọi start() hai lần → IllegalThreadStateException

// --- Các thao tác cơ bản ---
t.setName("worker-1");          // ⭐ đặt tên — rất quan trọng khi gỡ lỗi
t.setDaemon(true);              // luồng nền: JVM không đợi nó khi thoát
t.setPriority(Thread.MAX_PRIORITY);   // ⚠️ gợi ý thôi, hệ điều hành có thể bỏ qua
t.join();                       // CHỜ luồng t kết thúc
t.join(5000);                   // chờ tối đa 5 giây
t.interrupt();                  // YÊU CẦU dừng (không ép buộc)
Thread.sleep(1000);              // tạm dừng luồng HIỆN TẠI

// ❌ ĐÃ BỊ LOẠI BỎ — không bao giờ dùng
// t.stop();     — dừng đột ngột, để lại trạng thái hỏng
// t.suspend();  — dễ gây deadlock
// t.resume();

class BoDem {
    private int dem = 0;
    public void tang() { dem++; }        // ⚠️ trông như MỘT thao tác...
    public int get()   { return dem; }
}

BoDem bd = new BoDem();
List<Thread> luong = new ArrayList<>();
for (int i = 0; i < 10; i++) {
    Thread t = new Thread(() -> {
        for (int j = 0; j < 10_000; j++) bd.tang();
    });
    luong.add(t);
    t.start();
}
for (Thread t : luong) t.join();

System.out.println(bd.get());
// Kỳ vọng: 100000
// Thực tế: 87342, 91205, 100000, 76891...  MỖI LẦN CHẠY MỘT KHÁC ⚠️

// ① Method đồng bộ — khoá trên chính object (this)
class BoDem {
    private int dem = 0;
    public synchronized void tang() { dem++; }      // ✅ an toàn
    public synchronized int get()   { return dem; }  // ⚠️ CŨNG cần synchronized!
}

// ② Method static đồng bộ — khoá trên Class object
public static synchronized void f() { }
// tương đương: synchronized (BoDem.class) { }

// ③ Khối đồng bộ — ⭐ ưu tiên: phạm vi khoá HẸP hơn
class BoDem2 {
    private int dem = 0;
    private final Object khoa = new Object();   // khoá RIÊNG

    public void tang() {
        chuanBiTonThoiGian();       // ✅ không khoá phần này
        synchronized (khoa) {
            dem++;                    // chỉ khoá đúng phần cần
        }
        donDep();                    // ✅ không khoá
    }
}

// ❌ Không có volatile — vòng lặp có thể chạy MÃI MÃI
class Worker implements Runnable {
    private boolean chay = true;

    public void run() {
        while (chay) { }          // JIT có thể tối ưu thành while(true)!
        System.out.println("Dừng");
    }
    public void dung() { chay = false; }
}
// Luồng chính gọi dung(), nhưng luồng worker có thể KHÔNG BAO GIỜ thấy —
// vì nó đã cache `chay` vào thanh ghi CPU.

// ✅ volatile: buộc đọc/ghi thẳng từ bộ nhớ chính
private volatile boolean chay = true;

final Object khoaA = new Object();
final Object khoaB = new Object();

new Thread(() -> {
    synchronized (khoaA) {
        Thread.sleep(100);
        synchronized (khoaB) { ... }   // chờ B
    }
}).start();

new Thread(() -> {
    synchronized (khoaB) {           // ⚠️ thứ tự NGƯỢC LẠI
        Thread.sleep(100);
        synchronized (khoaA) { ... }   // chờ A
    }
}).start();

// 💥 DEADLOCK: luồng 1 giữ A chờ B, luồng 2 giữ B chờ A → treo vĩnh viễn

// ✅ CÁCH 1: thứ tự khoá NHẤT QUÁN — dùng một tiêu chí ổn định
void chuyenTien(TaiKhoan tu, TaiKhoan den, long tien) {
    TaiKhoan truoc = tu.getId() < den.getId() ? tu : den;
    TaiKhoan sau   = tu.getId() < den.getId() ? den : tu;

    synchronized (truoc) {         // LUÔN khoá id nhỏ trước
        synchronized (sau) {
            tu.rut(tien);
            den.nap(tien);
        }
    }
}

// ✅ CÁCH 2: tryLock có timeout — tự bỏ cuộc thay vì treo
ReentrantLock lockA = new ReentrantLock();
ReentrantLock lockB = new ReentrantLock();

if (lockA.tryLock(1, TimeUnit.SECONDS)) {
    try {
        if (lockB.tryLock(1, TimeUnit.SECONDS)) {
            try { ... } finally { lockB.unlock(); }
        }
    } finally { lockA.unlock(); }   // ⚠️ unlock PHẢI trong finally
}

// ✅ CÁCH 3 (tốt nhất): TRÁNH KHOÁ HOÀN TOÀN
//    Dùng collection an toàn luồng, Atomic, hoặc object bất biến.


// --- Chẩn đoán deadlock đã xảy ra ---
// $ jps                    → tìm PID
// $ jstack <pid>           → tìm mục "Found one Java-level deadlock"
// Hoặc dùng jconsole / VisualVM (nút "Detect Deadlock")

import java.util.concurrent.*;

// ❌ Tự tạo luồng cho mỗi tác vụ — tốn kém và không kiểm soát được
for (int i = 0; i < 10_000; i++) {
    new Thread(() -> xuLy()).start();   // 10.000 luồng × 1 MB = 10 GB!
}

// ✅ Dùng pool: tái sử dụng luồng, giới hạn số lượng
try (ExecutorService pool = Executors.newFixedThreadPool(4)) {   // Java 19+: AutoCloseable
    for (int i = 0; i < 10_000; i++) {
        pool.submit(() -> xuLy());
    }
}   // tự động shutdown và chờ xong

// Cách cũ (trước Java 19)
ExecutorService pool = Executors.newFixedThreadPool(4);
try {
    pool.submit(() -> xuLy());
} finally {
    pool.shutdown();                              // không nhận tác vụ mới
    if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
        pool.shutdownNow();                       // ngắt các tác vụ còn chạy
    }
}

// ⭐ Sản xuất thật nên tự cấu hình, không dùng Executors.* trần
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    4,                                    // số luồng lõi
    10,                                   // số luồng tối đa
    60L, TimeUnit.SECONDS,                 // luồng thừa sống bao lâu
    new ArrayBlockingQueue<>(100),          // ⭐ hàng đợi CÓ GIỚI HẠN
    new ThreadPoolExecutor.CallerRunsPolicy()  // đầy thì luồng gọi tự chạy
);
// Hàng đợi có giới hạn + chính sách từ chối = áp lực ngược (backpressure),
// bảo vệ hệ thống khỏi sập vì quá tải.

// Runnable: không trả gì, không ném checked exception
Runnable r = () -> System.out.println("xong");

// Callable<V>: TRẢ VỀ giá trị và NÉM ĐƯỢC checked exception
Callable<Integer> c = () -> {
    Thread.sleep(1000);
    return 42;
};

try (ExecutorService pool = Executors.newFixedThreadPool(2)) {
    Future<Integer> f = pool.submit(c);

    lamViecKhac();                // chạy song song

    int kq = f.get();              // ⚠️ CHẶN cho tới khi xong
    int kq2 = f.get(5, TimeUnit.SECONDS);   // có timeout
    f.isDone();
    f.cancel(true);                 // true = ngắt luồng nếu đang chạy

    // Chạy nhiều tác vụ
    List<Future<Integer>> tatCa = pool.invokeAll(List.of(c, c, c));  // chờ TẤT CẢ
    Integer dauTien = pool.invokeAny(List.of(c, c, c));            // lấy cái xong ĐẦU TIÊN
}

// ⚠️ Hạn chế lớn của Future: get() CHẶN. Không kết hợp được nhiều Future
//    mà không chặn. Đó là lý do CompletableFuture ra đời.

// --- Tạo ---
CompletableFuture<String> f1 = CompletableFuture
    .supplyAsync(() -> goiApi("/nguoi-dung"));      // có kết quả
CompletableFuture<Void> f2 = CompletableFuture
    .runAsync(() -> ghiLog());                     // không kết quả

// --- ⭐ Nối chuỗi KHÔNG CHẶN ---
CompletableFuture<String> kq = CompletableFuture
    .supplyAsync(() -> layNguoiDung(id))         // bước 1
    .thenApply(NguoiDung::getEmail)               // biến đổi (đồng bộ)
    .thenApplyAsync(String::toLowerCase)          // biến đổi (luồng khác)
    .thenCompose(e -> kiemTraEmailAsync(e))      // nối CF khác (như flatMap)
    .exceptionally(ex -> {                       // xử lý lỗi
        log.error("Thất bại", ex);
        return "mac-dinh@example.com";
    });

// --- Kết hợp nhiều Future ---
CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> layTen());
CompletableFuture<Integer> b = CompletableFuture.supplyAsync(() -> layTuoi());

// Chờ CẢ HAI rồi gộp
a.thenCombine(b, (ten, tuoi) -> ten + " (" + tuoi + ")");

// Chờ TẤT CẢ
CompletableFuture.allOf(a, b).join();

// Lấy cái nào xong TRƯỚC
CompletableFuture.anyOf(a, b).join();

// --- Gọi 100 API song song rồi gom kết quả ---
List<CompletableFuture<String>> ds = urls.stream()
    .map(u -> CompletableFuture.supplyAsync(() -> tai(u)))
    .toList();

CompletableFuture.allOf(ds.toArray(CompletableFuture[]::new))
    .thenApply(v -> ds.stream().map(CompletableFuture::join).toList())
    .thenAccept(System.out::println);

// --- ConcurrentHashMap ---
ConcurrentHashMap<String, Integer> m = new ConcurrentHashMap<>();

// ⚠️ Từng thao tác an toàn, nhưng GHÉP hai thao tác thì KHÔNG
if (!m.containsKey(k)) m.put(k, 1);   // ❌ hai luồng có thể cùng vào

// ✅ Dùng thao tác nguyên tử có sẵn
m.putIfAbsent(k, 1);
m.merge(k, 1, Integer::sum);                    // đếm an toàn
m.computeIfAbsent(k, x -> tinhToanTonKem(x));   // chỉ một luồng tính

// --- Atomic: CAS, không chặn luồng ---
AtomicInteger dem = new AtomicInteger();
dem.incrementAndGet();
dem.addAndGet(5);
dem.compareAndSet(5, 10);            // nếu đang là 5 thì đặt thành 10
dem.updateAndGet(x -> x * 2);

AtomicReference<String> ref = new AtomicReference<>("a");
ref.compareAndSet("a", "b");

// LongAdder: nhanh hơn AtomicLong khi RẤT nhiều luồng cùng ghi
LongAdder tong = new LongAdder();
tong.increment();
tong.sum();

// --- BlockingQueue: mẫu producer-consumer ---
BlockingQueue<String> q = new LinkedBlockingQueue<>(100);

new Thread(() -> {                // producer
    q.put("việc");                  // CHẶN nếu hàng đợi đầy → áp lực ngược
}).start();

new Thread(() -> {                // consumer
    while (true) {
        String v = q.take();          // CHẶN nếu rỗng — không cần vòng lặp bận
        xuLy(v);
    }
}).start();

// Tạo một virtual thread
Thread vt = Thread.ofVirtual().name("vt-1").start(() -> {
    System.out.println(Thread.currentThread());
});
vt.join();

// ⭐ Executor: MỖI TÁC VỤ MỘT VIRTUAL THREAD
try (var pool = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 1_000_000; i++) {   // MỘT TRIỆU luồng!
        pool.submit(() -> {
            Thread.sleep(Duration.ofSeconds(1));   // giả lập chờ I/O
            return null;
        });
    }
}
// Chạy tốt trên máy thường. Với luồng nền tảng thì OutOfMemoryError ngay.


// --- Vì sao quan trọng: mô hình lập trình đơn giản trở lại ---

// ❌ Trước Java 21: muốn xử lý nhiều kết nối phải viết bất đồng bộ
CompletableFuture.supplyAsync(() -> layNguoiDung(id))
    .thenCompose(u -> layDonHangAsync(u))
    .thenApply(this::tinhTong)
    .exceptionally(e -> 0L);
// Khó đọc, khó gỡ lỗi, stack trace vô nghĩa

// ✅ Với virtual thread: viết ĐỒNG BỘ nhưng vẫn mở rộng được
try (var pool = Executors.newVirtualThreadPerTaskExecutor()) {
    pool.submit(() -> {
        var u  = layNguoiDung(id);      // chặn — nhưng chỉ chặn virtual thread
        var dh = layDonHang(u);         // luồng OS được giải phóng để làm việc khác
        return tinhTong(dh);
    });
}
// Dễ đọc, dễ gỡ lỗi, stack trace đầy đủ — mà vẫn xử lý được hàng triệu kết nối


// --- StructuredTaskScope (Java 21 xem trước): quản lý nhóm tác vụ ---
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    var nd  = scope.fork(() -> layNguoiDung(id));
    var don = scope.fork(() -> layDonHang(id));

    scope.join();                // chờ cả hai
    scope.throwIfFailed();       // một cái lỗi → huỷ cái kia, ném lỗi ra

    return new KetQua(nd.get(), don.get());
}