How do I throttle concurrent threads using ThreadPoolExecutor settings?

By Wayan in Concurrency 0 Comment

To throttle concurrent threads using ThreadPoolExecutor settings in Java, you can configure its key parameters: core pool size, maximum pool size, and queue capacity. These settings control how ThreadPoolExecutor manages the number of concurrently running threads and queued tasks.

Explanation of Key ThreadPoolExecutor Settings:

  1. Core Pool Size:
    • This defines the number of threads that are kept in the pool even when they are idle.
    • If the number of actively running threads is less than the core pool size, a new thread is created to handle a task, even if there are idle threads.
  2. Maximum Pool Size:
    • This is the maximum number of threads that can exist in the pool.
    • If the pool reaches this limit, tasks are queued instead of creating new threads.
  3. Queue Capacity:
    • A BlockingQueue is used to hold tasks that are waiting to execute.
    • If the queue is full and the number of active threads is already at the maximum pool size, new tasks will be rejected according to the specified RejectedExecutionHandler.

By adjusting these parameters, you can throttle the number of active threads, controlling concurrency.

Steps to Throttle Threads:

  1. Use a Fixed Maximum Pool Size:
    Set a value for corePoolSize and maximumPoolSize, controlling the maximum number of threads allowed to execute concurrently.

  2. Configure the Queue Size:
    Use a bounded queue (e.g., ArrayBlockingQueue) with a fixed size to limit the number of pending tasks. Once the queue is full, no additional tasks will be accepted unless threads become available.

  3. Avoid Overloading the System:
    Ensure that the total number of threads and tasks in the queue doesn’t overwhelm system resources like CPU or memory.

Example Solution:

package org.kodejava.util.concurrent;

import java.util.concurrent.*;

public class ThreadPoolThrottle {
   public static void main(String[] args) {
      // Define Executor settings
      int corePoolSize = 5;  // Minimum threads
      int maxPoolSize = 10;  // Maximum threads
      int queueCapacity = 20; // Queue size
      long keepAliveTime = 1; // Threads idle time in seconds

      // Create a ThreadPoolExecutor
      ThreadPoolExecutor executor = new ThreadPoolExecutor(
              corePoolSize,
              maxPoolSize,
              keepAliveTime,
              TimeUnit.SECONDS,
              new ArrayBlockingQueue<>(queueCapacity),
              new ThreadPoolExecutor.CallerRunsPolicy() // Rejected tasks run in the caller thread
      );

      // Submit tasks to throttle
      for (int i = 0; i < 50; i++) {
         final int taskID = i;
         executor.execute(() -> {
            try {
               System.out.println("Task " + taskID + " is running");
               Thread.sleep(1000); // Simulate work
            } catch (InterruptedException e) {
               Thread.currentThread().interrupt();
            }
         });
      }

      // Shut down the executor
      executor.shutdown();
   }
}

Key Points in the Example:

  1. The corePoolSize is 5, meaning at least 5 threads are always active.
  2. The maximum number of threads is limited to maxPoolSize, which is 10 threads.
  3. ArrayBlockingQueue with a size of 20 prevents too many pending tasks from being enqueued at once.
  4. RejectedExecutionHandler.CallerRunsPolicy ensures that tasks are executed in the caller thread when the queue is full, preventing silent task rejection.

Resulting Throttling Behavior:

  1. No more than 10 threads will run concurrently.
  2. A maximum of 20 tasks will be queued at any time.
  3. Tasks beyond the queue/throttle limit are forced to run in the caller thread or handled by a custom rejection policy.

By tweaking these settings, you can fine-tune thread throttling behavior for specific performance and resource requirements.

How do I implement producer-consumer with LinkedBlockingQueue?

By Wayan in Concurrency 0 Comment

The LinkedBlockingQueue in Java is an implementation of the BlockingQueue interface, which is well-suited for implementing the Producer-Consumer problem. It manages a thread-safe queue where producers can add elements and consumers can take elements, with built-in thread synchronization.

