Skip to main content
HashnodeUntitled Publication

Command Palette

Search for a command to run...

Bài 8: Concurrent Collections

Published
•7 min read•View as Markdown
D
Donnie

Định nghĩa và Mục đích của Concurrent Collections

Concurrent Collections là một tập hợp các lớp trong Java Collections Framework được thiết kế để hỗ trợ các thao tác đồng thời (concurrent operations) trong môi trường đa luồng (multi-threaded). Các lớp này cung cấp cơ chế đồng bộ hóa nội bộ để đảm bảo an toàn khi truy cập từ nhiều luồng mà không cần đồng bộ hóa thủ công.

Mục đích:

  • Cung cấp các cấu trúc dữ liệu an toàn trong môi trường đa luồng.

  • Tối ưu hóa hiệu suất truy cập đồng thời bằng cách giảm thiểu việc khóa (locking) và tranh chấp (contention).

  • Đơn giản hóa việc lập trình đồng thời bằng cách cung cấp các API dễ sử dụng.

Các Thành phần Chính trong Concurrent Collections

  1. ConcurrentHashMap:

    • Mô tả: Một triển khai của Map cho phép truy cập đồng thời.

    • Đặc điểm: Chia khóa thành các phân đoạn (segments) để giảm thiểu tranh chấp, hỗ trợ thao tác đồng thời đọc và ghi.

    • Tình huống sử dụng: Thích hợp cho các ứng dụng cần một bản đồ an toàn trong môi trường đa luồng với hiệu suất cao.

  2. CopyOnWriteArrayList:

    • Mô tả: Một triển khai của List cho phép truy cập đồng thời bằng cách tạo bản sao của mảng mỗi khi có thao tác ghi.

    • Đặc điểm: An toàn cho các thao tác đọc đồng thời, chi phí cao cho các thao tác ghi.

    • Tình huống sử dụng: Thích hợp cho các ứng dụng đọc nhiều hơn ghi, như danh sách các listeners trong các sự kiện.

  3. CopyOnWriteArraySet:

    • Mô tả: Một triển khai của Set dựa trên CopyOnWriteArrayList, cho phép truy cập đồng thời.

    • Đặc điểm: An toàn cho các thao tác đọc đồng thời, chi phí cao cho các thao tác ghi.

    • Tình huống sử dụng: Thích hợp cho các ứng dụng cần một tập hợp an toàn trong môi trường đa luồng với các thao tác đọc nhiều hơn ghi.

  4. ConcurrentLinkedQueue:

    • Mô tả: Một triển khai của Queue sử dụng danh sách liên kết không đồng bộ.

    • Đặc điểm: Không có giới hạn về kích thước, hỗ trợ các thao tác thêm/xóa đồng thời.

    • Tình huống sử dụng: Thích hợp cho các ứng dụng cần hàng đợi không đồng bộ trong môi trường đa luồng.

  5. LinkedBlockingQueue:

    • Mô tả: Một triển khai của BlockingQueue sử dụng danh sách liên kết.

    • Đặc điểm: Có giới hạn hoặc không giới hạn về kích thước, hỗ trợ các thao tác chặn (blocking) khi hàng đợi đầy hoặc rỗng.

    • Tình huống sử dụng: Thích hợp cho các ứng dụng cần hàng đợi chặn trong môi trường đa luồng.

  6. ConcurrentSkipListMap:

    • Mô tả: Một triển khai của ConcurrentNavigableMap sử dụng thuật toán Skip List.

    • Đặc điểm: Các phần tử được sắp xếp theo thứ tự tự nhiên hoặc theo Comparator, hỗ trợ thao tác đồng thời.

    • Tình huống sử dụng: Thích hợp cho các ứng dụng cần bản đồ điều hướng an toàn trong môi trường đa luồng với các phần tử được sắp xếp.

Ví dụ và Tình huống Sử dụng Concurrent Collections

