บทที่ 12 — Concurrency (Thread, synchronized, volatile, Memory Model)

← บทที่ 11 | สารบัญ | บทที่ 13 →

📓 โซนอ้างอิง—เปิดตอนต้องใช้ (บท 11-19) บทนี้อยู่ในโซนที่ 2 และ เป็นหนึ่งในบทที่ยากที่สุดของเล่ม — เรื่องการทำงานหลายอย่างพร้อมกัน (concurrency) มีบั๊กที่หายากและศัพท์เทคนิคเยอะ มือใหม่ข้ามไปก่อนได้เต็มที่ ค่อยกลับมาตอนต้องเขียนโค้ดที่รันหลายงานพร้อมกันจริง ๆ (เช่น web server, งานเบื้องหลัง) ถ้าจะอ่าน อ่านช้า ๆ ทีละหัวข้อ

🛑 หยุดก่อน — ถ้าคุณกำลังเรียน Java พื้นฐาน หรือเพิ่งอ่านจบบทที่ 10 บทนี้พูดถึง JMM, happens-before, deadlock, virtual threads — ไม่จำเป็นสำหรับการเรียนพื้นฐานหรือเริ่ม Spring Boot กลับไปบทที่ 10 หรือกระโดดไป Spring Boot ก่อนได้เลย แล้วค่อยกลับมาอ่านบทนี้ตอนที่ต้องเขียนโค้ดที่รันหลาย thread จริง ๆ

ศัพท์ที่จะเจอบ่อยในบทนี้:
concurrency (คอนเคอร์เรนซี) = การทำงานหลายอย่างคาบเกี่ยวกัน
thread (เธรด) = เส้นการทำงานคู่ขนานในโปรแกรมเดียว
process (โพรเซส) = โปรแกรมที่กำลังรันอยู่
race condition = ภาวะแย่งกันเขียนข้อมูลจนผลลัพธ์เพี้ยน
JMM (Java Memory Model) = กฎที่ Java กำหนดว่าหลายเธรดเห็นค่าตัวแปรตรงกันเมื่อไหร่
happens-before = กฎลำดับ "เกิดก่อน-เห็นทีหลัง" ใน JMM
deadlock = สภาวะล็อกตาย ต่างฝ่ายต่างรอกันไม่จบ

ทำไมต้องเรียน? — โปรแกรมส่วนใหญ่ "หลายอย่างเกิดพร้อมกัน":

Web server รับ request พร้อมกัน 1000 connection
App download ไฟล์พร้อมกัน 5 ไฟล์
UI ไม่ค้าง ตอน background load

ทุกอย่างนี้ใช้ thread — และ thread ทำให้เกิด bug ที่หายากที่สุดในวงการ

หลังจบบท คุณจะ:

เข้าใจ thread, process ต่างกันยังไง
สร้าง thread + ใช้ ExecutorService
เข้าใจ race condition + ใช้ synchronized / volatile / AtomicInteger
รู้จัก Java Memory Model (happens-before) ระดับที่ใช้งานได้
รู้ว่าเมื่อไหร่ใช้ virtual threads vs platform threads

1. Process vs Thread

	Process	Thread
คืออะไร	โปรแกรมที่ run อยู่	"เส้น" การทำงานใน process
memory	แยก	share กับ thread อื่นใน process เดียวกัน
สร้างเร็ว	ช้า	เร็ว
สื่อสารกัน	ยาก (IPC = Inter-Process Communication เช่น pipe, socket)	ง่าย (variable เดียวกัน) — และอันตราย

แอป Java 1 ตัว = 1 process ที่มีหลาย thread

2. สร้าง Thread แรก

java

public class Hello {
    public static void main(String[] args) {
        Thread t = new Thread(() -> {
            System.out.println("Hello from " + Thread.currentThread().getName());
        });

        t.start();          // ✅ เริ่ม thread ใหม่
        // t.run();         // ❌ ไม่ใช่ — run() จะรันบน thread ปัจจุบัน

        System.out.println("Main: " + Thread.currentThread().getName());
    }
}

Output ออกมาไม่แน่ — Main อาจมาก่อนหรือหลัง Hello:

text

Main: main
Hello from Thread-0

// หรือ

Hello from Thread-0
Main: main

นี่คือ non-determinism (ผลลัพธ์ที่เดาไม่ได้แน่นอน) — order ของ thread ขึ้นอยู่กับ OS scheduler (ตัวจัดคิวการทำงานของ thread โดย OS)

`.join()` — รอ thread จบ

java

Thread t = new Thread(() -> longTask());
t.start();
t.join();                    // ✅ block main จนกว่า t จะเสร็จ
System.out.println("done");

3. ⚠️ อย่าสร้าง Thread เอง — ใช้ ExecutorService

สร้าง Thread เอง = สร้าง resource ใหม่ทุกครั้ง → แพง

ใช้ thread pool แทน:

java

import java.util.concurrent.*;

ExecutorService pool = Executors.newFixedThreadPool(4);     // 4 thread reuse

for (int i = 0; i < 100; i++) {
    int taskId = i;
    pool.submit(() -> {
        System.out.println("Task " + taskId + " on " + Thread.currentThread().getName());
    });
}

pool.shutdown();                                              // ไม่รับงานใหม่
pool.awaitTermination(1, TimeUnit.MINUTES);                   // รอจนเสร็จ

Thread pool ที่ใช้บ่อย

java

Executors.newFixedThreadPool(10);          // ขนาดคงที่
Executors.newCachedThreadPool();           // ขยายตามภาระ (สำหรับงานสั้น ๆ) ⚠️ ระวัง: ไม่มี cap (จำกัดขนาด) — load spike (โหลดพุ่งสูง) อาจสร้าง thread หลายพัน ทำ OOM ได้ ไม่แนะนำ production โดยไม่มี bounded pool (pool ที่จำกัดขนาดสูงสุด)
Executors.newSingleThreadExecutor();       // 1 thread (queue งาน)
Executors.newScheduledThreadPool(2);       // งานตั้งเวลา (delay/repeat) — ไม่ใช่ cron expression จริง ๆ (cron expression ต้องใช้ Quartz หรือ Spring @Scheduled)
Executors.newVirtualThreadPerTaskExecutor(); // Java 21+ — virtual threads

4. Future + Callable — รับ "ผลลัพธ์" จาก thread

งานที่รันใน thread บางครั้งเราอยากได้ "ผลลัพธ์" กลับมา — Runnable ทำงานแล้วจบ (ไม่คืนค่า) แต่ Callable<T> คืนค่าได้ และเราดึงผลผ่าน Future<T> (ตัวแทนผลที่จะมาในอนาคต เรียก .get() เพื่อรอรับ):

Runnable = run แล้วจบ (void)
Callable<T> = run แล้ว return ค่า

java

// Callable อนุญาตให้ throw checked exception (exception ที่ Java บังคับให้ประกาศหรือจัดการ เช่น IOException) ได้ — Runnable / Supplier ไม่ได้
Callable<Integer> task = () -> {
    Thread.sleep(1000);
    return 42;
};

Future<Integer> future = pool.submit(task);
Integer result = future.get();     // block จนกว่าจะ return (หรือ exception)

`.get()` มี timeout

java

Integer result = future.get(2, TimeUnit.SECONDS);
// → ถ้านานกว่า 2 วิ throw TimeoutException

5. CompletableFuture — async ที่ chain ได้

java

CompletableFuture<String> fetchUser = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException(e);
    }
    return "Anna";
});

CompletableFuture<String> greet = fetchUser
    .thenApply(name -> "Hello, " + name)               // transform
    .thenApply(String::toUpperCase);

System.out.println(greet.get());                        // HELLO, ANNA

Combine + handle error

java

CompletableFuture<Integer> a = CompletableFuture.supplyAsync(() -> 10);
CompletableFuture<Integer> b = CompletableFuture.supplyAsync(() -> 20);

CompletableFuture<Integer> sum = a.thenCombine(b, Integer::sum);   // 30

CompletableFuture<String> safe = fetchUser
    .thenApply(n -> "Hello, " + n)
    .exceptionally(ex -> "Hello, stranger (err: " + ex.getMessage() + ")");