Here’s how you can implement a basic producer-consumer solution using LinkedBlockingQueue:

Example: Producer-Consumer with LinkedBlockingQueue

package org.kodejava.util.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class ProducerConsumerExample {

   public static void main(String[] args) {
      // Shared BlockingQueue with a capacity of 10
      BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);

      // Create and start producer and consumer threads
      Thread producerThread = new Thread(new Producer(queue));
      Thread consumerThread = new Thread(new Consumer(queue));

      producerThread.start();
      consumerThread.start();
   }
}

// Producer class
class Producer implements Runnable {
   private final BlockingQueue<Integer> queue;

   public Producer(BlockingQueue<Integer> queue) {
      this.queue = queue;
   }

   @Override
   public void run() {
      try {
         for (int i = 0; i < 20; i++) { // Produce 20 items
            System.out.println("Producing: " + i);
            queue.put(i); // Adds an element to the queue, waits if full
            Thread.sleep(100); // Simulate production time
         }
      } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         System.out.println("Producer was interrupted");
      }
   }
}

// Consumer class
class Consumer implements Runnable {
   private final BlockingQueue<Integer> queue;

   public Consumer(BlockingQueue<Integer> queue) {
      this.queue = queue;
   }

   @Override
   public void run() {
      try {
         while (true) { // Consume indefinitely (or you can add a termination condition)
            Integer value = queue.take(); // Removes and retrieves the head of the queue, waits if empty
            System.out.println("Consuming: " + value);
            Thread.sleep(150); // Simulate consumption time
         }
      } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         System.out.println("Consumer was interrupted");
      }
   }
}

Explanation of Key Concepts

  1. Thread Safety: LinkedBlockingQueue ensures thread safety — no explicit synchronization is needed.
  2. Blocking Methods:
    • put(E e): Inserts an element into the queue, waiting if the queue is full.
    • take(): Retrieves and removes the next element from the queue, waiting if the queue is empty.
  3. Capacity: You can specify the queue’s maximum capacity to prevent overloading (in this example, it’s set to 10).
  4. Multi-threading:
    • Producer: Continuously adds elements to the queue until it reaches the specified capacity.
    • Consumer: Continuously retrieves and processes elements from the queue until it’s empty (or indefinitely, as shown).

Output

The output will interleave “Producing” and “Consuming” messages since the producer and consumer are running in separate threads:

Producing: 0
Consuming: 0
Producing: 1
Producing: 2
Consuming: 1
Producing: 3
...

Adding Multiple Producers and Consumers

You can easily extend this example to have multiple producers and consumers. For example:

Thread producer1 = new Thread(new Producer(queue));
Thread producer2 = new Thread(new Producer(queue));
Thread consumer1 = new Thread(new Consumer(queue));
Thread consumer2 = new Thread(new Consumer(queue));

producer1.start();
producer2.start();
consumer1.start();
consumer2.start();

With multiple producers and consumers, LinkedBlockingQueue automatically synchronizes all access.

This approach demonstrates how the LinkedBlockingQueue efficiently handles the producer-consumer problem without requiring explicit synchronization, making it a simple yet powerful tool for concurrent programming in Java.

How do I coordinate tasks with Phaser in Java concurrency?

By Wayan in Concurrency 0 Comment

Coordinating tasks with Phaser in Java concurrency involves leveraging its powerful synchronization mechanism, especially designed for dynamic scenarios where the number of threads (or parties) may change during execution. The Phaser class found in the java.util.concurrent package provides a flexible and reusable barrier, similar to CyclicBarrier or CountDownLatch, but with added versatility.

Key Features of Phaser:

  1. Registration and Deregistration: Unlike CyclicBarrier, you can dynamically register and deregister threads during runtime.
  2. Phases: A Phaser has multiple phases (steps) instead of being a one-time or single-step barrier.
  3. Thread Coordination: Tasks can wait for other threads to arrive at a particular phase using arrive and awaitAdvance.

Basic Terminology:

  • Parties: Threads/tasks participating in synchronization.
  • Phase: A synchronization cycle where all registered parties arrive and the Phaser advances to the next phase.

Main Methods of Phaser:

  1. register(): Adds a new party to the Phaser.
  2. bulkRegister(int parties): Registers multiple parties at once.
  3. arrive(): Marks a party’s arrival at a phase but does not block.
  4. arriveAndDeregister(): Marks arrival and reduces the count of parties.
  5. awaitAdvance(int phase): Waits for all parties to arrive at the given phase.
  6. arriveAndAwaitAdvance(): Marks arrival and blocks until all parties arrive, advancing the phaser.

Example: Using Phaser to Coordinate Tasks

package org.kodejava.util.concurrent;

import java.util.concurrent.Phaser;

public class PhaserExample {

   public static void main(String[] args) {
      // Create a Phaser with an initial count of 3 parties (threads)
      Phaser phaser = new Phaser(3);

      // Create and start tasks
      for (int i = 0; i < 3; i++) {
         final int threadId = i;
         new Thread(() -> {
            System.out.println("Thread " + threadId + " is starting phase 1");
            phaser.arriveAndAwaitAdvance(); // Wait for all parties to arrive at phase 1

            // Phase 2 work
            System.out.println("Thread " + threadId + " is starting phase 2");
            phaser.arriveAndAwaitAdvance(); // Wait for all parties to arrive at phase 2

            System.out.println("Thread " + threadId + " has finished.");
         }).start();
      }

      // Additional coordination or deregistration if needed
   }
}

Explanation:

  1. The Phaser is initialized with 3 parties.
  2. Each thread:
    • Does work for phase 1, arrives, and waits (arriveAndAwaitAdvance()).
    • Does work for phase 2, arrives, and waits again.
  3. After all threads arrive at the current phase, the Phaser advances to the next phase, and threads proceed.

Dynamic Registration and Deregistration

If the number of threads or tasks is not fixed, you can dynamically adjust using register() and arriveAndDeregister():

package org.kodejava.util.concurrent;

import java.util.concurrent.Phaser;

public class DynamicPhaserExample {

   public static void main(String[] args) {
      Phaser phaser = new Phaser(1); // Start with 1 to initiate the main thread

      for (int i = 0; i < 3; i++) {
         phaser.register(); // Dynamically register a new party
         final int threadId = i;
         new Thread(() -> {
            System.out.println("Thread " + threadId + " is starting work");
            phaser.arriveAndAwaitAdvance(); // Phase 1

            System.out.println("Thread " + threadId + " has finished.");
            phaser.arriveAndDeregister(); // Deregister after completion
         }).start();
      }

      // Main thread waits for all threads to finish their work
      phaser.arriveAndAwaitAdvance();
      System.out.println("All threads are done. Main thread exiting.");
   }
}
  • The Phaser starts with an initial party (main thread) to coordinate the process.
  • Threads register dynamically.
  • Once a thread finishes its work, it deregisters itself (arriveAndDeregister()).
  • The main thread waits for all worker threads to complete.

Phaser vs Other Synchronization Classes

Feature Phaser CountDownLatch CyclicBarrier
Number of Phases Multiple phases Single “latch” event Single phase, reusable
Dynamic Parties Yes No No
Reusability Yes No Yes

Best Practices

  1. Use Phaser when the number of threads/tasks may change dynamically or when multiple phases of synchronization are required.
  2. Avoid using Phaser if the number of threads/tasks is fixed and single-phase synchronization is sufficient (prefer CountDownLatch or CyclicBarrier for simplicity).
  3. Always deregister parties (arriveAndDeregister()) that no longer participate in the synchronization to avoid hanging or resource leaks.

