Akka

Akka是一个用于构建高性能、分布式和容错系统的工具包及运行时环境，特别适合于Ja va和Scala开发者在JVM平台上开发高并发应用。

它采用Actor模型来简化并发编程。Actor作为最小的计算单元，通过异步消息传递进行通信，从而有效地解决了传统并发问题。

Akka还提供了丰富的容错机制和位置透明性，使得构建可伸缩的分布式系统更为便捷。

此外，Akka还包括其他模块，如Akka Streams用于处理数据流，并支持反压机制，增强了系统在处理大量数据时的稳定性和效率。

akka本身是用scala编写的库，可以运行在JVM上。在c#中，用的是c#版本的akka，github地址如下：

• scala: github.com/akka/akka
• c#: github.com/akkadotnet/…

背景

由于现实情况限制，公司现有某项目的部分模块采用了C#开发，并通过Akka进行了进程间通信。现计划对选定模块进行重构，以Ja va语言替代C#实现。

要想让重构后的模块与现有系统保持兼容，新的Ja va项目必须作为被调用的服务端，遵循Akka协议。

只有这样，才能保证与原有的C#调用方接口实现无缝对接。虽然全面转向Ja va是个挺理想的方案，但鉴于存在多种复杂的现实因素，目前暂时没办法对整个项目进行彻底的语言迁移和技术栈统一。

因此，当前任务的核心在于，在Ja va环境中建立一个遵循Akka规范的服务端，以便于与遗留的C#客户端进行有效的通信交互。

初探

我把c#的akka源码下载下来，查阅相关文档并基于这个demo进行调试。

在c#版本akka的文档，关于序列化部分，描述了用到的所有序列化器：

我使用demo调试中，发现请求与响应对象的序列化/反序列化只用到了NewtonSoftJsonSerializer，这让我一开始以为事情的解决如此简单。

当时我以为，只需要在服务端以字符串接收并解析json即可。

事实并非如此简单。Akka Remote类的序列化器都可能用到，我们的项目中，框架把几种序列化器都用了，除了json这个。

再探

我开始深入debug c#版本的源码，同时把scala版本源码也下载下来进行调试分析。

scala版本必须是最远古的那几个版本。

两个版本的实现非常接近。c#的通信框架基于dot.netty实现，scala版本基于netty 3.x的版本实现。

通过源码分析整个实现流程之后，我对协议的实现有了大致思路。

基于netty实现时，流的编码采用LengthFieldPrepender，解码采用LengthFieldBasedFrameDecoder。

但这里有一个关键点：c#的实现使用小端网络字节序，而Ja va默认实现是大端网络字节序，因此这个地方实现的时候必须指定小端。

难解

我基于ja va实现一个简易版的服务端，尝试接收demo中c#客户端的请求，并在打印日志后给予响应。

结果，生活总是没有想象那么美好。

Akka请求分类

akka协议的请求分为两种：

• 指令，控制类消息
• 消息

消息又分为下面两种：

• 系统消息（akka自身）
• 业务消息（我们发的）

所以在编写解码器的时候，必须考虑指令类消息的解码：

private AkkaPduCodec.AkkaPdu decodeInstruction(final WireFormats.AkkaControlMessage message) {

        switch (message.getCommandType()) {
            case ASSOCIATE -> {
                if (message.hasHandshakeInfo()) {
                    WireFormats.AkkaHandshakeInfo handshakeInfo = message.getHandshakeInfo();

                    String cookie = null;
                    if (handshakeInfo.hasCookie()) {
                        cookie = handshakeInfo.getCookie();
                    }
                    WireFormats.AddressData origin = handshakeInfo.getOrigin();
                    Address address = new Address(origin.getProtocol(), origin.getSystem(), origin.getHostname(), origin.getPort());
                    HandshakeInfo info = new HandshakeInfo(address, (int) handshakeInfo.getUid(), Option.apply(cookie));
                    return new AkkaPduCodec.Associate(info);
                }
            }
            case DISASSOCIATE -> {
                return new AkkaPduCodec.Disassociate(AssociationHandle.Unknown$.MODULE$);
            }

            case DISASSOCIATE_SHUTTING_DOWN -> {
                return new AkkaPduCodec.Disassociate(AssociationHandle.Shutdown$.MODULE$);
            }
            case DISASSOCIATE_QUARANTINED -> {
                return new AkkaPduCodec.Disassociate(AssociationHandle.Quarantined$.MODULE$);
            }
            case HEARTBEAT -> {
                return AkkaPduCodec.Heartbeat$.MODULE$;
            }
            default -> log.error("Unknown：{}", message);
        }
        return null;
}