CompletableFuture = ตัวแทนผลลัพธ์ที่ยังไม่เสร็จ + สามารถผูกการทำงานต่อ ๆ กันได้ (คล้าย Promise ใน JavaScript — การสัญญาว่าจะมีผลลัพธ์ในอนาคต ถ้าเคยใช้ JS แล้วจะคุ้น)

Operator ที่ใช้บ่อย — cheat sheet

Operator	ทำอะไร	คล้าย JS
`thenApply(fn)`	transform ผลลัพธ์ (sync function)	`.then(v => ...)`
`thenCompose(fn)`	flatMap — fn return `CompletableFuture` อีกตัว	`.then(v => promise(v))`
`thenAccept(fn)`	consume ผล (return void)	`.then(v => { ... })`
`thenRun(r)`	ไม่สนค่า แค่ทำต่อ	`.then(() => ...)`
`thenCombine(other, fn)`	รวม 2 future → fn(a, b)	`Promise.all([a,b]).then(([a,b]) => ...)`
`applyToEither(other, fn)`	เอาตัวที่เสร็จก่อน	`Promise.race([a,b]).then(...)`
`exceptionally(fn)`	จับ error → return ค่าทดแทน	`.catch()`
`handle((v, ex) -> ...)`	จับทั้ง success + error	`.then(..., catch)`
`whenComplete((v, ex) -> ...)`	side-effect ดู ไม่เปลี่ยนค่า	`.finally()` แบบเห็นค่า
`orTimeout(t, unit)`	timeout → TimeoutException (Java 9+)	—
`completeOnTimeout(v, t, unit)`	timeout → ใช้ค่า v แทน	—

💡 คอลัมน์ "คล้าย JS" สำหรับคนที่เคยเขียน JavaScript — ข้ามได้ถ้าไม่เคย

`allOf` / `anyOf` — รวมหลาย future

java

CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> fetchA());
CompletableFuture<String> b = CompletableFuture.supplyAsync(() -> fetchB());
CompletableFuture<String> c = CompletableFuture.supplyAsync(() -> fetchC());

// รอครบทุกตัว
CompletableFuture<Void> all = CompletableFuture.allOf(a, b, c);
all.thenRun(() -> {
    System.out.println(a.join() + b.join() + c.join());   // join = get แต่ไม่ throw checked
});

// ตัวแรกที่เสร็จ (race)
CompletableFuture<Object> any = CompletableFuture.anyOf(a, b, c);
System.out.println(any.get());

// Pattern ใช้บ่อย: รวมเป็น list
List<CompletableFuture<String>> futures = List.of(a, b, c);
CompletableFuture<List<String>> joined = CompletableFuture
    .allOf(futures.toArray(new CompletableFuture[0]))
    .thenApply(v -> futures.stream().map(CompletableFuture::join).toList());

`thenCompose` vs `thenApply` — สำคัญ

java

// ❌ ผิด — ได้ CompletableFuture<CompletableFuture<User>>
CompletableFuture<CompletableFuture<User>> bad =
    fetchUserId().thenApply(id -> fetchUserById(id));

// ✅ ถูก — flatMap
CompletableFuture<User> good =
    fetchUserId().thenCompose(id -> fetchUserById(id));

thenCompose = chain future ที่ดึง future อื่น (เหมือน flatMap ของ CompletableFuture — ถ้า map ให้กล่องในกล่อง flatMap จะแกะออกมาให้เหลือกล่องเดียว แทนที่จะได้ CompletableFuture<CompletableFuture<X>> ก็ได้แค่ CompletableFuture<X> เดียว — ต่างจาก Stream.flatMap() ที่ใช้กับ collection)

6. ⚠️ Race Condition — ปัญหาทอง 1

java

// หมายเหตุ: ตัวอย่างใช้ field สาธารณะเพื่อความสั้น — production ใช้ private + getter
class Counter {
    int count = 0;

    void increment() {
        count++;          // ❌ ไม่ atomic
    }
}

Counter c = new Counter();
ExecutorService pool = Executors.newFixedThreadPool(10);

for (int i = 0; i < 1_000_000; i++) {
    pool.submit(c::increment);
}
pool.shutdown();
pool.awaitTermination(1, TimeUnit.MINUTES);

System.out.println(c.count);    // คาด 1,000,000 — แต่ได้ ~973,221 หรือ ~982,114 (สุ่ม)

ทำไมพัง?

count++ ไม่ใช่ 1 คำสั่ง — เป็น 3 step:

int temp = count; ← อ่าน
temp = temp + 1; ← บวก
count = temp; ← เขียน

ถ้า 2 thread ทำพร้อมกัน:

text

Thread A: อ่าน count (= 5)
Thread B: อ่าน count (= 5)
Thread A: +1 → เขียน 6
Thread B: +1 → เขียน 6     ← ควรจะ 7!

→ count หาย 1

7. ทางแก้ 1 — `synchronized`

ทำให้ method/block เป็น atomic — ทีละ thread เท่านั้น:

java

class Counter {
    int count = 0;

    synchronized void increment() {
        count++;
    }
}

หรือ block:

java

synchronized (lock) {
    count++;
}

`synchronized` ทำอะไร?

ก่อนเข้า → acquire monitor lock ของ object
ทำงาน
ออก → release lock
thread อื่นที่รอ → entry ทีละตัว

⚠️ ระวัง deadlock — ถ้าจะ lock หลาย object ต้อง lock ในลำดับเดียวกันเสมอ

8. ทางแก้ 2 — `AtomicInteger` (เร็วกว่า)

java

import java.util.concurrent.atomic.AtomicInteger;

class Counter {
    AtomicInteger count = new AtomicInteger(0);

    void increment() {
        count.incrementAndGet();      // atomic ที่ระดับ CPU (CAS = Compare-and-Swap)
    }
}

AtomicInteger, AtomicLong, AtomicReference<T> — ใช้เมื่อต้องการ atomic แค่ตัวเดียว เร็วกว่า synchronized

Compare-and-Swap (CAS)

java

AtomicInteger counter = new AtomicInteger(10);
boolean updated = counter.compareAndSet(10, 20);   // ถ้าเป็น 10 → set เป็น 20, return true
                                                     // ถ้าไม่ใช่ 10 → ไม่ทำ, return false

8.1 ทางแก้ 3 — `ReentrantLock` (เหมือน synchronized แต่ flexible)

ReentrantLock (reentrant = thread เดิมสามารถ lock ซ้ำได้โดยไม่ deadlock กับตัวเอง) ใน java.util.concurrent.locks = lock ที่มีฟีเจอร์มากกว่า synchronized:

java

import java.util.concurrent.locks.ReentrantLock;

class Counter {
    private final ReentrantLock lock = new ReentrantLock();
    private int count = 0;

    void increment() {
        lock.lock();                          // เหมือน synchronized
        try {
            count++;
        } finally {
            lock.unlock();                    // ⚠️ ต้อง unlock เสมอใน finally
        }
    }

    boolean tryIncrement() {
        if (!lock.tryLock()) return false;    // ไม่รอ — ไม่ได้ก็ false
        try { count++; return true; }
        finally { lock.unlock(); }
    }

    // หมายเหตุ: tryLock(time, unit) เป็น interruptible — throw InterruptedException ถ้า thread ถูก interrupt ระหว่างรอ
    void incrementWithTimeout() throws InterruptedException {
        if (lock.tryLock(2, TimeUnit.SECONDS)) {  // รอไม่เกิน 2 วินาที
            try { count++; }
            finally { lock.unlock(); }
        } else {
            throw new RuntimeException("could not acquire lock");
        }
    }
}

`synchronized` vs `ReentrantLock` — เลือกตัวไหน?

	`synchronized`	`ReentrantLock`
