Java, divide incoming work uniformly via hashing in multithreaded evnironments

by

It sounds like you need FIFO ordering per dispatch id, so the ideal would be to have dispatch queues as the abstraction. That would explain your concern about hashing as not providing uniform distribution, as some dispatch queues may be more active than others and unfairly balanced among workers. By separating the queue from the worker, you retain FIFO semantics and evenly spread out the work.

An inactive library that provides this abstraction is HawtDispatch. It is Java 6 compatible.

A very simple Java 8 approach is to use CompletableFuture as a queuing mechanism, ConcurrentHashMap for registration, and an Executor (e.g. ForkJoinPool) for computing. See EventDispatcher for an implementation of this idea, where registration is explicit. If your dispatchers are more dynamic then you may need to periodically prune the map. The basic idea is as follows.

ConcurrentMap<String, CompletableFuture<Void>> dispatchQueues = ...

public CompletableFuture<Void> dispatch(String queueName, Runnable task) {
  return dispatchQueues.compute(queueName, (k, queue) -> {
    return (queue == null)
        ? CompletableFuture.runAsync(task)
        : queue.thenRunAsync(task);
  });
}

Update (JDK7)

A backport of the above idea would be translated with Guava into something like,

ListeningExecutorService executor = ...
Striped<Lock> locks = Striped.lock(256);
ConcurrentMap<String, ListenableFuture<?>> dispatchQueues = ...

public ListenableFuture<?> dispatch(String queueName, final Runnable task) {
  Lock lock = locks.get(queueName);
  lock.lock();
  try {
    ListenableFuture<?> future = dispatchQueues.get(queueName);
    if (future == null) {
      future = executor.submit(task);
    } else {
      final SettableFuture<Void> next = SettableFuture.create();
      future.addListener(new Runnable() {
        try {
          task.run();
        } finally {
          next.set(null);
        }
      }, executor);
      future = next;
    }
    dispatchQueues.put(queueName, future);
  } finally {
    lock.unlock();
  }
}

More Related Contents:

  1. How to wait for all threads to finish, using ExecutorService?
  2. Is there an advantage to use a Synchronized Method instead of a Synchronized Block?
  3. Why is creating a Thread said to be expensive?
  4. When and how should I use a ThreadLocal variable?
  5. Why is i++ not atomic?
  6. What is the difference between Thread.start() and Thread.run()?
  7. Java Wait and Notify: IllegalMonitorStateException
  8. Are static variables shared between threads?
  9. How do synchronized static methods work in Java and can I use it for loading Hibernate entities?
  10. Is iterating ConcurrentHashMap values thread safe?
  11. Why use a ReentrantLock if one can use synchronized(this)?
  12. How to solve the “Double-Checked Locking is Broken” Declaration in Java?
  13. Is it safe to get values from a java.util.HashMap from multiple threads (no modification)?
  14. Java’s Fork/Join vs ExecutorService – when to use which?
  15. How do I implement task prioritization using an ExecutorService in Java 5?
  16. Constructor synchronization in Java
  17. More efficient way for pausing loop wanted
  18. Javafx: Difference between javafx.concurent and Platform.runLater?
  19. Instruction reordering & happens-before relationship [duplicate]
  20. When should we use Java’s Thread over Executor?
  21. Why should I use concurrent characteristic in parallel stream with collect?
  22. ThreadPoolExecutor Block When its Queue Is Full?
  23. Why doesn’t volatile in java 5+ ensure visibility from another thread?
  24. Concurrent JUnit testing
  25. How to pause/resume all threads in an ExecutorService in Java?
  26. Handling Exceptions for ThreadPoolExecutor
  27. Multithreading Socket communication Client/Server
  28. notifyAll() throws IllegalMonitorStateException
  29. Will a chain of method calls (CompletableFuture API) execute asynchronously if the first method in a chain is asynchronous?
  30. ExecutorCompletionService? Why do need one if we have invokeAll?

Leave a Comment