Ví dụ 1: Sử dụng ConcurrentHashMap để đếm từ trong một văn bản

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class WordCount {
    public static void main(String[] args) {
        ConcurrentHashMap<String, Integer> wordCounts = new ConcurrentHashMap<>();
        ExecutorService executor = Executors.newFixedThreadPool(4);

        String[] texts = {"Java is a programming language", 
                          "Java is widely used", 
                          "Concurrency in Java is powerful"};

        for (String text : texts) {
            executor.submit(() -> {
                String[] words = text.split("\\s+");
                for (String word : words) {
                    wordCounts.merge(word, 1, Integer::sum);
                }
            });
        }

        executor.shutdown();
        while (!executor.isTerminated()) {}

        System.out.println("Word counts: " + wordCounts);
    }
}

Ví dụ 2: Sử dụng CopyOnWriteArrayList để quản lý danh sách các listeners trong một sự kiện

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class EventManager {
    private final List<EventListener> listeners = new CopyOnWriteArrayList<>();

    public void addListener(EventListener listener) {
        listeners.add(listener);
    }

    public void removeListener(EventListener listener) {
        listeners.remove(listener);
    }

    public void notifyListeners(Event event) {
        for (EventListener listener : listeners) {
            listener.onEvent(event);
        }
    }

    public static void main(String[] args) {
        EventManager manager = new EventManager();
        manager.addListener(event -> System.out.println("Listener 1: " + event.getMessage()));
        manager.addListener(event -> System.out.println("Listener 2: " + event.getMessage()));

        Event event = new Event("Event occurred");
        manager.notifyListeners(event);
    }
}

interface EventListener {
    void onEvent(Event event);
}

class Event {
    private final String message;

    public Event(String message) {
        this.message = message;
    }

    public String getMessage() {
        return message;
    }
}

Bài tập Thực hành với Concurrent Collections

Bài tập 1: Sử dụng ConcurrentLinkedQueue để quản lý hàng đợi các yêu cầu trong một hệ thống đa luồng. Thêm 10 yêu cầu vào hàng đợi từ nhiều luồng và xử lý các yêu cầu này từ một luồng khác.

  • Gợi ý:

      import java.util.concurrent.ConcurrentLinkedQueue;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;

  public class RequestQueue {
      public static void main(String[] args) {
          ConcurrentLinkedQueue<String> requests = new ConcurrentLinkedQueue<>();
          ExecutorService producerExecutor = Executors.newFixedThreadPool(2);
          ExecutorService consumerExecutor = Executors.newSingleThreadExecutor();

          // Thêm yêu cầu vào hàng đợi từ nhiều luồng
          for (int i = 0; i < 10; i++) {
              final int requestId = i;
              producerExecutor.submit(() -> {
                  String request = "Request " + requestId;
                  requests.add(request);
                  System.out.println("Thêm: " + request);
              });
          }

          producerExecutor.shutdown();
          while (!producerExecutor.isTerminated()) {}

          // Xử lý yêu cầu từ hàng đợi
          consumerExecutor.submit(() -> {
              while (!requests.isEmpty()) {
                  String request = requests.poll();
                  System.out.println("Xử lý: " + request);
              }
          });

          consumerExecutor.shutdown();
      }
  }

Bài tập 2: Sử dụng LinkedBlockingQueue để quản lý hàng đợi các nhiệm vụ cần xử lý trong một hệ thống đa luồng. Thêm các nhiệm vụ vào hàng đợi từ nhiều luồng và xử lý các nhiệm vụ này từ nhiều luồng khác.

  • Gợi ý:

      import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.LinkedBlockingQueue;

  public class TaskQueue {
      public static void main(String[] args) {
          LinkedBlockingQueue<String> tasks = new LinkedBlockingQueue<>();
          ExecutorService producerExecutor = Executors.newFixedThreadPool(2);
          ExecutorService consumerExecutor = Executors.newFixedThreadPool(2);

          // Thêm nhiệm vụ vào hàng đợi từ nhiều luồng
          for (int i = 0; i < 10; i++) {
              final int taskId = i;
              producerExecutor.submit(() -> {
                  String task = "Task " + taskId;
                  try {
                      tasks.put(task);
                      System.out.println("Thêm: " + task);
                  } catch (InterruptedException e) {
                      Thread.currentThread().interrupt();
                  }
              });
          }

          producerExecutor.shutdown();
          while (!producerExecutor.isTerminated()) {}

          // Xử lý nhiệm vụ từ hàng đợi từ nhiều luồng
          for (int i = 0; i < 2; i++) {
              consumerExecutor.submit(() -> {
                  while (true) {
                      try {
                          String task = tasks.take();
                          System.out.println("Xử lý: " + task);
                      } catch (InterruptedException e) {
                          Thread.currentThread().interrupt();
                          break;
                      }
                  }
              });
          }

          consumerExecutor.shutdown();
      }
  }