syntax	ง่าย กระชับ	verbose (`lock`/`try`/`finally`/`unlock`)
`tryLock()`	❌	✅ ไม่บล็อก
timeout	❌	✅
interruptible	❌	✅ (`lockInterruptibly()`)
fairness	❌	✅ FIFO ได้ (`new ReentrantLock(true)`)
Condition (wait/notify)	1 ตัว (`wait`/`notify`)	หลายตัว (`newCondition()`)
Virtual thread pin (Java 21-23)	⚠️ pin	✅ ไม่ pin
performance	ปรับ JIT-optimization ดี (ส่วนใหญ่เร็วเท่ากัน)	flexible แต่ overhead นิดหน่อย

💡 Condition = เงื่อนไขที่ thread ใช้รอสัญญาณ (signal) จาก thread อื่น — ช่วยให้ thread หยุดรอโดยไม่เปลืองทรัพยากร

กฎ: ใช้ synchronized เป็น default ถ้าไม่ต้องการ feature พิเศษ — โค้ดง่ายกว่า

Condition — wait/signal แบบหลายเงื่อนไข

java

ReentrantLock lock = new ReentrantLock();
Condition notFull = lock.newCondition();
Condition notEmpty = lock.newCondition();

// Producer
lock.lock();
try {
    while (queue.isFull()) notFull.await();
    queue.add(item);
    notEmpty.signal();
} finally { lock.unlock(); }

สำหรับ producer-consumer ทั่วไป → ใช้ BlockingQueue (ดูหัวข้อ 12. Producer-Consumer ในบทนี้) — เขียนเองยุ่งและพลาดง่าย

8.2 ReadWriteLock + StampedLock — เมื่ออ่านบ่อยกว่าเขียน

ReentrantReadWriteLock ให้ "หลาย read พร้อมกัน, write ตัวเดียว" — ใช้กับ cache, config reload:

java

import java.util.concurrent.locks.ReentrantReadWriteLock;

class CachedConfig {
    private final ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
    private Map<String, String> data = new HashMap<>();

    String get(String key) {
        rw.readLock().lock();                 // หลาย thread ได้ พร้อมกัน
        try { return data.get(key); }
        finally { rw.readLock().unlock(); }
    }

    void reload(Map<String, String> next) {
        rw.writeLock().lock();                // exclusive — รอ reader หมดก่อน
        try { data = next; }
        finally { rw.writeLock().unlock(); }
    }
}

StampedLock (Java 8+) = เร็วกว่าอีก เพราะมี optimistic read (อ่านแบบมองโลกในแง่ดี — ลองอ่านก่อน ค่อยตรวจทีหลังว่ามีคนมาแก้ไหม ถ้ามีค่อย lock จริง) — ไม่ acquire lock จริง แค่จด stamp (ตัวเลขที่ใช้ตรวจสอบว่ามีการเขียนเกิดขึ้นระหว่างอ่านหรือเปล่า):

java

import java.util.concurrent.locks.StampedLock;

private int data;                              // shared field ที่หลาย thread อ่าน/เขียน
StampedLock sl = new StampedLock();
long stamp = sl.tryOptimisticRead();          // ไม่บล็อก
int value = data;                              // อ่านโดยไม่ lock
if (!sl.validate(stamp)) {                    // ถ้าตอนอ่านมีคนเขียน → invalid
    stamp = sl.readLock();                     // fallback → lock จริง
    try { value = data; }
    finally { sl.unlockRead(stamp); }
}

💡 StampedLock เร็วกว่ามากตอน read-heavy + reentry น้อย — แต่ ไม่ reentrant (เรียก lock ซ้ำใน thread เดียวกัน = deadlock เอง)

8.3 Synchronizers — ประสานงานหลาย thread

java.util.concurrent มีของพร้อมใช้ ไม่ต้องเขียน wait/notify เอง:

`Semaphore` — จำกัดจำนวน thread เข้าพร้อมกัน

java

import java.util.concurrent.Semaphore;

// ให้ 3 thread เข้า DB query พร้อมกัน
Semaphore dbSlots = new Semaphore(3);

void query() throws InterruptedException {
    dbSlots.acquire();                        // รอ ถ้า > 3 ตัวกำลังใช้
    try {
        db.execute(...);
    } finally {
        dbSlots.release();                    // คืน slot
    }
}

ใช้ทำ: rate limiting, connection pool, throttling

`CountDownLatch` — รอ "N เหตุการณ์เกิดครบ"

java

import java.util.concurrent.CountDownLatch;

CountDownLatch ready = new CountDownLatch(3);

// 3 worker (สมมติ prepareData() ทำงานบางอย่าง เช่น load config)
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        prepareData();
        ready.countDown();                     // -1
    }).start();
}

ready.await();                                 // รอจนเหลือ 0
System.out.println("ทุกคน ready แล้ว เริ่มจริง");

ใช้ทำ: รอ initialization, ทดสอบ "ปล่อย thread พร้อมกัน"

⚠️ ใช้ครั้งเดียวเลย — count ไม่ reset

`CyclicBarrier` — เหมือน latch แต่ใช้ซ้ำได้ + รอ "พร้อมกัน N ตัว"

java

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.BrokenBarrierException;

CyclicBarrier barrier = new CyclicBarrier(4, () -> {
    System.out.println("ทุกคนถึง barrier — เริ่ม phase ใหม่");
});

for (int i = 0; i < 4; i++) {
    new Thread(() -> {
        for (int phase = 0; phase < 3; phase++) {
            doPhase(phase);
            try {
                barrier.await();              // รอครบ 4 ตัว → run barrier action → reset
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } catch (BrokenBarrierException e) {
                // BrokenBarrierException = เกิดเมื่อ thread อื่นออกจาก barrier กลางคัน
                // (ถูก interrupt หรือ timeout) ทำให้ barrier นั้น 'พัง' ต้องจับแยกจาก InterruptedException
                return;
            }
        }
    }).start();
}

ใช้ทำ: simulation ที่มีหลาย phase, parallel algorithm sync

`Phaser` — ยืดหยุ่นกว่า barrier (จำนวน party เปลี่ยนได้)

java

import java.util.concurrent.Phaser;

Phaser phaser = new Phaser(1);                // start with 1 (main)

for (int i = 0; i < 3; i++) {
    phaser.register();                         // เพิ่ม party
    new Thread(() -> {
        doWork();
        phaser.arriveAndAwaitAdvance();       // เหมือน barrier.await()
        moreWork();
        phaser.arriveAndDeregister();         // ออกจาก party
    }).start();
}
phaser.arriveAndDeregister();

`Exchanger` — แลก data ระหว่าง 2 thread

java

import java.util.concurrent.Exchanger;

Exchanger<List<String>> ex = new Exchanger<>();

// thread A
List<String> mine = new ArrayList<>();
mine.add("a"); mine.add("b");
List<String> got = ex.exchange(mine);          // ส่ง mine, รับของ B

// thread B
List<String> b = List.of("x", "y");
List<String> gotB = ex.exchange(b);

ใช้น้อยใน production — ส่วนใหญ่แทนด้วย BlockingQueue

8.4 ScheduledExecutorService — งานตามเวลา

java

import java.util.concurrent.*;

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

// run ครั้งเดียวหลัง 5 วินาที
scheduler.schedule(() -> System.out.println("delayed"), 5, TimeUnit.SECONDS);

// run ซ้ำทุก ๆ 10 วินาที — เริ่มหลัง 0 วินาที
scheduler.scheduleAtFixedRate(
    () -> doHealthCheck(),
    0, 10, TimeUnit.SECONDS);

// run ซ้ำ — หน่วง 10 วินาทีหลัง task ก่อนหน้าจบ (ไม่ทับซ้อนกัน)
scheduler.scheduleWithFixedDelay(
    () -> doCleanup(),
    0, 10, TimeUnit.SECONDS);

// ⚠️ อย่าเรียก shutdown() ทันที — จะหยุดงาน periodic ทุกตัวทันที
// เรียก scheduler.shutdown() ตอนปิดแอปเท่านั้น (เช่น ใน ShutdownHook)

`scheduleAtFixedRate` vs `scheduleWithFixedDelay` — สำคัญ!

	`fixedRate`	`fixedDelay`
