Akka TCP使用指南

时间 2019-11-06

原文原文链接

Akka TCP是AkkaIO模块下的内容，全部的AkkaIO API只能经过mangaer对象来访问。当咱们使用Akka IO时，第一步要先获取合适的manager对象，以下获取TcpManager：html

ActorRef tcpManager = Tcp.get(getContext().getSystem()).manager();
复制代码

manager是一个处理底层IO资源(selectors,channels)的actor，同时能够给任务实例化workers。bash

Tcp客户端

第一步是向TcpManager发送TcpMessage.conenct方法。
而后TcpManager会返回一个内部的actor表明链接对象，也可能返回CommandFailed 消息
新的actor对象，而后发送一个Connected 消息给最初发送TcpMessage.connect的actor。
想要激活链接，必需要发送一个TcpMessage.register给链接对象。若是不作这一步，这个链接就是不能用的
链接对象(connection actor)，监听注册的handler，当两边的actor某个关闭的时候，关闭链接对象。而后清除这个链接的资源

@Slf4j
public class AkkaTcpClientDemo extends AbstractActor {

    final InetSocketAddress remote;
    final ActorRef listener;

    public static Props props(InetSocketAddress remote, ActorRef listener) {
        return Props.create(AkkaTcpClientDemo.class, remote, listener);
    }

    public AkkaTcpClientDemo(InetSocketAddress remote, ActorRef listener) {
        this.remote = remote;
        this.listener = listener;
		// 链接
        final ActorRef tcp = Tcp.get(getContext().getSystem()).manager();
        tcp.tell(TcpMessage.connect(remote), getSelf());
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Tcp.Bound.class,msg -> {
                    log.info("Bounded {} ,sender = {}",msg,getSender());
                })
                .match(
                        Tcp.CommandFailed.class,
                        msg -> {
                            log.info("failed msg{},sender {}",msg,getSender());
                            listener.tell("failed ", getSelf());
                            getContext().stop(getSelf());
                        })
                .match(
                        Tcp.Connected.class,
                        msg -> {
                            log.info("Conencted : msg {} sender{}",msg,getSender());
                            listener.tell(msg, getSelf());
                            getSender().tell(TcpMessage.register(getSelf()), getSelf());
                            getContext().become(connected(getSender()));
                            ActorRef conn = getSender();
                            for (int i = 0; i < 10; i++) {
                                conn.tell(TcpMessage.write(ByteString.fromString("Hello" + i)),getSelf());
                            }

                        })
                .build();
    }


    private Receive connected(final ActorRef connection) {
        return receiveBuilder()
                .match(
                        ByteString.class,
                        msg -> {
                            connection.tell(TcpMessage.write((ByteString) msg), getSelf());
                        })
                .match(
                        Tcp.CommandFailed.class,
                        msg -> {
                            // OS kernel socket buffer was full
                        })
                .match(
                        Tcp.Received.class,
                        msg -> {
                            listener.tell(msg.data().utf8String(), getSelf());
                        })
                .matchEquals(
                        "close",
                        msg -> {
                            connection.tell(TcpMessage.close(), getSelf());
                        })
                .match(
                        Tcp.ConnectionClosed.class,
                        msg -> {
                            getContext().stop(getSelf());
                        })
                .match(Tcp.Register.class,msg -> {
                    log.info("register msg {}",msg);
                })
                .build();
    }

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("demo", ConfigFactory.load("my"));
//        ActorRef manager = Tcp.get(system).manager();
        ActorRef listen = system.actorOf(Props.create(ServerHandler.class));
//        system.actorOf( Props.create(AkkaTcpClientDemo.class,new InetSocketAddress("127.0.0.1",1234),listen));
        system.actorOf( Props.create(AkkaTcpClientDemo.class,new InetSocketAddress("30.11.33.16",3307),listen));

    }
}
复制代码

Tcp服务端

发送TcpMessage.bind方法给TcpManager，而后TcpManager会监听在指定的地址端口上
经过TcpMessage.bind方法指定的actor会收到Bound消息，表明服务端已经准备好接收请求了。Bound消息也会包含socket实际绑定的端口
当收到Connected消息以后，发送TcpMessage.Register消息激活当前的链接，在Register方法中也能够指定代理actor来处理后续收到的数据

@Slf4j
public class AkkaTcpServerDemo extends AbstractActor {
    final ActorRef manager;

    public AkkaTcpServerDemo(ActorRef manager) {
        this.manager = manager;
    }

    public static Props props(ActorRef manager) {
        return Props.create(AkkaTcpServerDemo.class, manager);
    }

