How to extend AbstractExecutorService to ensure thread diversity

user_8520235 0
Viewed 60

I've been successfully using ExecutorService implementations to manage thread pools.

Now I've found a situation where dozens of concurrent threads are used to insert large data objects into a data warehouse. Data may come from different sources. When two or more threads are inserting data from the same source, deadlocks happen at a certain rate. These deadlocks are well understood and can only be avoided by not having two threads with data from the same source at the same time.

I have read about the internals of the ExecutorService interface and found the abstract class AbstractExecutorService can be a good starting point. I also imagine that overriding the newTaskFor method could be enough for the objective.

I am a bit lost about how to label the submitted threads with the name of the source, to later choose a good one in newTaskFor based on that label. The main idea is to run a submitted thread only if its label is unique among the running threads. The number of running threads will be of course limited by the miminum of (maximum pool size, current label diversity) at any time.

Any help on this? Thanks on advance.

java-threads threadpoolexecutor executorservice java
1 Answers

The point of the Executors framework is to relieve you of the burden of manipulating threads yourself.

Apparently you want to process inputs from any one source serially, one at a time. To accomplish that, simply define multiple executor services, each one dedicated to a particular data source. And make each a single-threaded executor.

ExecutorService executorService_A = Executors.newSingleThreadExecutor() ;
ExecutorService executorService_B = Executors.newSingleThreadExecutor() ;
ExecutorService executorService_C = Executors.newSingleThreadExecutor() ;

Submit tasks from data source “A” to executorService_A, tasks from data source “B” to executorService_B, and so on.

I don’t see a problem with many, even hundreds, of single-threaded executor services… unless you have few cores and many simultaneously executing tasks. If that is the case, then consolidate to fewer executor services, one for several inputs, still single-threaded. Have an “ABC” executor service for sources A, B, and C. Have another “XYZ” executor service for sources X, Y, and Z.

You could manage the tracking of which source to which executor service by using a Map< String , ExecutorService > where the key of the map is the source name, and the value of the map is an executor service.

Map< String , ExecutorService > map = new HashMap<>() ;

map.put( "A" , Executors.newSingleThreadExecutor() ) ;       // A, B, & C share the same single-threaded executor service. 
map.put( "B" , Objects.requireNonNull( map.get( "A" ) ) ) ; 
map.put( "C" , Objects.requireNonNull( map.get( "A" ) ) ) ;

map.put( "X" , Executors.newSingleThreadExecutor() ) ;       // X, Y, & Z share the same single-threaded executor service. 
map.put( "Y" , Objects.requireNonNull( map.get( "X" ) ) ) ;
map.put( "Z" , Objects.requireNonNull( map.get( "X" ) ) ) ;

For safety, I would then make that map unmodifiable.

map = Map.copyOf( map ) ;
user_6427060
Related