คำนวณรอบใหม่จาก	เริ่ม ของรอบก่อน	จบ ของรอบก่อน
ถ้า task ใช้เวลานานกว่า period	รอบต่อไปเริ่มทันที (catch up)	period นับใหม่หลัง task จบ
ใช้กรณี	งานที่ต้อง "ทุก ๆ X นาทีพอดี" (เช่น metrics)	งานที่ "หน่วง X นาทีระหว่างรอบ" (เช่น polling)

⚠️ ถ้า task throw exception → task ถูกหยุดทันที (ไม่ schedule ต่อ) — wrap ด้วย try/catch เสมอ

8.5 ForkJoinPool + parallelStream

ForkJoinPool (ฟอร์ก-จอยน์-พูล) = pool พิเศษสำหรับ divide-and-conquer (แตกงานเป็นย่อย ๆ แล้ว combine) ใช้เทคนิค work stealing (ขโมยงาน — thread ที่ว่างอยู่จะดึงงานที่ค้างอยู่ใน queue ของ thread อื่นมาช่วยทำ เพื่อให้ CPU ไม่ว่างเปล่า)

java

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

class SumTask extends RecursiveTask<Long> {
    private final long[] arr;
    private final int from, to;
    private static final int THRESHOLD = 10_000;

    SumTask(long[] arr, int from, int to) {
        this.arr = arr; this.from = from; this.to = to;
    }

    @Override
    protected Long compute() {
        if (to - from <= THRESHOLD) {
            long sum = 0;
            for (int i = from; i < to; i++) sum += arr[i];
            return sum;
        }
        int mid = (from + to) / 2;
        SumTask left = new SumTask(arr, from, mid);
        SumTask right = new SumTask(arr, mid, to);
        left.fork();                          // fork() ในที่นี้ = แตก task ย่อยให้ทำงาน async (ต่างจาก OS fork)
        long rightResult = right.compute();   // run sync ตัวนี้
        return left.join() + rightResult;     // join() = รอผลของ task ย่อย (ต่างจาก Thread.join())
    }
}

long[] data = new long[10_000_000];
// ... fill ...
long sum = ForkJoinPool.commonPool().invoke(new SumTask(data, 0, data.length));

`parallelStream` ใช้อะไร?

stream.parallel() / parallelStream() = ใช้ common ForkJoinPool เบื้องหลัง:

java

long sum = LongStream.range(0, 10_000_000).parallel().sum();

⚠️ กฎ: common pool shared ทั้ง JVM — task ที่ block I/O ใน parallelStream จะ exhaust pool (ใช้ thread ใน pool จนหมด ไม่เหลือสำหรับงานอื่น) → สำหรับ I/O ใช้ virtual threads, สำหรับ CPU เท่านั้นใช้ parallelStream

8.6 ThreadLocal — ตัวแปร per-thread

แต่ละ thread มี "ตู้เก็บของ" ของตัวเอง — value ไม่ตัดกัน:

java

public class RequestContext {
    private static final ThreadLocal<String> TRACE_ID = new ThreadLocal<>();

    public static void set(String id) { TRACE_ID.set(id); }
    public static String get() { return TRACE_ID.get(); }
    public static void clear() { TRACE_ID.remove(); }   // ⚠️ สำคัญ
}

// servlet filter = โค้ดที่รันก่อน/หลังทุก HTTP request ใน web app (จะเจอจริงตอนเรียน Spring)
// ตอนนี้ดูแค่ pattern: set → try → finally clear
RequestContext.set(UUID.randomUUID().toString());
try {
    chain.doFilter(req, res);                 // logger เห็น TRACE_ID ลึก ๆ ได้
} finally {
    RequestContext.clear();
}

ใช้: trace id, user context, transaction context (Spring TransactionSynchronizationManager), MDC ของ logger

⚠️ ปัญหาของ ThreadLocal

Memory leak ใน thread pool — thread ถูก reuse → ค่าเก่าอยู่ → กลายเป็นข้ามคำขอ → ต้อง .remove() ทุกครั้งใน finally
ไม่ flow ผ่าน async boundary — ส่งงานไป CompletableFuture → ใหม่ thread ใหม่ → context หาย (ต้อง copy เอง)
กิน memory ใน virtual threads — มี virtual thread เป็นล้าน × ThreadLocal ก็เป็นล้าน

`InheritableThreadLocal` — ส่ง context ไปยัง child thread

java

static final InheritableThreadLocal<String> USER = new InheritableThreadLocal<>();

USER.set("anna");
new Thread(() -> System.out.println(USER.get())).start();   // "anna" — ส่งต่อ

⚠️ InheritableThreadLocal copy เฉพาะตอนสร้าง thread — ถ้าใช้ thread pool (reuse thread) → ค่าไม่ตรง
โซลูชั่นใหม่: ScopedValue (stable ตั้งแต่ Java 25 — Java 21-24 ต้องเปิด --enable-preview) — ดูหัวข้อ 14.6 ด้านล่างในบทนี้ หรือบทที่ 10 (Modern Java) ส่วน ScopedValue

9. ⚠️ Visibility — ปัญหาทอง 2

java

class Worker {
    boolean running = true;

    void run() {
        while (running) {           // อ่าน
            // ทำงาน
        }
    }

    void stop() {
        running = false;            // เขียน — จากอีก thread
    }
}

ปัญหา: thread ที่ run อาจ ไม่เห็น ว่า running = false แล้ว → loop ไม่จบ

ทำไม? — JVM/CPU มี cache ของแต่ละ core → thread A เขียนแล้ว แต่ thread B อ่านจาก cache เก่า

ทางแก้: `volatile`

java

volatile boolean running = true;

volatile = บังคับให้ทุก thread มองเห็นค่าล่าสุดเสมอ — JVM จัดการรายละเอียด CPU cache ให้เอง กลไกที่แท้จริงคือ happens-before (การเขียน volatile ใน thread หนึ่งจะเห็นได้จาก thread อื่นที่อ่าน volatile นั้นภายหลัง) ซึ่งในทางเทคนิค JVM ใส่ memory barrier ให้ (ภาพ "อ่าน/เขียน main memory โดยตรง" เป็นแค่ mental model ที่ช่วยให้จำได้ง่าย)

💡 volatile ไม่ใช่ atomic — ใช้แค่สำหรับ flag (boolean) หรือ reference ที่ swap ทั้งตัว
count++ ใช้ volatile ไม่ได้ — ต้อง synchronized หรือ Atomic

10. Java Memory Model — Happens-Before

อ่านพอเป็นแนวคิด:

ปัญหา: compiler/CPU อาจ reorder คำสั่งเพื่อ optimize → ผลใน thread เดียวยังถูก แต่ใน multi-thread อาจผิด

Memory Model = กฎที่บอกว่าอะไรเกิด "happens-before" อะไร — ถ้า A happens-before B → ทุกอย่างที่ A เขียน B จะเห็น

Action	Happens-Before
ทุกอย่างก่อน `unlock`	ทุกอย่างหลัง `lock` (ของ object เดียวกัน)
เขียน `volatile`	อ่าน `volatile` (ตัวเดียวกัน) ภายหลัง
`thread.start()`	code ใน thread นั้น
code ใน thread	`thread.join()` กลับ

กฎใช้งาน: ใช้ synchronized, volatile, หรือ class ใน java.util.concurrent — ทั้งหมดมี happens-before guarantee ในตัว → ไม่ต้องคิดเอง

10.1 กฎ happens-before ที่ตำราย่อทิ้ง — ต้องรู้สำหรับ debug

นอกจากตารางข้างบน JMM ยังมีกฎที่เจอบ่อยใน production:

กฎ	คืออะไร
Program order	คำสั่งใน thread เดียวกัน เห็นกันตามลำดับที่เขียน
Transitivity	A hb B, B hb C → A hb C (chain ผ่าน volatile/lock ได้)
Final field freeze	field `final` ที่ assign ใน constructor → thread อื่นที่เห็น reference หลัง constructor จบ จะเห็นค่า final ที่ถูกต้องเสมอ (ไม่ต้อง synchronized) — base ของ immutable object thread safety
Classloader init	static field ที่ assign ใน `<clinit>` happens-before ทุก thread ที่ใช้ class นั้นภายหลัง
`Thread.interrupt()`	hb `InterruptedException` / interrupt detect ใน target

