java - 如何使用 Akka 对字符串求和？

Question

我有一个 txt 文件，其中包含 100000 个字符串，例如 ID;amount。在这个文件中我有 1000 个唯一的 ID。我想计算每个 id 的金额并将这些字符串写入文件。因此结果文件应包含 1000 个具有唯一 id 的字符串。

这是我的代码:主类

public class Main {

private static ActorSystem system;

public static void main(String[] args) throws Exception{
    system = ActorSystem.create("ClientSystem");
    system.actorOf(Props.create(ClientActor.class));
}
}

客户端 Actor

public class ClientActor extends UntypedActor{


    ActorRef worker = getContext().actorOf(Props.create(WorkerActor.class));
    @Override
    public void preStart() throws Exception{
        FileInputStream fis = new FileInputStream(new File("100000.txt"));
        BufferedReader br = new BufferedReader(new InputStreamReader(fis));
        String line = null;
        while ((line = br.readLine()) != null) {
            worker.tell(line, getSelf());
        }
        br.close();
    }
    @Override
    public void onReceive(Object o) throws Exception {

    }
}

worker Actor

public class WorkerActor extends UntypedActor {

    Map sum = new HashMap();

    private String getId(String s){
       return s.substring(0, s.indexOf(";"));
    }

    private String getAmount(String s){
        return s.substring(s.lastIndexOf(";") + 1);
    }

    @Override
    public void onReceive(Object o) throws Exception {
        sum.put(getId((String)o), sum.get(getId((String)o) + getAmount(getAmount((String)o))));

//        clientActor.tell("", clientActor);
    }
}

在 ClientActor 中，我用 amount;id 字符串解析文件，并将这些字符串发送到 WorkerActor，在那里我对结果求和并将其放入总和映射中。现在我想将此总和映射写入文件，但无法理解如何执行此操作(如何知道所有行都已处理？)。

Answer 1

使用Akka Streams可以更好地实现这样的工作。 ，特别是因为您不应该像上面的简单示例中那样在 Actor 内执行任何阻塞操作；这些应该使用单独的调度程序进行隔离，这样即使 IO 被阻塞，系统也可以保持响应。 Akka Streams 会为您处理这个问题，因此更容易做正确的事情。

你可以编写这样的代码:

val futureBytesWritten = 
  Source.file(fIn)
  .via(Framing.delimiter(ByteString(System.lineSeparator), Int.MaxValue, true).map(_.utf8String.split(";")))
  .fold(Map[String, Long]().withDefaultValue(0l))({
    (m, v) => m.updated(v(0), m(v(0)) + Integer.parseInt(v(1)))
  })
  .mapConcat(_ map { case (k, v) => k+";"+v+System.lineSeparator })
  .runWith(Sink.file(fOut))

提供有关 Streaming IO 的文档的这一部分阅读并阅读quickstart about Akka Streams .