如何扩展AbstractExecutorService以确保线程多样性

Question

我一直在成功地使用ExecutorService实现来管理线程池。

现在，我发现了一种情况，即使用数十个并发线程将大型数据对象插入数据仓库。数据可能来自不同的来源。当两个或多个线程从同一个源插入数据时，死锁会以一定的速度发生。这些死锁是很好理解的，只有通过才能避免，而不是同时拥有来自同一源的数据的两个线程()。

我读过关于ExecutorService接口内部的文章，发现抽象类AbstractExecutorService是一个很好的起点。我还设想，覆盖newTaskFor方法对目标来说就足够了。

对于如何用源的名称标记提交的线程，我有点不知所措，以后在newTaskFor中根据这个标签选择一个好的线程。主要思想是，只有当提交的线程的标签在运行的线程中是唯一的时，才能运行它。运行线程的数量当然会受到模拟(最大池大小，当前标签多样性)的限制。

在这方面有什么帮助吗？提前谢了。

Answer 1

Executor框架的目的是减轻您自己操作线程的负担。

显然，您希望依次处理来自任何一个源的输入，每次处理一个。要做到这一点，只需定义多个executor服务，每个服务专用于特定的数据源。让每个人都成为单螺纹执行器。

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

从数据源“A”向executorService_A提交任务，从数据源“B”提交任务到executorService_B，等等。

我看不出许多，甚至几百个单线程执行器服务…的问题。除非你有几个核心和许多同时执行的任务。如果是这样的话，那么合并为更少的执行器服务，其中一个用于几个输入，仍然是单线程的。为源A、B和C提供“ABC”执行器服务，为源X、Y和Z提供另一种“XYZ”执行器服务。

您可以通过使用Map< String , ExecutorService >来管理对哪个执行程序服务的源的跟踪，其中映射的键是源名称，而映射的值是executor服务。

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" ) ) ) ;

为了安全起见，我会制作地图不可修改。

map = Map.copyOf( map ) ;