10.2 `VarHandle` — modern alternative กับ volatile (Java 9+)

VarHandle (JEP 193) ให้ memory ordering ละเอียดกว่า volatile/synchronized:

java

class Counter {
    private static final VarHandle COUNT;
    static {
        try { COUNT = MethodHandles.lookup()
            .findVarHandle(Counter.class, "count", int.class); }
        catch (Exception e) { throw new ExceptionInInitializerError(e); }
    }
    private int count;

    void increment() {
        COUNT.getAndAdd(this, 1);              // atomic + memory barrier
    }
    int getAcquire() {
        return (int) COUNT.getAcquire(this);   // read with acquire semantics (lighter than volatile)
    }
}

🔬 ส่วนนี้เป็น advanced — สามารถข้ามไปก่อนได้ กลับมาอ่านตอนเขียน low-level concurrent structure

ระดับ memory order (ลด→เพิ่ม overhead):

ระดับ	ความหมาย
`plain`	ไม่มี guarantee — เร็วที่สุด ไม่ปลอดภัย multi-thread
`opaque`	แบบทึบ — รับรองว่าค่าเปลี่ยนได้ แต่ไม่การันตีลำดับ
`acquire/release`	คู่สัญญาณ memory — `acquire` กันไม่ให้ read ที่ตามหลังถูกย้ายขึ้นมาก่อน `release` กันไม่ให้ write ที่อยู่ก่อนหน้าถูกย้ายลงไปทีหลัง เบากว่า `volatile` ใช้กันในโครงสร้างแบบ lock-free handshake (ส่งสัญญาณระหว่าง thread โดยไม่ใช้ lock)
`volatile`	sequential consistency เต็ม — ทุก thread เห็นลำดับ operation เหมือนกัน — หนักที่สุด

เมื่อไหร่ใช้:

เขียน low-level concurrent data structure (lock-free queue, custom atomic)
ต้องการ acquire/release ที่เบากว่า volatile (volatile = sequential consistency, ขัด JIT optimization — JIT คือตัว compile โค้ดขณะรันเพื่อให้เร็วขึ้น แต่ sequential consistency จำกัดการ reorder)
สำหรับ application code ทั่วไป → ใช้ j.u.c หรือ AtomicXXX ก็พอ ไม่ต้อง VarHandle

11. Collection สำหรับ multi-thread

collection ปกติ (ArrayList, HashMap) ไม่ thread-safe — ใช้พร้อมกันหลาย thread แล้วข้อมูลพังหรือ crash Java มี collection เฉพาะสำหรับ concurrent ให้ใช้แทน เช่น ConcurrentHashMap, CopyOnWriteArrayList, BlockingQueue:

java

// ❌ ไม่ thread-safe
List<String> list = new ArrayList<>();
Map<String, Integer> map = new HashMap<>();

// ✅ thread-safe (ใช้ใน multi-thread)
List<String> list = new CopyOnWriteArrayList<>();           // CopyOnWrite = คัดลอก list ทั้งตัวทุกครั้งที่มีการเขียน (อ่านเร็ว เขียนช้า — ใช้เมื่ออ่านบ่อย เขียนน้อย)
Map<String, Integer> map = new ConcurrentHashMap<>();       // ทั่วไป
Queue<Task> queue = new ConcurrentLinkedQueue<>();
BlockingQueue<Task> blocking = new LinkedBlockingQueue<>(); // producer-consumer

ตัวอย่าง ConcurrentHashMap

java

ConcurrentHashMap<String, Integer> counts = new ConcurrentHashMap<>();

words.parallelStream().forEach(word -> {
    counts.merge(word, 1, Integer::sum);    // atomic: ถ้ามี = +1, ถ้าไม่มี = 1
});

12. Producer-Consumer ด้วย BlockingQueue

"Producer-Consumer" เป็นแพตเทิร์น concurrency คลาสสิก — thread หนึ่งผลิตงานใส่คิว อีก thread ดึงไปทำ BlockingQueue จัดการให้อัตโนมัติ: ถ้าคิวเต็ม producer รอ, ถ้าคิวว่าง consumer รอ ไม่ต้องเขียน lock เอง:

java

BlockingQueue<String> queue = new LinkedBlockingQueue<>(100);   // capacity 100

