multithreading vs parallel stream

// 1. Thread thuần
Thread thread = new Thread(() -> processOrder(orderId));
thread.start();
 
// 2. ExecutorService — thread pool có quản lý
ExecutorService executor = Executors.newFixedThreadPool(10);
Future<OrderResult> future = executor.submit(() -> processOrder(orderId));
OrderResult result = future.get(); // blocking
 
// 3. CompletableFuture — non-blocking composition
CompletableFuture<OrderResult> cf = CompletableFuture
    .supplyAsync(() -> fetchOrder(orderId), executor)
    .thenApplyAsync(order -> enrichOrder(order), executor)
    .thenApplyAsync(order -> calculateTotal(order), executor);

@Service
@RequiredArgsConstructor
public class OrderEnrichmentService {
 
    // Ví dụ cho gọn. Thực tế nên inject một executor bean có bounded queue và được shutdown khi app tắt
    private final ExecutorService executor = Executors.newFixedThreadPool(5);
    private final CustomerClient customerClient;
    private final InventoryClient inventoryClient;
    private final PricingClient pricingClient;
 
    public EnrichedOrder enrich(UUID orderId) {
        CompletableFuture<Customer> customerFuture = CompletableFuture
            .supplyAsync(() -> customerClient.getCustomer(orderId), executor);
 
        CompletableFuture<Inventory> inventoryFuture = CompletableFuture
            .supplyAsync(() -> inventoryClient.checkStock(orderId), executor);
 
        CompletableFuture<Pricing> pricingFuture = CompletableFuture
            .supplyAsync(() -> pricingClient.getPrice(orderId), executor);
 
        // Chờ tất cả hoàn thành, có timeout (Java 9+) để một service treo không kéo treo cả request
        CompletableFuture.allOf(customerFuture, inventoryFuture, pricingFuture)
            .orTimeout(2, TimeUnit.SECONDS)
            .join();
 
        return EnrichedOrder.builder()
            .customer(customerFuture.join())
            .inventory(inventoryFuture.join())
            .pricing(pricingFuture.join())
            .build();
    }
 
    @PreDestroy
    void shutdown() {
        executor.shutdown();
    }
}

// Chuyển từ sequential sang parallel
List<OrderDTO> results = orders.parallelStream()
    .filter(order -> order.getStatus() == OrderStatus.PENDING)
    .map(this::enrichOrder)
    .collect(Collectors.toList());
 
// Hoặc từ array
int sum = Arrays.stream(numbers).parallel().sum();

ForkJoinPool customPool = new ForkJoinPool(4);
List<OrderDTO> results = customPool.submit(() ->
    orders.parallelStream()
        .map(this::cpuIntensiveTransform)
        .collect(Collectors.toList())
).get();
customPool.shutdown();

// Ví dụ: Gửi notification cho 1000 users, giới hạn số lời gọi đồng thời + backpressure
ExecutorService executor = new ThreadPoolExecutor(
    5, 10, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(100),
    new ThreadPoolExecutor.CallerRunsPolicy() // backpressure
);
 
for (User user : users) {
    executor.submit(() -> notificationService.send(user));
}

// Ví dụ: Tính toán thống kê trên 100K records
BigDecimal totalRevenue = transactions.parallelStream()
    .filter(tx -> tx.getYear() == 2024)
    .map(Transaction::getAmount)
    .reduce(BigDecimal.ZERO, BigDecimal::add);

// ❌ Collection nhỏ — overhead > benefit
List<String> names = List.of("A", "B", "C");
names.parallelStream().map(String::toLowerCase).toList(); // slower than sequential
 
// ❌ I/O operations — block common pool threads
users.parallelStream()
    .map(user -> httpClient.fetchProfile(user.getId())) // ĐỪNG
    .toList();
 
// ❌ Shared mutable state — race condition
List<String> results = new ArrayList<>(); // NOT thread-safe
items.parallelStream().forEach(item -> results.add(transform(item))); // BUG
 
// ❌ Order-dependent operations
list.parallelStream().forEachOrdered(this::process); // phần cuối bị ép chạy tuần tự, mất phần lớn lợi ích
 
