通过 http 的 Akka 对象流答案

【问题标题】：Akka stream of objects over http通过 http 的 Akka 对象流
【发布时间】：2016-04-03 15:29:12
【问题描述】：

我有一段代码（见下文），它生成了一个服务器，该服务器回显它从端口 6001 接收到的每个 ByteString 流。该示例还定义了一个连接到服务器并发送包含列表的 ByteString 流的客户端从字母“a”到“z”的字符数。

此时我的问题是，akka 是否提供了一种通过 http 发送和接收对象流而不是 ByStreams 的方法？例如，Client 类的对象。

如果是这样，我如何发送和接收这样的对象流？你能给我一个显示如何执行它的sn-p吗？

Akka 文档对于非玩具示例的用户友好性不强...

感谢您的帮助

公共类 TcpEcho {

/**
 * Use without parameters to start both client and server.
 *
 * Use parameters `server 0.0.0.0 6001` to start server listening on port
 * 6001.
 *
 * Use parameters `client 127.0.0.1 6001` to start client connecting to
 * server on 127.0.0.1:6001.
 *
 */
public static void main(String[] args) throws IOException {
    if (args.length == 0) {
        ActorSystem system = ActorSystem.create("ClientAndServer");
        InetSocketAddress serverAddress = new InetSocketAddress("127.0.0.1", 6000);
        server(system, serverAddress);
        client(system, serverAddress);
    } else {
        InetSocketAddress serverAddress;
        if (args.length == 3) {
            serverAddress = new InetSocketAddress(args[1], Integer.valueOf(args[2]));
        } else {
            serverAddress = new InetSocketAddress("127.0.0.1", 6000);
        }
        if (args[0].equals("server")) {
            ActorSystem system = ActorSystem.create("Server");
            server(system, serverAddress);
        } else if (args[0].equals("client")) {
            ActorSystem system = ActorSystem.create("Client");
            client(system, serverAddress);
        }
    }
}

public static void server(ActorSystem system, InetSocketAddress serverAddress) {
    final ActorMaterializer materializer = ActorMaterializer.create(system);

    final Sink<IncomingConnection, CompletionStage<Done>> handler = Sink.foreach(conn -> {
        System.out.println("Client connected from: " + conn.remoteAddress());
        conn.handleWith(Flow.<ByteString> create(), materializer);
    });

    final CompletionStage<ServerBinding> bindingFuture = Tcp.get(system)
            .bind(serverAddress.getHostString(), serverAddress.getPort()).to(handler).run(materializer);

    bindingFuture.whenComplete((binding, throwable) -> {
        System.out.println("Server started, listening on: " + binding.localAddress());
    });

    bindingFuture.exceptionally(e -> {
        System.err.println("Server could not bind to " + serverAddress + " : " + e.getMessage());
        system.terminate();
        return null;
    });

}

public static void client(ActorSystem system, InetSocketAddress serverAddress) {
    final ActorMaterializer materializer = ActorMaterializer.create(system);

    final List<ByteString> testInput = new ArrayList<>();
    for (char c = 'a'; c <= 'z'; c++) {
        testInput.add(ByteString.fromString(String.valueOf(c)));
    }

    Source<ByteString, NotUsed> responseStream = Source.from(testInput)
            .via(Tcp.get(system).outgoingConnection(serverAddress.getHostString(), serverAddress.getPort()));

    CompletionStage<ByteString> result = responseStream.runFold(ByteString.empty(), (acc, in) -> acc.concat(in),
            materializer);

    result.whenComplete((success, failure) -> {

        if (failure != null) {
            System.err.println("Failure: " + failure.getMessage());
        } else {
            System.out.println("Result: " + success.utf8String());
        }
        System.out.println("Shutting down client");
        system.terminate();

    });
}

}

【问题讨论】：

你看过this example关于如何创建Bidi流的，我认为它或多或少地做了你想要的吗？
我没看，但我会看的。无论如何，您有更好的想法/建议吗？谢谢

标签： java stream akka akka-stream

【解决方案1】：

akka.stream.{javadsl,scaladsl}.Framing 包含可帮助您构建一致消息的实用程序。例如，您可以通过Framing.simpleFramingProtocolEncoder(maxLength) 发送消息以自动添加长度信息。在另一端，Framing.simpleFramingProtocolDecoder(maxLength) 将负责根据包含的长度信息对消息进行解码。

如果您想操作普通对象，您只需在通过编码器发送它们之前将它们序列化为ByteString，并在从解码器接收到它们的表示后从ByteString 反序列化它们。

【讨论】：