将带有Executor服务的Java多线程迁移到Akka

Question

我只是想知道是否有可能替换用Java的Executor服务编写的旧的多线程代码到Akka。我对此没有什么疑问。

Is akka actor runs in their own thread? 

How Threads will be assigned for the Actors ?

What are the pros and cons of migration of it is possible?

目前我使用固定线程池多线程，并提交一个可调用的。

示例代码,

public class KafkaConsumerFactory {

    private static Map<String,KafkaConsumer> registry = new HashMap<>();

    private static ThreadLocal<KafkaConsumer> consumers = new ThreadLocal<KafkaConsumer>(){
        @Override
        protected KafkaConsumer initialValue() {
            return new KafkaConsumer(createConsumerConfig());
        }
    };

    static {
        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run() {
                registry.forEach((tid,con) -> {
                    try{
                        con.close();
                    } finally {
                        System.out.println("Yes!! Consumer for " + tid + " is closed.");
                    }
                });
            }
        });
    }

    private static Properties createConsumerConfig() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "newcon-grp5");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", KafkaKryoSerde.class.getName());
        return props;
    }


    public static <K,V> KafkaConsumer<K,V> createConsumer(){
        registry.put(Thread.currentThread().getName(),consumers.get());
        return consumers.get();
    }
}

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

public class KafkaNewConsumer {
    public static int MAX_THREADS = 10;
    private ExecutorService es = null;
    private boolean stopRequest = false;




    public static void main(String[] args){
        KafkaNewConsumer knc = new KafkaNewConsumer();
        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run(){
                knc.es.shutdown();
                try {
                    knc.es.awaitTermination(500, TimeUnit.MILLISECONDS);
                } catch (InterruptedException ignored) {

                }finally {
                    System.out.println("Finished");
                }
            }
        });

        knc.consumeTopic("rtest3",knc::recordConsuemer);

    }

    public void recordConsuemer(ConsumerRecord<?,?> record){
        String result = new StringJoiner(": ")
                .add(Thread.currentThread().getName())
                .add("ts").add(String.valueOf(record.timestamp()))
                .add("offset").add(String.valueOf(record.offset()))
                .add("data").add(String.valueOf(record.value()))
                .add("value-len").add(String.valueOf(record.serializedValueSize()))
                .toString();
        System.out.println(result);
    }
    public void  consumeTopic(String topicName, Consumer<ConsumerRecord<?,?>> fun){
        KafkaConsumer con= KafkaConsumerFactory.createConsumer();
        int paritions = con.partitionsFor(topicName).size();
        int noOfThread = (MAX_THREADS < paritions) ? MAX_THREADS :paritions;
         es = Executors.newFixedThreadPool(noOfThread);
        con.close();
        for(int i=0;i<noOfThread;i++){
            es.submit(()->{
                KafkaConsumer consumer = KafkaConsumerFactory.createConsumer();
                try{
                    while (!stopRequest){
                        consumer.subscribe(Collections.singletonList(topicName));
                        ConsumerRecords<?,?> records = consumer.poll(5000);

                        records.forEach(fun);
                        consumer.commitSync();
                    }
                }catch(Exception e){
                    e.printStackTrace();
                } finally {
                    consumer.close();
                }
            });
        }
    }
}

但没有解释它如何变得比线程更快？

我尝试了一些示例Akka（activator中的Akka示例）代码，并在所有Actor中打印了Thread.currentThread.getName，发现创建了名为（helloakka-akka.actor.default-dispatcher-x）的不同调度程序线程。

但怎么做？谁在创建那些线程？它们的配置在哪里？线程和执行元之间的映射关系是什么？

如果我需要100个线程来并行执行some任务的部分，我是否需要创建100个参与者并向每个参与者发送1条消息？或者我需要创建一个actor并在它的队列中放入100个消息，它将分叉成100个线程。

真的很困惑

Answer 1

迁移到actor系统对于基于executor的系统来说不是一个小任务，但这是可以完成的。它要求您重新思考您设计系统的方式，并考虑参与者的影响。例如，在一个线程架构中，您为业务流程创建一些处理程序，将其放在一个runnable中，并让它在线程上执行操作。这对于一个行动者范式来说是完全不合适的。您必须重新构建系统以处理消息传递和使用消息调用任务。此外，您还必须改变对业务流程的思考方式，从命令式方法转变为基于消息的方法。请考虑购买产品的简单任务。我会假设你知道如何在一个执行者做这件事。在actor系统中,您可以执行以下操作：

（购买产品）->UserActor->（BillCredit Card）->CCProcessing Actor->（购买已批准和已记账的项目）->库存管理器->……依此类推

在每个阶段，括号中的是一个异步消息，发送给相关的执行元，该执行元执行业务逻辑，然后将消息转发给流程中的下一个执行元。

现在这只是创建基于actor的系统的一种方法，还有很多其他技术，但核心的基本原则是，您不能将其作为一组步骤来考虑，而应将其作为每个步骤独立运行的集合来考虑。然后，消息按常规顺序通过系统，但您无法确定顺序，甚至无法确定消息是否会到达那里，因此您必须在语义上进行设计来处理这一点。在上面的系统中，我可能会让另一个参与者每两分钟检查一次未提交给Billing的孤立订单。当然，这意味着我的消息需要是理想的，以确保如果我发送他们第二次是可以的，他们不会向用户支付两次账单。

我知道我没有处理您的具体示例，我只是想为您提供一些上下文，即actor不仅仅是创建执行器的另一种方式（好吧，我想您可以用这种方式滥用它们，但这是不可取的），而是一种完全不同的设计范式。这是一个非常值得学习的范例，如果你有了飞跃，你就再也不会想做执行者了。