这里面用到的一些类，实际是我引用了scala版本的akka依赖。

注意，akka通信也采用了protobuf的序列化。而如何判断是否是指令类型还是其它消息类型，我的实现也很简单：

        WireFormats.AkkaProtocolMessage message = WireFormats.AkkaProtocolMessage.parseFrom(bytes);

        if (message.hasPayload()) {
            AkkaPduCodec.Payload payload = new AkkaPduCodec.Payload(ByteString.
                    fromByteBuffer(message.getPayload().asReadOnlyByteBuffer()));
            out.add(payload);
        } else if (message.hasInstruction()) {
            AkkaPduCodec.AkkaPdu akkaPdu = decodeInstruction(message.getInstruction());
            out.add(akkaPdu);
        } else {
            log.error("Unknown：{}", message);
        }

我以为问题如此简单，便兴致勃勃地进行了实现，和本地的c#的客户端联调也通过了。

但当我和实际业务系统进行联调的时候，才发现并不简单。

• 业务系统并不是使用的如此简单
• 消息的序列化也并不是json
• 并且出现了其它我无法解析的消息格式

这把我之前的认知和推断都推翻了。好难。

好难解

我对业务系统的配置和源码进行了分析，发现还使用了Remote DeathWatch:

同时，我进行了抓包分析并结合源码，查询这个命令的请求情况。

结果发现，它会根据配置的间隔定时请求，还是有不少额外开销的。

通过源码分析，这个请求由客户端发送到服务端。服务端本身的实现是进行相关监听器的注册处理，但是不需要给予客户端相关响应。

这样我实现起来就方便多了，只管接收这个请求，不需要响应。

但是比较麻烦的是，不同的请求用的反序列化器不一样。

我本身以为只是固定的一两个，结果发现不是。实现这一两个已经耗费不少精力，如果要考虑所有情况，把所有的都实现一遍，成本太高。

完美解决

仔细核对c#和scala版本akka的序列化及网络传输部分的实现，包括使用的protobuf定义。

然后把c#版本的定义拿过来在ja va里调整过后，重新实现。

在不懈努力下，重新写出了一个通用的协议的序列化与反序列化版本。

下面是一些debug看到的格式示例：

envelope {
  recipient {
    path: "akka.tcp://server@localhost:4567/"
  }
  message {
    message: "n20301nMb0220b32Gakka.tcp://client@localhost:53151/system/inbox-1#1492312352002320AkkaDemo.Proto.Request, AkkaDemo22bb022204user2222b022216DispatchServer"
    serializerId: 6
  }
  sender {
    path: "akka.tcp://client@localhost:53151/user/dispatch-client-8-0#1086482885"
  }
  seq: 18446744073709551615
}


ack {
  cumulativeAck: 1
}
envelope {
  recipient {
    path: "akka.tcp://server@localhost:4567/"
  }
  message {
    message: "nb20203204RWHB22nb022206system2222b022216remote-watcher"
    serializerId: 6
  }
  sender {
    path: "akka.tcp://client@localhost:58542/system/remote-watcher#2129229977"
  }
  seq: 18446744073709551615
}

最终，终于成功实现了这个服务端。

其它方案

实现的时候，也考虑过是否能直接用scala版本的实现直接对接c#版本。

最终发现两个问题，让我放弃了这个方案：

1. 上面示例的包格式里c#版本的带有一个命名空间，是自动生成的，scala实现应该是包路径，两边完全对不上。

2. 远古版本采用的是3.x的netty，而项目里又存在4.x的netty，这就导致了版本冲突。要是想用4.x的，我得把scala实现的通信协议改成4.x的netty再重新打包。

   可我对sbt这个工具（scala用的就是它）不太了解，感觉了解它所花费的时间，我都能把这活儿干完了。而且新版本的实现已经不再使用netty了，用的是另一套基于udp协议的，这就使得和c#版本的对接变得更困难了。

总之，直接对接的方案，改动scala实现并重新打包的成本，大于我直接实现的成本。

所以考虑之后就放弃了。