Akka TCP是AkkaIO模块下的内容,所有的AkkaIO API只能通过mangaer对象来访问。当我们使用Akka IO时,第一步要先获取合适的manager对象,如下获取TcpManager:
ActorRef tcpManager = Tcp.get(getContext().getSystem()).manager();
manager是一个处理底层IO资源(selectors,channels)的actor,同时可以给任务实例化workers。
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在激活连接之后,通过如下代码重新处理接收的消息
getContext().become(connected(getSender()));
TcpServer在激活连接的时候指定代理Actor来处理后续的消息
final ActorRef handler = getContext().actorOf(Props.create(SimplisticHandler.class));
getSender().tell(TcpMessage.register(handler), getSelf());
在交互的时候,使用TcpMessage.write方法发送给TcpManager创建的connection actor
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即可。
- TcpMessage.close,发送FIN消息,但是不等待对方的确认。还没有处理的写数据会被清空,如果成功关闭连接,监听者会收到Closed事件。
- TcpMessage.confirmedClose,发送方会发送FIN信号,但是还可以收到数据,一直到对方也关闭连接。发送方后续的写数据会被清空,监听者会收到ConfirmedClosed事件
- TcpMessage.abort,立刻关闭连接,发送
RST
消息给对方连接,如果成功关闭连接,会收到Aborted消息
当对方关闭的时候,监听者会收到PeerClosed
事件,同时连接会自动关闭。如果想要支持办关闭的连接,在使用TcpMessage.register的时候设置keepOpenOnPeerClosed
参数为true,这个时候连接仍然保持打开,除非收到了上面三个的任意一个关闭消息。
当某一方出错的时候ErrorClosed
事件会发送给监听者,同时强制关闭连接。
所有的关闭连接事件都是ConnectionClosed
的子事件,所以监听这一个事件就可以处理所有关闭的事件。
读写拥塞控制
Akka的连接actor没有内部的缓冲区,这代表着只有当数据在内核中满的时候才会发送出去,需要我们自己对Akka TCP的读写做一些拥塞控制。
对于对方连接回复写(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的拥塞控制。
参考: