How do I process tasks as they complete using CompletionService?

By I Wayan Saryada in Concurrency 0 Comment

To process tasks as they complete using a CompletionService in Java, you can take advantage of the ExecutorCompletionService. This class provides a mechanism to submit tasks for execution and retrieve their results in the order of completion, rather than the order of submission.

Key Steps for Using CompletionService:

  1. Create an ExecutorService: This handles the thread pooling for concurrent task execution.
  2. Create an ExecutorCompletionService: Wrap the ExecutorService in an ExecutorCompletionService.
  3. Submit Tasks: Use the submit method to submit tasks to the CompletionService.
  4. Process Results as They Complete: Retrieve the results using the poll or take methods of CompletionService.

Example Code:

package org.kodejava.util.concurrent;

import java.util.concurrent.*;

public class CompletionServiceExample {

   public static void main(String[] args) throws InterruptedException {
      int numTasks = 5;

      // Step 1: Create an ExecutorService with fixed thread pool
      ExecutorService executorService = Executors.newFixedThreadPool(3);

      // Step 2: Create an ExecutorCompletionService
      CompletionService<String> completionService = new ExecutorCompletionService<>(executorService);

      // Step 3: Submit tasks to the CompletionService
      for (int i = 0; i < numTasks; i++) {
         int taskId = i;
         completionService.submit(() -> {
            Thread.sleep((long) (Math.random() * 2000)); // Simulate work
            return "Result from Task " + taskId;
         });
      }

      // Step 4: Process tasks as they complete
      for (int i = 0; i < numTasks; i++) {
         try {
            Future<String> resultFuture = completionService.take(); // Retrieves the next completed task
            String result = resultFuture.get(); // Blocks until the result is available
            System.out.println(result);
         } catch (ExecutionException e) {
            System.err.println("Task execution failed: " + e.getMessage());
         }
      }

      // Shutdown the ExecutorService
      executorService.shutdown();
   }
}

Explanation of the Code:

  1. ExecutorService: A thread pool of 3 worker threads is created using Executors.newFixedThreadPool(3).
  2. ExecutorCompletionService: Wraps the ExecutorService to handle submission and retrieval of tasks.
  3. Submitting Tasks: Each task is computed in the background and asynchronously submitted to the completionService.
  4. Result Retrieval:
    • The completionService.take() method blocks until the next completed task result is available.
    • completionService.poll() could also be used if you want non-blocking retrieval (e.g., you check if a result is ready).

  5. Task Results in Completion Order: Results are processed as tasks complete, regardless of their submission order.

When to Use CompletionService

  • When you want to process tasks as they finish, rather than waiting for all tasks to complete.
  • In scenarios where tasks may have uneven execution times, and you want to immediately handle the results of the fastest tasks.

How do I cancel long-running tasks in ExecutorService?

By I Wayan Saryada in Concurrency 0 Comment

To cancel long-running tasks in an ExecutorService, you can use the Future object returned when you submit a task and invoke its cancel method. Below are the steps and some important considerations for canceling tasks:

1. Submit Tasks to the ExecutorService

When you submit a task to an ExecutorService, it returns a Future object that can be used to monitor the task’s progress and cancel it if needed.

ExecutorService executor = Executors.newFixedThreadPool(2);

Future<?> future = executor.submit(() -> {
    // Simulate a long-running task
    try {
        while (!Thread.currentThread().isInterrupted()) {
            System.out.println("Running...");
            Thread.sleep(1000);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt(); // Restore the interrupted status
        System.out.println("Task interrupted.");
    }
});

2. Cancel the Task

To cancel the task, invoke the cancel method on the Future object:

// Cancel the task after 5 seconds
Thread.sleep(5000); // Simulating some delay
boolean wasCancelled = future.cancel(true); // true means interrupt if running

System.out.println("Task cancelled: " + wasCancelled);
  • cancel(true) attempts to stop the execution of the task by interrupting the thread running it. For this to work, the task must regularly check its interrupted status (via Thread.interrupted() or Thread.currentThread().isInterrupted()) and gracefully terminate if interrupted.
  • cancel(false) does not interrupt the running task but prevents it from starting if it hasn’t already begun.

3. Handle Interruption Gracefully

For the cancellation to work, ensure that the task checks the interrupted status and responds accordingly. The task should periodically call Thread.interrupted() or Thread.currentThread().isInterrupted() to detect interruptions.

try {
    while (!Thread.currentThread().isInterrupted()) {
        // Simulate work
        System.out.println("Working...");
        Thread.sleep(1000); // This can throw InterruptedException
    }
} catch (InterruptedException e) {
    Thread.currentThread().interrupt(); // Re-set the interrupted status
    System.out.println("Task interrupted and exiting.");
}

4. Shutdown the ExecutorService

Once you’re done submitting tasks, shut down the ExecutorService to release resources:

executor.shutdown(); // Wait for running tasks to complete
try {
    if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
        executor.shutdownNow(); // Forcefully shut down if tasks don't finish in time
    }
} catch (InterruptedException e) {
    executor.shutdownNow(); // Force an immediate shutdown
    Thread.currentThread().interrupt(); // Reset the interrupted status
}

Keynotes

  • Interruptible Tasks: The tasks you submit must be designed to handle interruptions for cancellation to effectively work. For example, long-running loops or blocking calls should handle the interrupted status.
  • Blocking Methods: If the task is waiting on a blocking call (e.g., Thread.sleep(), Object.wait(), Future.get()), calling cancel(true) will usually interrupt these methods.
  • Non-Interruptible Work: If the task is not interruptible (e.g., performing intensive computations without checking the interrupted flag), cancel(true) will not have an immediate effect.
  • Future API: You can also check the status of a task using methods like isDone(), isCancelled() before or after attempting to cancel it.

This approach ensures your long-running task can be terminated gracefully and resourcefully.

How do I control access to resources using Semaphore?

By I Wayan Saryada in Concurrency 0 Comment

To control access to shared resources in a multithreaded environment, a Semaphore is frequently used, which is part of the java.util.concurrent package. A semaphore manages a set number of permits that control how many threads can access a shared resource simultaneously. Threads acquire permits before accessing the resource and release the permits after they are done, ensuring controlled and synchronized access.

Here’s how you can use a semaphore to control access to resources:

1. Key Points About Semaphore:

  • Permits: The semaphore holds a set number of permits, which represent the number of threads that can access the resource concurrently.
  • Acquire/Release:
    • A thread must acquire a permit using the acquire() method to access the resource.
    • It must release the permit using release() after finishing its access to the resource.
  • Blocking Behavior: If no permits are available, the acquiring thread will block until a permit is released by another thread.

  • Fairness: You can construct a semaphore in a fair mode to ensure that waiting threads acquire permits in the order they requested them.

2. Example: Semaphore with Limited Access to Resources

Here is a simple example where a semaphore is used to control access to a shared resource (e.g., a connection pool or a printer):

package org.kodejava.util.concurrent;

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

public class SemaphoreExample {

    // Semaphore initialized with 2 permits (only 2 threads can access simultaneously).
    private static final Semaphore semaphore = new Semaphore(2);

    public static void main(String[] args) {
        // Create a thread pool with 5 threads
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        // Simulate 5 threads trying to access the shared resource
        for (int i = 1; i <= 5; i++) {
            final int threadId = i;
            executorService.submit(() -> {
                try {
                    // Try to acquire a permit
                    System.out.println("Thread " + threadId + " is trying to acquire a permit.");
                    semaphore.acquire();  // Blocks if no permit is available

                    // Access the shared resource
                    System.out.println("Thread " + threadId + " has acquired a permit.");
                    Thread.sleep(2000);  // Simulate the resource usage

                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    // Release the permit after use
                    System.out.println("Thread " + threadId + " is releasing the permit.");
                    semaphore.release();
                }
            });
        }

        executorService.shutdown();
    }
}

3. How It Works:

  1. Initialization: The semaphore is initialized with a number of permits (new Semaphore(2)), allowing only 2 threads to access the resource concurrently.
  2. Acquire: A thread attempts to access the resource by calling semaphore.acquire(). If permits are unavailable, the thread is blocked until a permit is released by another thread.
  3. Critical Section: Once the permit is acquired, it enters the critical section and uses the resource.
  4. Release: After the thread is done using the resource, it calls semaphore.release() to return a permit, allowing other threads to acquire it.

4. Output Example:

When you run the example above, you might see an output like this, showing how only 2 threads can access the resource simultaneously:

Thread 1 is trying to acquire a permit.
Thread 1 has acquired a permit.
Thread 2 is trying to acquire a permit.
Thread 2 has acquired a permit.
Thread 3 is trying to acquire a permit.
Thread 1 is releasing the permit.
Thread 3 has acquired a permit.
Thread 4 is trying to acquire a permit.
Thread 2 is releasing the permit.
Thread 4 has acquired a permit.
Thread 5 is trying to acquire a permit.
Thread 3 is releasing the permit.
Thread 5 has acquired a permit.
Thread 4 is releasing the permit.
Thread 5 is releasing the permit.

Here, only 2 threads are allowed to acquire permits at a time, while others are blocked until permits are released.

5. Fair Ordering:

If you want the semaphore to provide fairness (FIFO order), you can use the constructor:

Semaphore semaphore = new Semaphore(2, true);

The second argument (true) enables fair ordering, making sure the threads acquire permits in the order they requested them.

6. Use Cases:

  • Database Connection Pools: Managing the number of simultaneous connections to a database.
  • Printers: Limiting how many jobs can access a shared printer.
  • Rate Limiting: Throttling the number of threads processing tasks in high-volume systems.

How do I gracefully shut down an ExecutorService?

By I Wayan Saryada in Concurrency 0 Comment

To gracefully shut down an ExecutorService in Java, you should follow these steps:

  1. Call shutdown():
    • This will prevent the ExecutorService from accepting any new tasks while allowing already submitted tasks to be completed.
  2. Wait for Termination:
    • You can use awaitTermination(long timeout, TimeUnit unit) to wait for a specified amount of time for all tasks to finish their execution.
  3. Force Shutdown if Necessary:
    • If tasks haven’t completed after the wait period, you can call shutdownNow() to attempt to cancel all currently executing tasks and halt further task execution.

Here’s an example:

package org.kodejava.util.concurrent;

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

public class GracefulShutdownExample {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        // Submit some tasks to the executor
        for (int i = 0; i < 10; i++) {
            executorService.submit(() -> {
                try {
                    System.out.println("Task executing: " + Thread.currentThread().getName());
                    Thread.sleep(1000); // Simulate task processing
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    System.out.println("Task interrupted: " + Thread.currentThread().getName());
                }
            });
        }

        // Initiate graceful shutdown
        executorService.shutdown();
        System.out.println("Shutdown initiated");

        try {
            // Wait for all tasks to finish execution or timeout
            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                System.out.println("Timeout reached, forcing shutdown...");
                // Force shutdown if tasks are still running
                executorService.shutdownNow();

                // Wait again to ensure shutdownNow has time to interrupt tasks
                if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    System.err.println("Executor did not terminate");
                }
            }
        } catch (InterruptedException e) {
            // Re-cancel if the current thread was interrupted
            executorService.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }

        System.out.println("Executor service shut down");
    }
}

Explanation of Key Methods:

  • shutdown():
    • Initiates an orderly shutdown where previously submitted tasks are executed but no new tasks are accepted.
  • awaitTermination(long timeout, TimeUnit unit):
    • Waits for the executor to terminate for the given timeout. Returns true if termination occurs within the timeout, false otherwise.
  • shutdownNow():
    • Attempts to stop all running tasks and halts task processing. It returns a list of tasks that were waiting to be executed.

Best Practices:

  • Always include exception handling for InterruptedException.
  • Use a timeout value that suits your application’s requirements.
  • Avoid forcing a shutdown (shutdownNow()) unless absolutely necessary, as it can leave tasks in an inconsistent state.

By following these steps, you can shut down your ExecutorService gracefully and ensure that resources are properly released.

How do I submit multiple tasks and get results using invokeAll?

By I Wayan Saryada in Concurrency 0 Comment

To submit multiple tasks and get results using invokeAll in Java, you can make use of the ExecutorService. The invokeAll method submits a collection of Callable tasks to the executor and waits for all of them to complete. Once completed, it returns a list of Future objects, each representing the result of a corresponding task.

Here’s how it works:

  1. Create a collection of Callable tasks: These tasks are units of work that the executor will execute in parallel.
  2. Submit the tasks using invokeAll: The invokeAll method blocks until all tasks are complete or timed out.
  3. Retrieve the results from the Future objects: Each Future object allows you to get the result of its corresponding task or check for exceptions.

Example Code

package org.kodejava.util.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class InvokeAllExample {
   public static void main(String[] args) {
      // Create a fixed thread pool
      ExecutorService executorService = Executors.newFixedThreadPool(3);

      // Create a collection of Callable tasks
      List<Callable<String>> tasks = new ArrayList<>();
      tasks.add(() -> {
         // Simulate doing some work
         Thread.sleep(1000);
         return "Task 1 completed";
      });
      tasks.add(() -> {
         Thread.sleep(2000);
         return "Task 2 completed";
      });
      tasks.add(() -> {
         Thread.sleep(1500);
         return "Task 3 completed";
      });

      try {
         // Submit the tasks and wait for all of them to complete
         List<Future<String>> results = executorService.invokeAll(tasks);

         // Iterate through the futures to retrieve the results
         for (Future<String> future : results) {
            try {
               // Get the result of each task
               System.out.println(future.get());
            } catch (ExecutionException e) {
               System.err.println("Task encountered an issue: " + e.getMessage());
            }
         }
      } catch (InterruptedException e) {
         System.err.println("Task execution was interrupted: " + e.getMessage());
      } finally {
         // Shutdown the executor service
         executorService.shutdown();
      }
   }
}

Explanation:

  1. ExecutorService:
    • A thread pool is created (Executors.newFixedThreadPool(3)), which allows up to 3 threads to run simultaneously.
  2. List of Callable tasks:
    • Each task implements the Callable interface and returns a result. For example, the tasks simulate work by Thread.sleep() and return a string.
  3. invokeAll Method:
    • executorService.invokeAll(tasks) submits all tasks at once and blocks until all tasks are complete.
  4. Retrieving Results:
    • The method returns a list of Future objects, where future.get() is used to retrieve the result of each task.
  5. Exceptions:
    • Handle InterruptedException (if the current thread is interrupted) and ExecutionException (if a task fails with an exception).
  6. Shutdown the Executor:
    • Always call shutdown() to properly terminate the executor service and release resources.

Output:

Task 1 completed
Task 3 completed
Task 2 completed

(Note: The order may vary since the tasks run concurrently.)

Keynotes:

  • Use ExecutorService to manage thread pools efficiently.
  • The invokeAll method blocks until all tasks are complete.
  • Handle exceptions like InterruptedException and ExecutionException.
  • Always shut down the executor service to free resources.