    @Override
    public void preStart() throws Exception {
        final ActorRef tcp = Tcp.get(getContext().getSystem()).manager();
        tcp.tell(TcpMessage.bind(getSelf(), new InetSocketAddress("localhost", 1234), 100), getSelf());
    }


    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(
                        Tcp.Bound.class,
                        msg -> {
                            log.info("Aihe Bounded {}, sender {}",msg,getSender());
                            manager.tell(msg, getSelf());
                        })
                .match(
                        Tcp.CommandFailed.class,
                        msg -> {
                            log.info("Aihe CommandFailed {}",msg);
                            getContext().stop(getSelf());
                        })
                .match(
                        Tcp.Connected.class,
                        conn -> {
                            log.info("Server Conencted: {},sneder {}",conn,getSender());
                            manager.tell(conn, getSelf());
                            final ActorRef handler =
                                    getContext().actorOf(Props.create(SimplisticHandler.class));
                            getSender().tell(TcpMessage.register(handler), getSelf());

                        })
                .match(Tcp.Register.class,msg -> {
                    log.info("注册消息：{}",msg);
                })
                .build();
    }


    static class SimplisticHandler extends AbstractActor {
        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .match(
                            Tcp.Received.class,
                            msg -> {
                                final ByteString data = msg.data();
                                log.info("收到数据：{}",data.utf8String());
                                getSender().tell(TcpMessage.write(data), getSelf());
                            })
                    .match(
                            Tcp.ConnectionClosed.class,
                            msg -> {
                                getContext().stop(getSelf());
                            })
                    .build();
        }
    }

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("demo", ConfigFactory.load("my"));
        ActorRef manager = Tcp.get(system).manager();
        ActorRef server = system.actorOf(Props.create(AkkaTcpServerDemo.class, manager));

//        system.actorOf(Props.create(AkkaTcpServerDemo.class));

    }
}

复制代码

TCP客户端与服务端细节

TcpClient在激活链接以后，经过以下代码从新处理接收的消息socket

getContext().become(connected(getSender()));
复制代码

TcpServer在激活链接的时候指定代理Actor来处理后续的消息tcp

final ActorRef handler =                            getContext().actorOf(Props.create(SimplisticHandler.class));
getSender().tell(TcpMessage.register(handler), getSelf());
复制代码

在交互的时候，使用TcpMessage.write方法发送给TcpManager建立的connection actoride

ActorRef conn = getSender();
for (int i = 0; i < 10; i++) {
	conn.tell(TcpMessage.write(ByteString.fromString("Hello" + i)),getSelf());
}
复制代码

关闭链接

想要关闭链接发送：TcpMessage.close, TcpMessage.confirmedClose 或者 TcpMessage.abort给conenctor actor便可。ui

TcpMessage.close，发送FIN消息，可是不等待对方的确认。尚未处理的写数据会被清空，若是成功关闭链接，监听者会收到Closed事件。
TcpMessage.confirmedClose，发送方会发送FIN信号，可是还能够收到数据，一直到对方也关闭链接。发送方后续的写数据会被清空，监听者会收到ConfirmedClosed事件
TcpMessage.abort，马上关闭链接，发送RST 消息给对方链接，若是成功关闭链接，会收到Aborted消息

当对方关闭的时候，监听者会收到PeerClosed 事件，同时链接会自动关闭。若是想要支持办关闭的链接，在使用TcpMessage.register的时候设置keepOpenOnPeerClosed 参数为true，这个时候链接仍然保持打开，除非收到了上面三个的任意一个关闭消息。this

当某一方出错的时候ErrorClosed 事件会发送给监听者，同时强制关闭链接。spa

全部的关闭链接事件都是ConnectionClosed 的子事件，因此监听这一个事件就能够处理全部关闭的事件。代理

读写拥塞控制

Akka的链接actor没有内部的缓冲区，这表明着只有当数据在内核中满的时候才会发送出去，须要咱们本身对Akka TCP的读写作一些拥塞控制。code

对于对方链接回复写(back-pressuring writes)数据，有三种模式：

Ack-based：每次Write 命令都携带object数据，这个object不是Tcp.NoAck对象，而后在成功写入全部的数据到socket以后，这个object会返回给发送方。
NACK-based：在发送写命令后，若是上一次的写数据尚未完成就会返回CommandFailed 消息
NACK-based with write suspending: 和NACK-based很像，但在写入失败以后，若是没有收到TcpMessage.resumeWriting方法的话，就不会继续写入。而后回复一个WritingResumed 消息给链接actor。

对于读数据(back-pressuring reads)的控制，有两种模式：

Push-reading：链接actor发送给读数据的actor Received 事件，若是读actor发送TcpMessage.suspendReading方法给链接actor表明，它当前想要挂起接收新数据。只有再发送ResumeReading 事件发送后，表明读actor准备好接收新数据了。
Pull-reading：在接收到Received 事件以后，链接actor自动的挂起接收的数据。直到读actor发送ResumeReading 消息。

读数据是相对于链接actor(connction actor)而言的。

最后

以上是Akka TCP的使用说明，在处理的时候须要考虑Tcp的拥塞控制。

参考：

Akka Tcp