2.Netty TCP服务器（TcpServer）

07-13 1376阅读

Netty TCP服务器（TcpServer）

Reactor Netty提供了一个易于使用和配置的TcpServer。它隐藏Netty了创建TCP服务器所需的大部分功能并增加了Reactive Streams背压（Reactive Streams是具有无阻塞背压的异步流处理的标准）

启动和停止

import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
	public static void main(String[] args) {
		DisposableServer server =
				TcpServer.create()		// 创建一个TcpServer 准备好配置的实例
				         .bindNow();	// 以阻塞方式启动服务器并等待它完成初始化
		server.onDispose()
		      .block();
	}
}

2.Netty TCP服务器（TcpServer）

启动和停止（主机和端口）

要在特定的host和上提供服务port，可以将以下配置应用到TCP服务器：

import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
	public static void main(String[] args) {
		DisposableServer server =
				TcpServer.create()
				         .host("localhost") 
				         .port(8080)        
				         .bindNow();
		server.onDispose()
		      .block();
	}
}

import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class MultiAddressApplication {
	public static void main(String[] args) {
		TcpServer tcpServer = TcpServer.create();
         
         // 配置第一台TCP服务器主机、端口
		DisposableServer server1 = tcpServer
				.host("localhost") 
				.port(8080)        
				.bindNow();
         // 配置第二台TCP服务器主机、端口
		DisposableServer server2 = tcpServer
				.host("0.0.0.0") 
				.port(8081)      
				.bindNow();
		Mono.when(server1.onDispose(), server2.onDispose())
				.block();
	}
}

急切初始化

默认情况下，TcpServer资源的初始化是按需进行的。这意味着bind operation吸收了初始化和加载所需的额外时间：

①：事件循环组

②：本机传输库（使用本机传输时）

③：用于安全性的本机库（在的情况下OpenSsl）

import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
	public static void main(String[] args) {
		TcpServer tcpServer =
				TcpServer.create()
				         .handle((inbound, outbound) -> inbound.receive().then());
		tcpServer.warmup()  // 初始化并加载事件循环组、本机传输库和用于安全性的本机库
		         .block();
		DisposableServer server = tcpServer.bindNow();
		server.onDispose()
		      .block();
	}
	
}

消费客户端数据

为了从连接的客户端接收数据，必须附加一个I/O 处理程序。I/O 处理程序可以访问NettyInbound以读取数据。

import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
	public static void main(String[] args) {
		DisposableServer server =
				TcpServer.create()
                          // 从连接的客户端接收数据
				         .handle((inbound, outbound) -> inbound.receive().then()) 
				         .bindNow();
		server.onDispose()
		      .block();
	}
	
}

生命周期回调

TcpServer提供了以下生命周期回调以便扩展。

回调函数	描述
doOnBind	在服务器通道即将绑定时调用
doOnBound	在绑定服务器通道时调用
doOnChannelInit	初始化通道时调用
doOnConnection	连接远程客户端时调用
doOnUnbound	当服务器通道未绑定时调用

使用doOnConnection和doOnChannelInit回调：

import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
import java.util.concurrent.TimeUnit;
public class Application {
	public static void main(String[] args) {
		DisposableServer server =
				TcpServer.create()
                         // 当连接远程客户端时，Netty管道使用ReadTimeoutHandler进行扩展
				         .doOnConnection(conn ->
				             conn.addHandler(new ReadTimeoutHandler(10, TimeUnit.SECONDS))) 
                         // 初始化通道时，Netty管道使用LoggingHandler进行扩展
				         .doOnChannelInit((observer, channel, remoteAddress) ->
				             channel.pipeline()
				                    .addFirst(new LoggingHandler("reactor.netty.examples")))
				         .bindNow();
		server.onDispose()
		      .block();
	}
	
}