Java:具有多个队列的执行器服务

Question

要求:

1. 我将消息分为不同的类型，例如 Type1、Type2 ... Type100。
2. 我想并行执行不同类型的消息。比方说在 10 个线程中，但是所有相同类型的消息必须一个接一个地执行。执行顺序无关紧要。
3. 一旦线程完成TypeX 的所有消息。它应该开始处理另一种类型。

我经历了不同的答案: 他们中的大多数建议执行程序服务来处理多线程。 假设我们创建执行器服务，如

ExecutorService executorService = Executors.newFixedThreadPool(10);

但是一旦我们使用 executorService.submit(runnableMessage);

提交消息

我们无法控制将特定类型的消息分配给特定线程。

解决方案:

创建单线程执行器数组

ExecutorService[] pools = new ExecutorService[10];

并最初传递 Type1, Type2 ... Type10 的消息 然后，如果任何执行程序已完成执行，则将 Type11 分配给它并继续这样做，直到所有类型都得到处理。

有没有更好的方法呢？

类似于具有多个队列的执行器服务，我可以在其中将每种类型的消息推送到不同的队列？

Answer 1

我建议看一下 Akka .他们提供了一个更适合这种用例的 Actor 框架。除了定义您自己的 ExecutorService 接口(interface)实现之外，JDK 提供的默认实现并没有提供对调度的太多控制。

创建一个硬编码的 ExecutionServices 数组不会非常动态或健壮，尤其是因为每个 ExecutionService 都有一个线程池。可以用散列映射替换数组，然后将其放在 ExecutionService 的自定义实现之后，这样的优势是对调用者隐藏这些细节，但它不会解决拥有这么多线程池的线程浪费问题。

在 Akka 中，每个 Actor 都有自己的与之关联的消息队列。每个 Actor 在其自己的线程中有效地运行，一次处理其队列中的每条消息。 Akka 将管理跨多个 Actor 的线程共享。因此，如果您要为每种消息类型创建一个 Actor，然后将这些消息与这些 Actor 一起排队，那么您将达到这样的目标，即每种消息类型一次最多由一个线程处理，同时仅由一个池支持线程数。

技术演示:

Maven 对 Akka 的依赖。

    <dependency>
        <groupId>com.typesafe.akka</groupId>
        <artifactId>akka-actor_2.11</artifactId>
        <version>2.4.17</version>
    </dependency>

Java 8 代码。复制并粘贴到一个 Java 文件中，然后在您的 IDE 中运行 main 方法。

package com.softwaremosaic.demos.akka;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;


public class ActorDemo {

    public static void main( String[] args ) throws InterruptedException {
        // The following partitioner will spread the requests over
        // multiple actors, which I chose to demonstrate the technique.
        // You will need to change it to one that better maps the the
        // jobs to your use case.   Remember that jobs that get mapped
        // to the same key, will get executed in serial (probably
        // but not necessarily) by the same thread.
        ExecutorService exectorService = new ActorExecutionService( job -> job.hashCode()+"" );

        for ( int i=0; i<100; i++ ) {
            int id = i;
            exectorService.submit( () -> System.out.println("JOB " + id) );
        }

        exectorService.shutdown();
        exectorService.awaitTermination( 1, TimeUnit.MINUTES );

        System.out.println( "DONE" );
    }

}


class ActorExecutionService extends AbstractExecutorService {

    private final ActorSystem                              actorSystem;
    private final Function<Runnable, String>               partitioner;
    private final ConcurrentHashMap<String,ActorRef>       actors = new ConcurrentHashMap<>();

    public ActorExecutionService( Function<Runnable,String> partitioner ) {
        this.actorSystem = ActorSystem.create("demo");
        this.partitioner = partitioner;
    }


    public void execute( Runnable command ) {
        String partitionKey = partitioner.apply( command );

        ActorRef actorRef = actors.computeIfAbsent( partitionKey, this::createNewActor );

        actorRef.tell( command, actorRef );
    }

    private ActorRef createNewActor( String partitionKey ) {
        return actorSystem.actorOf( Props.create(ExecutionServiceActor.class), partitionKey );
    }


    public void shutdown() {
        actorSystem.terminate();
    }

    public List<Runnable> shutdownNow() {
        actorSystem.terminate();

        try {
            awaitTermination( 1, TimeUnit.MINUTES );
        } catch ( InterruptedException e ) {
            throw new RuntimeException( e );
        }

        return Collections.emptyList();
    }

    public boolean isShutdown() {
        return actorSystem.isTerminated();
    }

    public boolean isTerminated() {
        return actorSystem.isTerminated();
    }

    public boolean awaitTermination( long timeout, TimeUnit unit ) throws InterruptedException {
        actorSystem.awaitTermination();

        return actorSystem.isTerminated();
    }
}

 class ExecutionServiceActor extends UntypedActor {
    public void onReceive(Object message) throws Exception {
        if (message instanceof Runnable) {
            ((Runnable) message).run();
        } else {
            unhandled(message);
        }
    }
}

注意上面的代码将以未定义的顺序打印 1-100。由于批处理(Akka 这样做是为了获得额外的性能优势)，订单看起来主要是串行的。但是，随着不同线程穿插在工作中，您会看到数字有些随机性。每个作业运行的时间越长，分配给 Akka 线程池的线程越多，使用的分区键越多，底层 CPU 内核越多，序列可能变得越随机。