Bài tập Nâng cao

  1. Tạo một ứng dụng sử dụng ConcurrentHashMap để lưu trữ và quản lý phiên bản của các tài liệu trong một hệ thống đa luồng. Mỗi luồng sẽ cập nhật phiên bản của một tài liệu khác nhau. Sử dụng phương thức compute để đảm

bảo các phiên bản được cập nhật một cách an toàn.

  • Gợi ý:

      import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;

  public class DocumentVersioning {
      public static void main(String[] args) {
          ConcurrentHashMap<String, Integer> documentVersions = new ConcurrentHashMap<>();
          ExecutorService executor = Executors.newFixedThreadPool(4);

          String[] documents = {"Doc1", "Doc2", "Doc3", "Doc4"};

          for (String doc : documents) {
              executor.submit(() -> {
                  for (int i = 0; i < 10; i++) {
                      documentVersions.compute(doc, (key, value) -> (value == null) ? 1 : value + 1);
                      System.out.println("Cập nhật phiên bản của " + doc + " lên " + documentVersions.get(doc));
                  }
              });
          }

          executor.shutdown();
      }
  }

  1. Tạo một ứng dụng sử dụng ConcurrentSkipListMap để quản lý danh sách các giao dịch tài chính theo thứ tự thời gian trong một hệ thống đa luồng. Mỗi giao dịch sẽ được thêm vào danh sách từ nhiều luồng khác nhau và danh sách sẽ được duyệt để tính tổng số tiền của tất cả các giao dịch.

    • Gợi ý:

        import java.util.concurrent.ConcurrentSkipListMap;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;

  public class FinancialTransactions {
      public static void main(String[] args) {
          ConcurrentSkipListMap<Long, Double> transactions = new ConcurrentSkipListMap<>();
          ExecutorService executor = Executors.newFixedThreadPool(4);

          for (int i = 0; i < 10; i++) {
              final int transactionId = i;
              executor.submit(() -> {
                  long timestamp = System.currentTimeMillis() + transactionId;
                  double amount = 100.0 + transactionId;
                  transactions.put(timestamp, amount);
                  System.out.println("Thêm giao dịch: " + timestamp + " - " + amount);
              });
          }

          executor.shutdown();
          while (!executor.isTerminated()) {}

          double totalAmount = transactions.values().stream().mapToDouble(Double::doubleValue).sum();
          System.out.println("Tổng số tiền của các giao dịch: " + totalAmount);
      }
  }

Tổng kết

Các lớp Concurrent Collections trong Java cung cấp một bộ công cụ mạnh mẽ và hiệu quả để quản lý các cấu trúc dữ liệu trong môi trường đa luồng. Hiểu và sử dụng đúng các lớp này sẽ giúp lập trình viên tối ưu hóa hiệu suất và đảm bảo tính an toàn khi thao tác với dữ liệu đồng thời. Việc lựa chọn đúng loại Concurrent Collection dựa trên yêu cầu cụ thể của ứng dụng là một kỹ năng quan trọng mà mọi lập trình viên Java cần có.

#java

Comments

Join the discussion

No comments yet. Be the first to comment.

More from this blog

Untitled Publication

218 posts