// Producer
new Thread(() -> {
    try {
        for (int i = 0; ; i++) {
            queue.put("event-" + i);     // block ถ้าเต็ม
            Thread.sleep(100);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();   // restore flag
        // exit thread
    }
}).start();

// Consumer
new Thread(() -> {
    try {
        while (true) {
            String event = queue.take(); // block ถ้าว่าง
            System.out.println("got " + event);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

put() / take() block อัตโนมัติ — ไม่ต้อง wait/notify เอง

13. ⚠️ Pitfall ที่เจอบ่อย

1. ลืม `pool.shutdown()`

java

ExecutorService pool = Executors.newFixedThreadPool(4);
pool.submit(...);
// ❌ ลืม shutdown → JVM ไม่ exit, thread ยังอยู่

ใช้ try-with-resources (Java 21+ — ExecutorService implement AutoCloseable):

java

try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
    pool.submit(...);
}   // auto shutdown

2. `Thread.sleep()` ใน lock

java

synchronized (lock) {
    Thread.sleep(10000);   // ❌ thread อื่นรอ 10 วินาที!
}

→ lock ให้สั้นที่สุด

3. Deadlock — ทุกตัว "รอกัน" ตลอดกาล

java

// Thread A
synchronized (lock1) {
    synchronized (lock2) { ... }
}

// Thread B
synchronized (lock2) {
    synchronized (lock1) { ... }   // ❌ A ถือ 1 รอ 2, B ถือ 2 รอ 1 → ค้างตลอด
}

→ acquire lock ในลำดับเดียวกันเสมอ

ตรวจ deadlock: jstack <pid> หรือ jcmd <pid> Thread.print — JVM detect cycle ให้แสดงตรง ๆ ว่า "Found 1 deadlock"
(jstack/jcmd เป็นเครื่องมือที่มากับ JDK, รันบน terminal ขณะที่ Java program กำลังทำงาน; <pid> = process ID ของ JVM — หาได้จาก jps หรือ task manager)

4. Livelock — ขยับตลอด แต่ไปไหนไม่ได้

ตัวอย่าง: 2 thread "ใจดี" ถอยให้กัน

java

while (!done) {
    if (other.isWaiting()) {
        sleep(10);
        continue;                              // ถอยให้
    }
    doWork();
}
// ❌ ถ้า other ก็ "ถอยให้" เหมือนกัน → ทั้งคู่ sleep รอกัน ไม่มีใครได้ทำ

→ ใส่ randomness (backoff) หรือ priority

5. Starvation — มี thread ตัวหนึ่งไม่ได้ run

เกิดจาก:

thread priority ต่ำมาก ๆ ใน load สูง
ReentrantLock (non-fair) อาจ "เลือก" ที่จะให้ thread ที่ active ก่อน
ReadWriteLock: writer รอตลอดเพราะมี reader ใหม่ ๆ มาเรื่อย ๆ

→ ใช้ ReentrantLock(true) (fair mode) หรือ design ใหม่

6. Daemon thread vs User thread

java

Thread t = new Thread(() -> doWork());
t.setDaemon(true);                            // daemon: JVM exit ได้แม้ thread นี้ยัง run อยู่
t.start();

User thread (default) — JVM รอจนจบทุก user thread ก่อน exit
Daemon thread — background (GC, finalizer) — JVM exit ไม่รอ

⚠️ ห้ามใช้ daemon thread ทำงาน "ต้องจบ" เช่น flush log / commit DB — เพราะ JVM อาจฆ่ากลางทาง

7. ใช้ `Date`/`SimpleDateFormat` จากหลาย thread

SimpleDateFormat ไม่ thread-safe — bug หายากมาก

→ ใช้ DateTimeFormatter (จาก java.time) — thread-safe

8. `AtomicLong` ภายใต้ contention สูง → ใช้ `LongAdder` แทน

ภายใต้ high contention (การแย่งกันใช้ทรัพยากรเดียวกัน) AtomicLong.incrementAndGet() อาจ ช้ากว่า synchronized ด้วยซ้ำ เพราะทุก thread พยายาม CAS บน long เดียวกัน → livelock retry loop

java

// ❌ counter ใน hot path ที่หลายร้อย thread เพิ่มพร้อมกัน
AtomicLong requestCount = new AtomicLong();
requestCount.incrementAndGet();    // CAS contention สูง

// ✅ LongAdder / LongAccumulator (Java 8+) แตก counter เป็น cell (ช่องนับแยกต่อ thread — เพื่อลดการแย่งกัน)
LongAdder requestCount = new LongAdder();
requestCount.increment();          // เพิ่ม cell ของตัวเอง — ไม่ contend
long total = requestCount.sum();   // อ่านรวมทุก cell (ช้ากว่า แต่อ่านน้อยกว่าเขียน)

ใช้ LongAdder เมื่อ: write บ่อยมาก, read น้อย, ไม่ต้องการ atomic CAS semantics

9. `CompletableFuture` async ที่ไม่ระบุ executor

supplyAsync(...) / thenApplyAsync(...) ที่ไม่ส่ง executor → ใช้ common ForkJoinPool (default) ซึ่ง:

pool เล็ก (CPU count - 1 = จำนวน core ของ CPU ลบ 1, ขั้นต่ำ 1 thread — เช่น เครื่อง 8 core มีแค่ 7 thread) สามารถ override ได้ด้วย -Djava.util.concurrent.ForkJoinPool.common.parallelism=N
shared กับ parallelStream
blocking I/O ใน task = exhaust pool ทันที — สำหรับ I/O ควรสร้าง custom ForkJoinPool แยกหรือใช้ virtual threads

java

// ❌ ใช้ common pool → blocking JDBC พัง pool
CompletableFuture.supplyAsync(() -> jdbcRepo.findById(id));

// ✅ ระบุ executor ของตัวเอง
ExecutorService io = Executors.newVirtualThreadPerTaskExecutor();
CompletableFuture.supplyAsync(() -> jdbcRepo.findById(id), io);

10. `Thread.interrupt()` — ลืม restore flag

InterruptedException ใน catch block ต้อง restore interrupt flag เสมอ ไม่งั้น caller ที่เช็ค interrupt status จะไม่รู้:

java

// ✅
try {
    Thread.sleep(1000);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();    // ⬅ restore flag
    return;
}

14. Virtual Threads (Java 21+) — เปลี่ยนเกม

ปัญหาของ thread เก่า (เรียก platform thread):

หนัก (~2 MB stack)
limit ~ไม่กี่พัน thread
I/O block → thread เสีย resources

Virtual thread:

เบามาก (~ไม่กี่ KB)
มีได้ ล้าน thread ได้
block ที่ I/O → JVM ยก thread ออก → ตัวอื่นใช้ CPU แทน

java

// แทน newFixedThreadPool
// (HttpClient, HttpResponse, BodyHandlers — ดูบทที่ 13 สำหรับวิธีใช้จริง)
try (ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
        pool.submit(() -> {
            HttpResponse<String> r = client.send(req, BodyHandlers.ofString());
            return r.body();
        });
    }
}

หรือสร้างตรง ๆ:

java

Thread.startVirtualThread(() -> doSomething());

💡 Virtual thread = ดีสำหรับงานที่ block ที่ I/O (database, http call) — ไม่ช่วย CPU-bound

⚠️ Virtual Thread Pinning — ปัญหาที่เคยมี (และแก้แล้ว)

ศัพท์สำคัญ:
carrier thread (แคเรียร์-เธรด) = platform thread จริง ๆ ที่ทำหน้าที่ "แบก" virtual thread ไว้ขณะทำงาน
pin (พิน) = virtual thread ติดอยู่กับ carrier thread จนออกไม่ได้ ทำให้ carrier thread ไม่ว่างรับ virtual thread อื่น ทำลายประสิทธิภาพ

ก่อน Java 24: ถ้า virtual thread กำลังรอ (I/O, lock) อยู่ภายใน synchronized block — มันจะ pin กับ platform thread (carrier thread) — ไม่ปล่อย → ทำลายข้อดีของ virtual thread

java

// Java 21-23 — มีปัญหา
synchronized (lock) {
    httpCall();           // ❌ pin carrier — virtual thread อื่นรอ
}

Java 24 (JEP 491) แก้แล้ว: synchronized ไม่ pin อีกต่อไป → virtual thread ใช้ได้ตรง ๆ ปลอดภัย

สำหรับ Java 21-23: ถ้าทำ I/O ใน critical section ใช้ ReentrantLock แทน synchronized เพื่อเลี่ยง pin

💡 Spring Boot + Virtual Threads (Java 21-23): ถ้าเปิด spring.threads.virtual.enabled=true กับ Spring Boot 3.2+ พร้อม JDBC/HikariCP — JDBC driver บางตัวใช้ synchronized ภายใน ทำให้ virtual thread อาจ pin carrier thread ใต้ high concurrency ทำให้ประสิทธิภาพลดลง JEP 491 (Java 24) แก้ปัญหานี้แล้ว — แนะนำใช้ Java 24+ สำหรับ production workload ที่ใช้ virtual thread + JDBC

`Thread.ofVirtual()` builder

java

// สร้างทีละตัว
Thread.ofVirtual().name("worker-1").start(() -> doWork());
Thread.ofVirtual().unstarted(() -> doWork());

// factory สำหรับ ExecutorService
var factory = Thread.ofVirtual().name("io-", 0).factory();
var pool = Executors.newThreadPerTaskExecutor(factory);

14.5. Structured Concurrency (Java 25, JEP 505 preview)

ปัญหาของ pattern เก่า — เปิดหลาย thread แล้วรอผล:

java

ExecutorService pool = Executors.newFixedThreadPool(2);
Future<String> user = pool.submit(() -> fetchUser(id));
Future<List<Order>> orders = pool.submit(() -> fetchOrders(id));

String u = user.get();
List<Order> o = orders.get();
// ❌ ถ้า fetchUser fail → fetchOrders ยังรันต่อ (เสีย resource)
// ❌ error handling ยุ่งมาก
// ❌ ดู thread tree ไม่ออกว่าใครเป็นลูกใคร

Structured Concurrency บอกว่า "task ที่แตก thread ลูก ต้อง จบพร้อมกัน" — เหมือน try-with-resources แต่สำหรับ thread:

💡 Java 25 (JEP 505) redesign API ใหม่ — ใช้ StructuredTaskScope.open() + Joiner แทน new ShutdownOnFailure() เก่า (API เก่า Java 21-24 ถูกถอดแล้วใน Java 25)
Joiner คืออะไร: object ที่กำหนด "นโยบาย" ว่าจะทำอะไรเมื่อ subtask จบ — เช่น รวมผลทั้งหมด, หยุดทันทีถ้า fail, หรือเอาตัวแรกที่สำเร็จ Joiner เป็น strategy pattern ที่ทำให้ scope รู้ว่าต้องรอแบบไหน

java

import java.util.concurrent.StructuredTaskScope;

// Joiner.allSuccessfulOrThrow() = fail-fast: ถ้าตัวไหน fail → cancel ที่เหลือ + throw
try (var scope = StructuredTaskScope.open(StructuredTaskScope.Joiner.allSuccessfulOrThrow())) {
    var user   = scope.fork(() -> fetchUser(id));
    var orders = scope.fork(() -> fetchOrders(id));

    scope.join();                  // รอทุก subtask (+ throw ถ้า fail ตาม Joiner policy)

    return new UserView(user.get(), orders.get());
}   // scope ปิด → ทุก subtask ถูก cancel แน่นอน

ข้อดี

ทั้งหมดหรือไม่มีเลย (All-or-nothing) — ถ้าตัวหนึ่ง fail → cancel ที่เหลือทันที (Joiner.allSuccessfulOrThrow())
ตัวแรกที่สำเร็จ (First-success) — ใช้ Joiner.anySuccessfulResultOrThrow() ถ้าอยากได้ตัวแรกที่สำเร็จ (เช่น query หลาย mirror)
Lifetime ชัด — scope จบ = thread ลูกจบหมด ไม่หลุด
Thread dump อ่านได้ — เห็น tree ของ task แม่-ลูก
Cancel ผ่าน scope — exception ขึ้น → scope ปิด → ทุกตัวโดน interrupt อัตโนมัติ

Combo กับ Scoped Values

java

final static ScopedValue<String> USER_ID = ScopedValue.newInstance();

ScopedValue.where(USER_ID, "u-123").run(() -> {
    try (var scope = StructuredTaskScope.open(StructuredTaskScope.Joiner.allSuccessfulOrThrow())) {
        scope.fork(() -> fetchProfile());     // เห็น USER_ID
        scope.fork(() -> fetchPrefs());       // เห็น USER_ID
        scope.join();                         // throws ถ้า fail
    }
});

— Scoped Value flow ผ่าน subtask อัตโนมัติ ไม่ต้อง pass parameter

📌 สถานะ: ยังเป็น preview ใน Java 25 (JEP 505) — API ผ่านการ re-incubate หลายรอบ (JEP 428→453→462→480→499→505) และยังอยู่ระหว่าง finalize อย่าใช้ใน production critical-path; API signature อาจเปลี่ยนได้ในรุ่นถัดไป

14.6. Scoped Values — แทน ThreadLocal กับ Virtual Threads

(ดูเต็มในบทที่ 10 section 12) — สรุปสั้น:

java

final static ScopedValue<String> TENANT = ScopedValue.newInstance();

ScopedValue.where(TENANT, "acme").run(() -> {
    // ทุก method ที่เรียกจากนี่ เห็น TENANT.get() = "acme"
    // ไม่มี leak — ออก scope = หาย
});

เมื่อไหร่ใช้อะไร

สถานการณ์	ใช้
ส่งค่า context (userId, tenant, traceId) ลึก ๆ ลงไปใน call chain	Scoped Value
ค่าต้องเปลี่ยนได้ระหว่าง thread	ThreadLocal (เก่า)
Per-virtual-thread cache	อย่าใช้ ThreadLocal (ไม่เหมาะ) — ส่งผ่าน method parameter แทน

15. กฎทอง

สถานการณ์	ใช้
งาน I/O ขนาน (HTTP, DB)	Virtual threads / `CompletableFuture`
งาน CPU-intensive	`parallelStream` / `ForkJoinPool`
Counter / flag	`AtomicInteger` / `AtomicReference` / `volatile`
Shared mutable state ซับซ้อน	`synchronized` (สั้น ๆ) หรือ refactor ให้ไม่ share
Producer-Consumer	`BlockingQueue`
Map ใน multi-thread	`ConcurrentHashMap`

กฎสุดท้าย — ดีที่สุด: ออกแบบให้ ไม่ share mutable state ตั้งแต่แรก

Immutable object (record) → thread-safe ฟรี
Pass-by-value (copy) แทน shared reference
ใช้ message passing (queue) แทน shared variable

16. Checkpoint

💡 Checkpoint ในบทนี้เป็น แบบฝึกหัดขั้นสูง ไม่ใช่เงื่อนไขที่ต้องผ่านก่อนไปต่อ — ถ้ายังไม่พร้อมเขียน concurrency เอง ข้ามไปอ่าน Spring Boot ก่อนได้เลย แล้วค่อยกลับมาลองทำตอนต้องเขียนโค้ดหลาย thread จริง ๆ

🛠️ Checkpoint 12.1 — Parallel sum
ทำ method ที่รับ List<Integer> ใหญ่ ๆ (1 ล้านตัว) แล้วใช้ ExecutorService แบ่งงานเป็น chunk → return ผลรวม
เทียบความเร็วกับ list.stream().mapToInt(...).sum() กับ parallelStream

แนวเฉลย 12.1

แบ่ง list เป็น N chunk ส่งแต่ละ chunk ให้ pool.submit(...) → เก็บ Future<Long> → รวมผลด้วย .get()
โดยทั่วไป parallelStream() จะเร็วพอ ๆ กัน (ใช้ ForkJoinPool เดียวกัน) — ความแตกต่างเห็นชัดตอน list ใหญ่มากและ CPU-bound จริง ๆ

🛠️ Checkpoint 12.2 — Crawl ขนาน
รับ list ของ URL → ใช้ virtual threads ดาวน์โหลดพร้อมกัน → return list ของ response body
(ใช้ HttpClient จาก java.net.http — ดูบทที่ 13 ถ้ายังไม่เคยใช้)

แนวเฉลย 12.2

📌 หมายเหตุ: เฉลยนี้ใช้ HttpClient, HttpRequest, และ HttpResponse.BodyHandlers จาก java.net.http ซึ่งจะอธิบายละเอียดในบทที่ 13 — ถ้ายังไม่ได้อ่านบทนั้น ข้ามดูเฉลยนี้ก่อนได้ หรือแทนด้วย Thread.sleep(100) เพื่อจำลอง I/O delay แทน

java

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

List<String> crawl(List<String> urls) throws Exception {
    HttpClient client = HttpClient.newHttpClient();
    try (var pool = Executors.newVirtualThreadPerTaskExecutor()) {
        List<Future<String>> futures = urls.stream()
            .map(url -> pool.submit(() -> {
                var req = HttpRequest.newBuilder(URI.create(url)).build();
                return client.send(req, HttpResponse.BodyHandlers.ofString()).body();
            })).toList();
        List<String> results = new ArrayList<>();
        for (var f : futures) results.add(f.get());
        return results;
    }
}

🛠️ Checkpoint 12.3 — Race condition demo

สร้าง counter ที่ไม่ใช้ synchronized → run 10 thread ละ 100k increment → ดู count ที่ออก
แก้ด้วย synchronized → ดู count
แก้ด้วย AtomicInteger → เทียบเวลา

แนวเฉลย 12.3

ข้อ 1: ผลลัพธ์จะน้อยกว่า 1,000,000 เสมอ (เช่น ~970,000) — แต่ละรันได้ค่าต่างกัน (non-deterministic)
ข้อ 2: ผลลัพธ์ = 1,000,000 ตรง — แต่ช้ากว่าเพราะ thread ต้องรอ lock
ข้อ 3: ผลลัพธ์ = 1,000,000 ตรง และเร็วกว่า synchronized บน load ปกติ — ทดสอบด้วย System.nanoTime() ก่อน/หลัง pool.awaitTermination()

17. สรุปบท

✅ Thread = "เส้น" ใน process เดียว, share memory
✅ ใช้ ExecutorService (thread pool) — อย่าสร้าง new Thread() เอง
✅ Callable + Future รับ result, CompletableFuture chain งาน async — allOf/anyOf/thenCompose ครบเครื่อง
✅ Race condition → synchronized / ReentrantLock / AtomicInteger / ConcurrentHashMap
✅ Lock พิเศษ: ReadWriteLock (อ่านบ่อย), StampedLock (optimistic), ReentrantLock (timeout/interrupt)
✅ Synchronizers: Semaphore (จำกัด), CountDownLatch (รอ N event), CyclicBarrier/Phaser (sync phase)
✅ ScheduledExecutorService — fixedRate (start-to-start) vs fixedDelay (end-to-start)
✅ ForkJoinPool + parallelStream = CPU-bound divide-and-conquer (common pool — shared ทั้ง JVM!)
✅ ThreadLocal — per-thread state, ⚠️ ต้อง .remove() ใน finally (ป้องกัน leak ใน pool)
✅ Visibility → volatile (สำหรับ flag), happens-before
✅ Pitfall: deadlock (lock order ต่างกัน), livelock (ถอยให้กัน), starvation (rare run)
✅ Daemon vs User thread — JVM รอแค่ user thread
✅ Virtual threads (Java 21+) → million-thread สำหรับงาน I/O; Java 24 แก้ pinning แล้ว
✅ Structured Concurrency (Java 25 preview) — task tree all-or-nothing
✅ กฎสูงสุด: ออกแบบให้ไม่ share mutable state ตั้งแต่แรก

← บทที่ 11 | บทที่ 13 → Networking, Regex, Crypto

บทที่ 12 — Concurrency (Thread, synchronized, volatile, Memory Model) ​

1. Process vs Thread ​

2. สร้าง Thread แรก ​

.join() — รอ thread จบ ​

3. ⚠️ อย่าสร้าง Thread เอง — ใช้ ExecutorService ​

Thread pool ที่ใช้บ่อย ​

4. Future + Callable — รับ "ผลลัพธ์" จาก thread ​

.get() มี timeout ​

5. CompletableFuture — async ที่ chain ได้ ​

Combine + handle error ​

Operator ที่ใช้บ่อย — cheat sheet ​

allOf / anyOf — รวมหลาย future ​

thenCompose vs thenApply — สำคัญ ​

6. ⚠️ Race Condition — ปัญหาทอง 1 ​

7. ทางแก้ 1 — synchronized ​

synchronized ทำอะไร? ​

8. ทางแก้ 2 — AtomicInteger (เร็วกว่า) ​

Compare-and-Swap (CAS) ​

8.1 ทางแก้ 3 — ReentrantLock (เหมือน synchronized แต่ flexible) ​

synchronized vs ReentrantLock — เลือกตัวไหน? ​

Condition — wait/signal แบบหลายเงื่อนไข ​

8.2 ReadWriteLock + StampedLock — เมื่ออ่านบ่อยกว่าเขียน ​

8.3 Synchronizers — ประสานงานหลาย thread ​

Semaphore — จำกัดจำนวน thread เข้าพร้อมกัน ​

CountDownLatch — รอ "N เหตุการณ์เกิดครบ" ​

CyclicBarrier — เหมือน latch แต่ใช้ซ้ำได้ + รอ "พร้อมกัน N ตัว" ​

Phaser — ยืดหยุ่นกว่า barrier (จำนวน party เปลี่ยนได้) ​

Exchanger — แลก data ระหว่าง 2 thread ​

8.4 ScheduledExecutorService — งานตามเวลา ​

scheduleAtFixedRate vs scheduleWithFixedDelay — สำคัญ! ​

8.5 ForkJoinPool + parallelStream ​

parallelStream ใช้อะไร? ​

8.6 ThreadLocal — ตัวแปร per-thread ​

⚠️ ปัญหาของ ThreadLocal ​

InheritableThreadLocal — ส่ง context ไปยัง child thread ​

9. ⚠️ Visibility — ปัญหาทอง 2 ​

ทางแก้: volatile ​

10. Java Memory Model — Happens-Before ​

10.1 กฎ happens-before ที่ตำราย่อทิ้ง — ต้องรู้สำหรับ debug ​

10.2 VarHandle — modern alternative กับ volatile (Java 9+) ​

11. Collection สำหรับ multi-thread ​

ตัวอย่าง ConcurrentHashMap ​

12. Producer-Consumer ด้วย BlockingQueue ​

13. ⚠️ Pitfall ที่เจอบ่อย ​

1. ลืม pool.shutdown() ​

2. Thread.sleep() ใน lock ​

3. Deadlock — ทุกตัว "รอกัน" ตลอดกาล ​

4. Livelock — ขยับตลอด แต่ไปไหนไม่ได้ ​

5. Starvation — มี thread ตัวหนึ่งไม่ได้ run ​

6. Daemon thread vs User thread ​

7. ใช้ Date/SimpleDateFormat จากหลาย thread ​

8. AtomicLong ภายใต้ contention สูง → ใช้ LongAdder แทน ​

9. CompletableFuture async ที่ไม่ระบุ executor ​

10. Thread.interrupt() — ลืม restore flag ​

14. Virtual Threads (Java 21+) — เปลี่ยนเกม ​

⚠️ Virtual Thread Pinning — ปัญหาที่เคยมี (และแก้แล้ว) ​

Thread.ofVirtual() builder ​

14.5. Structured Concurrency (Java 25, JEP 505 preview) ​

ข้อดี ​

Combo กับ Scoped Values ​

14.6. Scoped Values — แทน ThreadLocal กับ Virtual Threads ​

เมื่อไหร่ใช้อะไร ​

15. กฎทอง ​

16. Checkpoint ​

17. สรุปบท ​

บทที่ 12 — Concurrency (Thread, synchronized, volatile, Memory Model)

1. Process vs Thread

2. สร้าง Thread แรก

`.join()` — รอ thread จบ

3. ⚠️ อย่าสร้าง Thread เอง — ใช้ ExecutorService

Thread pool ที่ใช้บ่อย

4. Future + Callable — รับ "ผลลัพธ์" จาก thread

`.get()` มี timeout

5. CompletableFuture — async ที่ chain ได้

Combine + handle error

Operator ที่ใช้บ่อย — cheat sheet

`allOf` / `anyOf` — รวมหลาย future

`thenCompose` vs `thenApply` — สำคัญ

6. ⚠️ Race Condition — ปัญหาทอง 1

7. ทางแก้ 1 — `synchronized`

`synchronized` ทำอะไร?

8. ทางแก้ 2 — `AtomicInteger` (เร็วกว่า)

Compare-and-Swap (CAS)

8.1 ทางแก้ 3 — `ReentrantLock` (เหมือน synchronized แต่ flexible)

`synchronized` vs `ReentrantLock` — เลือกตัวไหน?

Condition — wait/signal แบบหลายเงื่อนไข

8.2 ReadWriteLock + StampedLock — เมื่ออ่านบ่อยกว่าเขียน

8.3 Synchronizers — ประสานงานหลาย thread

`Semaphore` — จำกัดจำนวน thread เข้าพร้อมกัน

`CountDownLatch` — รอ "N เหตุการณ์เกิดครบ"

`CyclicBarrier` — เหมือน latch แต่ใช้ซ้ำได้ + รอ "พร้อมกัน N ตัว"

`Phaser` — ยืดหยุ่นกว่า barrier (จำนวน party เปลี่ยนได้)

`Exchanger` — แลก data ระหว่าง 2 thread

8.4 ScheduledExecutorService — งานตามเวลา

`scheduleAtFixedRate` vs `scheduleWithFixedDelay` — สำคัญ!

8.5 ForkJoinPool + parallelStream

`parallelStream` ใช้อะไร?

8.6 ThreadLocal — ตัวแปร per-thread

⚠️ ปัญหาของ ThreadLocal

`InheritableThreadLocal` — ส่ง context ไปยัง child thread

9. ⚠️ Visibility — ปัญหาทอง 2

ทางแก้: `volatile`

10. Java Memory Model — Happens-Before

10.1 กฎ happens-before ที่ตำราย่อทิ้ง — ต้องรู้สำหรับ debug

10.2 `VarHandle` — modern alternative กับ volatile (Java 9+)

11. Collection สำหรับ multi-thread

ตัวอย่าง ConcurrentHashMap

12. Producer-Consumer ด้วย BlockingQueue

13. ⚠️ Pitfall ที่เจอบ่อย

1. ลืม `pool.shutdown()`

2. `Thread.sleep()` ใน lock

3. Deadlock — ทุกตัว "รอกัน" ตลอดกาล

4. Livelock — ขยับตลอด แต่ไปไหนไม่ได้

5. Starvation — มี thread ตัวหนึ่งไม่ได้ run

6. Daemon thread vs User thread

7. ใช้ `Date`/`SimpleDateFormat` จากหลาย thread

8. `AtomicLong` ภายใต้ contention สูง → ใช้ `LongAdder` แทน

9. `CompletableFuture` async ที่ไม่ระบุ executor

10. `Thread.interrupt()` — ลืม restore flag

14. Virtual Threads (Java 21+) — เปลี่ยนเกม

⚠️ Virtual Thread Pinning — ปัญหาที่เคยมี (และแก้แล้ว)

`Thread.ofVirtual()` builder

14.5. Structured Concurrency (Java 25, JEP 505 preview)

ข้อดี

Combo กับ Scoped Values

14.6. Scoped Values — แทน ThreadLocal กับ Virtual Threads

เมื่อไหร่ใช้อะไร

15. กฎทอง

16. Checkpoint

17. สรุปบท