By combining these methods with configurable task logic, you can effectively use Phaser to coordinate complex concurrent workflows in Java.

How do I synchronize phases of execution with CyclicBarrier?

By Wayan in Concurrency 0 Comment

The CyclicBarrier class in Java allows you to synchronize phases or threads at a common point. It is particularly useful when you have multiple threads working on subtasks that need to wait for each other to proceed to the next phase of their work.

Key Features of CyclicBarrier

  • Reusable: The barrier can be reused once all threads have reached the barrier.
  • Action on Barrier Completion: You can specify a barrier action (a task to run only once by one of the threads) that gets executed when all threads reach the barrier.

How CyclicBarrier Works

  • A CyclicBarrier is initialized with a specific number of parties (threads) that must reach the barrier before they are permitted to proceed.
  • When a thread reaches the barrier, it calls the await() method.
  • The thread is blocked until all the required threads reach the barrier (i.e., call await()).
  • Once all threads reach the barrier:
    • Optionally, the barrier action (if defined) is executed by one thread.
    • All threads are released to continue execution.

Example: Synchronizing Multiple Threads with CyclicBarrier

Here’s an example of synchronizing threads using CyclicBarrier. In this case, multiple worker threads perform some task in phases, and all must wait for one another at the end of each phase before proceeding.

package org.kodejava.util.concurrent;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {

   public static void main(String[] args) {
      // Number of threads (parties) to synchronize
      int numThreads = 3;

      // Create a CyclicBarrier with a barrier action
      CyclicBarrier barrier = new CyclicBarrier(numThreads, () -> {
         System.out.println("All threads have reached the barrier. Proceeding to the next phase...");
      });

      // Create and start worker threads
      for (int i = 0; i < numThreads; i++) {
         new Thread(new Worker(barrier), "Thread " + (i + 1)).start();
      }
   }

   static class Worker implements Runnable {
      private final CyclicBarrier barrier;

      public Worker(CyclicBarrier barrier) {
         this.barrier = barrier;
      }

      @Override
      public void run() {
         try {
            System.out.println(Thread.currentThread().getName() + " is performing the first phase of task...");
            Thread.sleep((long) (Math.random() * 3000)); // Simulate work
            System.out.println(Thread.currentThread().getName() + " has finished the first phase. Waiting at the barrier...");
            barrier.await(); // Wait for other threads to reach the barrier

            System.out.println(Thread.currentThread().getName() + " is performing the second phase of task...");
            Thread.sleep((long) (Math.random() * 3000)); // Simulate work
            System.out.println(Thread.currentThread().getName() + " has finished the second phase. Waiting at the barrier...");
            barrier.await(); // Wait for other threads at the next barrier

            System.out.println(Thread.currentThread().getName() + " has completed all phases.");
         } catch (Exception e) {
            e.printStackTrace();
         }
      }
   }
}

Explanation of Code:

  1. Barrier Creation:
    • new CyclicBarrier(numThreads, action):
      • numThreads: Number of threads involved in synchronization.
      • action: A Runnable task that executes after all threads reach the barrier.
  2. Phase Execution:
    • Each thread performs its task and then calls barrier.await() to wait for others.
    • When all threads have called await(), the barrier opens, the optional action (if defined) executes, and threads proceed.
  3. Random Delays:
    • Simulated with Thread.sleep((long) (Math.random() * 3000)) to illustrate different thread run times.
  4. Multiple Phases:
    • The example includes two phases of execution, and the barrier synchronizes threads at the end of each phase.

Output (Example Output):