// ❌ LinkedList source — poor split performance
LinkedList<Item> items = ...;
items.parallelStream(); // splits poorly → unbalanced work

// Setup: 1 triệu integers, CPU-bound operation (tính sqrt rồi filter)
List<Integer> numbers = IntStream.rangeClosed(1, 1_000_000).boxed().toList();
 
// Con số minh hoạ, đo thô không qua JMH, chênh nhiều theo máy và JIT warm-up.
// Thử nhanh trên máy 10 core sau warm-up: sequential ~12ms, parallel ~3ms;
// lần chạy đầu tiên parallel còn chậm hơn sequential.
 
// Sequential
numbers.stream()
    .map(Math::sqrt)
    .filter(n -> n > 500)
    .toList();
 
// Parallel
numbers.parallelStream()
    .map(Math::sqrt)
    .filter(n -> n > 500)
    .toList();
 
// Nhưng với I/O (HTTP call):
// Sequential: 1000 calls × 100ms = 100s
// Parallel Stream (common pool 7 thread + thread gọi, máy 8 core): ~12-14s nhưng BLOCK common pool
// ExecutorService(20 threads): ~5s và isolated

// 1. Blocking common ForkJoinPool ảnh hưởng toàn JVM
// Nếu parallel stream chứa I/O, TẤT CẢ parallel streams khác trong JVM bị chậm
 
// 2. Autoboxing overhead
IntStream.range(0, 1_000_000).parallel().sum(); // tốt — primitive
List<Integer> list = ...;
list.parallelStream().mapToInt(i -> i).sum(); // kém hơn — unboxing
 
// 3. Encounter order preservation
list.parallelStream()
    .filter(...)
    .findFirst(); // bắt buộc giữ order → giảm parallelism
// Thay bằng findAny() nếu không cần thứ tự
 
// 4. Collector không phù hợp
// groupingBy() với downstream không concurrent → bottleneck
list.parallelStream()
    .collect(Collectors.groupingByConcurrent(Item::getCategory)); // dùng concurrent version

// 1. Thread leak — quên shutdown executor
ExecutorService executor = Executors.newFixedThreadPool(10);
// ... dùng xong không shutdown → thread sống mãi
 
// 2. Deadlock
synchronized(lockA) {
    synchronized(lockB) { ... } // Thread 1
}
synchronized(lockB) {
    synchronized(lockA) { ... } // Thread 2 → DEADLOCK
}
 
// 3. Unbounded queue → OOM
Executors.newFixedThreadPool(10); // dùng LinkedBlockingQueue(Integer.MAX_VALUE)
// Nếu task produce nhanh hơn consume → queue vô hạn → OutOfMemoryError
 
// Fix: dùng bounded queue + rejection policy
new ThreadPoolExecutor(5, 10, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(1000),
    new ThreadPoolExecutor.CallerRunsPolicy());

// Tạo 100K virtual threads — mỗi cái chỉ tốn vài trăm byte tới vài KB heap
List<Future<Response>> futures;
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    futures = users.stream()
        .map(user -> executor.submit(() -> httpClient.fetch(user.getId())))
        .toList();
} // close() chờ mọi task xong
 
List<Response> results = futures.stream()
    .map(Future::resultNow) // Java 19+, ném IllegalStateException nếu task bị lỗi
    .toList();

Can xu ly song song?
|
+-- Task khac nhau, can orchestrate? -> Multithreading (CompletableFuture / ExecutorService)
|   +-- I/O-bound (HTTP, DB, File)? -> Virtual Threads (Java 21+) hoac fixed thread pool
|   +-- Can retry/timeout/fallback? -> CompletableFuture + custom executor
|
+-- Cung operation tren large dataset? -> Parallel Stream
    +-- N x Q > 10K (du luong viec)?  -> Parallel Stream
    +-- CPU-bound (no I/O)?           -> Parallel Stream
    +-- Stateless, no side effects?   -> Parallel Stream
    +-- Bat ky dieu kien nao = NO?    -> Sequential Stream hoac Multithreading