Thread 2 is performing the first phase of task...
Thread 1 is performing the first phase of task...
Thread 3 is performing the first phase of task...
Thread 2 has finished the first phase. Waiting at the barrier...
Thread 1 has finished the first phase. Waiting at the barrier...
Thread 3 has finished the first phase. Waiting at the barrier...
All threads have reached the barrier. Proceeding to the next phase...
Thread 2 is performing the second phase of task...
Thread 3 is performing the second phase of task...
Thread 1 is performing the second phase of task...
Thread 2 has finished the second phase. Waiting at the barrier...
Thread 3 has finished the second phase. Waiting at the barrier...
Thread 1 has finished the second phase. Waiting at the barrier...
All threads have reached the barrier. Proceeding to the next phase...
Thread 2 has completed all phases.
Thread 1 has completed all phases.
Thread 3 has completed all phases.

Keynotes:

  1. Thread Releasing:
    • All threads are released simultaneously when all of them reach the barrier.
  2. BarrierAction Execution:
    • The Runnable passed to the CyclicBarrier constructor (optional) is run by one of the threads before proceeding.
  3. Reuse:
    • The CyclicBarrier resets automatically after releasing the threads, so it can be reused for the next phase.
  4. Exceptions:
    • If one thread fails (e.g., throws an exception during await()), the barrier is broken, and other threads waiting at that barrier will also throw a BrokenBarrierException.

This implementation is widely used in parallel processing scenarios where tasks are executed in phases and synchronized at specific points.

How do I use CountDownLatch to wait for other threads to finish?

By Wayan in Concurrency 0 Comment

In Java, the CountDownLatch is a synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads is complete. It is part of the java.util.concurrent package.

How It Works

  • A CountDownLatch is initialized with a given count.
  • Each time one of the threads completes its task, it calls countDown(), which decreases the count by 1.
  • The threads waiting on this latch call await(). These threads remain blocked until the count reaches zero.
  • Once the count reaches zero, all waiting threads are unblocked, and they can proceed.

Example: Using CountDownLatch

Below is an example to demonstrate how to use CountDownLatch to make one thread wait for three other threads to finish:

Code Example

package org.kodejava.util.concurrent;

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {
    public static void main(String[] args) {
        // Initialize CountDownLatch with a count of 3
        CountDownLatch latch = new CountDownLatch(3);

        // Create three worker threads
        for (int i = 1; i <= 3; i++) {
            new Thread(new Worker(i, latch)).start();
        }

        System.out.println("Main thread is waiting for workers to finish...");

        try {
            // The main thread waits for the latch count to reach zero
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("All workers have finished. Main thread resumes.");
    }
}

class Worker implements Runnable {
    private int id;
    private CountDownLatch latch;

    public Worker(int id, CountDownLatch latch) {
        this.id = id;
        this.latch = latch;
    }

    @Override
    public void run() {
        System.out.println("Worker " + id + " started.");
        try {
            // Simulating work with sleep
            Thread.sleep((long) (Math.random() * 3000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Worker " + id + " finished.");

        // Decrement the latch count when work is done
        latch.countDown();
    }
}

Output

The output of the program will be as follows (the order might vary due to thread scheduling):

Main thread is waiting for workers to finish...
Worker 1 started.
Worker 2 started.
Worker 3 started.
Worker 1 finished.
Worker 3 finished.
Worker 2 finished.
All workers have finished. Main thread resumes.

Explanation

  1. CountDownLatch latch = new CountDownLatch(3);
    • Initializes a latch with a count of 3, meaning 3 decrements are required for the latch to reach zero.
  2. latch.countDown();
    • This is called by each worker thread after completing its task to decrement the latch count by 1.
  3. latch.await();
    • The main thread calls this method and waits until the count of the latch becomes zero. Once it’s zero, the main thread resumes execution.
  4. Threads finish their tasks in parallel (order is not guaranteed, as shown in the output), and the latch ensures the main thread waits until all workers are done.

Keynotes

  • CountDownLatch cannot be reused once the count reaches zero. For reusable functionality, consider using CyclicBarrier or Phaser.
  • It’s thread-safe and can be used across multiple threads.
  • Always handle InterruptedException properly when using await().

This synchronization tool is highly useful in scenarios where you need multiple threads to finish their tasks before proceeding to the next step in your program!