Networking

Computers running on the Internet communicate to each other using either the Transmission Control Protocol (TCP) or the User Datagram Protocol (UDP).

When you write Java programs that communicate over the network, you are programming at the application layer.

https://docs.oracle.com/javase/tutorial/networking/overview/networking.html

基本概念

• TCP (Transmission Control Protocol) is a connection-based protocol that provides a reliable flow of data between two computers.
• UDP (User Datagram Protocol) is a protocol that sends independent packets of data, called datagrams, from one computer to another with no guarantees about arrival.
• The TCP and UDP protocols use ports to map incoming data to a particular process running on a computer.
• The Hypertext Transfer Protocol (HTTP), File Transfer Protocol (FTP), and Telnet are all examples of applications that require a reliable communication channel.

URL

URL is an acronym for Uniform Resource Locator and is a reference (an address) to a resource on the Internet. For instance: http://example.com

• Protocol identifier
• Resource name
  • Host Name
  • Filename
  • Port Number
  • Reference

Reading Directly from a URL

public static void main(String[] args) throws Exception {
    URL oracle = new URL("http://www.oracle.com/");
    BufferedReader in = new BufferedReader(
        new InputStreamReader(oracle.openStream()));

    String inputLine;
    while ((inputLine = in.readLine()) != null)
        System.out.println(inputLine);
    in.close();
}

Connecting to a URL

try {
    URL myURL = new URL("http://example.com/");
    URLConnection myURLConnection = myURL.openConnection();
    myURLConnection.connect();
} 
catch (MalformedURLException e) { 
    // new URL() failed
    // ...
} 
catch (IOException e) {   
    // openConnection() failed
    // ...
}

Reading from a URLConnection

public static void main(String[] args) throws Exception {
    URL oracle = new URL("http://www.oracle.com/");
    URLConnection yc = oracle.openConnection();
    BufferedReader in = new BufferedReader(new InputStreamReader(
                                yc.getInputStream()));
    String inputLine;
    while ((inputLine = in.readLine()) != null) 
        System.out.println(inputLine);
    in.close();
}

Writing to a URLConnection

String stringToReverse = "Hello World";
URL url = new URL("http://example.com/servlet/ReverseServlet");
URLConnection connection = url.openConnection();
connection.setDoOutput(true);

OutputStreamWriter out = new OutputStreamWriter(connection.getOutputStream());
out.write("string=" + stringToReverse);
out.close();

BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
String decodedString;
while ((decodedString = in.readLine()) != null) {
    System.out.println(decodedString);
}
    in.close();

HTTP客户端（Java 11+）

Java 11引入了新的HttpClient，支持HTTP/2、异步请求和响应式流。

// 同步请求
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create("https://api.example.com/data"))
    .timeout(Duration.ofSeconds(10))
    .header("Content-Type", "application/json")
    .POST(HttpRequest.BodyPublishers.ofString("{\"key\":\"value\"}"))
    .build();

HttpResponse<String> response = client.send(request, 
    HttpResponse.BodyHandlers.ofString());
System.out.println(response.statusCode());
System.out.println(response.body());

异步HTTP请求

// 异步请求
HttpClient client = HttpClient.newBuilder()
    .executor(Executors.newFixedThreadPool(10))
    .build();

HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create("https://api.example.com/data"))
    .build();

CompletableFuture<HttpResponse<String>> future = 
    client.sendAsync(request, HttpResponse.BodyHandlers.ofString());

future.thenApply(HttpResponse::body).thenAccept(System.out::println).join();

HTTP/2特性

HttpClient client = HttpClient.newBuilder()
    .version(HttpClient.Version.HTTP_2)  // 使用HTTP/2
    .build();

HTTP/2 vs HTTP/1.1：

• HTTP/2：多路复用，头部压缩，服务器推送
• HTTP/1.1：每个请求一个连接，头部冗余

WebSocket

WebSocket提供全双工通信，适合实时应用（长连接）。

WebSocket webSocket = HttpClient.newHttpClient()
    .newWebSocketBuilder().buildAsync(URI.create("ws://localhost:8080/chat"), 
        new WebSocket.Listener() {
            @Override
            public void onOpen(WebSocket webSocket) {
                webSocket.sendText("Hello Server", true);
            }
            @Override
            public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
                System.out.println("Received: " + data);
                return null;
            }
        }).join();

WebSocket服务器（JSR 356）

@ServerEndpoint("/chat")
public class ChatEndpoint {
    @OnOpen
    public void onOpen(Session session) {System.out.println("Client connected: " + session.getId()); }
    @OnMessage
    public void onMessage(String message, Session session) {
        session.getOpenSessions().forEach(s -> {
            try {
                s.getBasicRemote().sendText(message);  //广播消息给所有客户端
            } catch (IOException e) { e.printStackTrace(); });
    }
    @OnClose
    public void onClose(Session session) {System.out.println("Client disconnected: " + session.getId());}
}

Sockets

URLs and URLConnections provide a relatively high-level mechanism for accessing resources on the Internet. Sometimes your programs require lower-level network communication, for example, when you want to write a client-server application.

To communicate over TCP, a client program and a server program establish a connection to one another. Each program binds a socket to its end of the connection. To communicate, the client and the server each reads from and writes to the socket bound to the connection.

Socket

A socket is one endpoint of a two-way communication link between two programs running on the network. A socket is bound to a port number so that the TCP layer can identify the application that data is destined to be sent to.

Server listen

A server runs on a specific computer and has a socket that is bound to a specific port number.

The server just waits, listening to the socket for a client to make a connection request.

ServerSocket serverSocket = new ServerSocket(80);

Client connect

The client knows the hostname of the machine on which the server is running and the port number on which the server is listening. The client also needs to identify itself to the server so it binds to a local port number that it will use during this connection.

Socket echoSocket = new Socket(hostName, portNumber);

Server accept

If everything goes well, the server accepts the connection. Upon acceptance, the server gets a new socket bound to the same local port and also has its remote endpoint set to the address and port of the client. It needs a new socket so that it can continue to listen to the original socket for connection requests while tending to the needs of the connected client.

Socket clientSocket = serverSocket.accept();     

Client-server communication

The client and server can now communicate by writing to or reading from their sockets.

序列化与网络传输

Java序列化

// 发送对象
ObjectOutputStream oos = new ObjectOutputStream(
    socket.getOutputStream());
oos.writeObject(new User("Alice", 25));
oos.flush();

// 接收对象
ObjectInputStream ois = new ObjectInputStream(
    socket.getInputStream());
User user = (User) ois.readObject();

但是：性能差，序列化后体积大；安全性问题；版本兼容性差

高效序列化方案

1. JSON（Jackson/Gson）

// 使用Jackson
ObjectMapper mapper = new ObjectMapper();
String json = mapper.writeValueAsString(user);
User user = mapper.readValue(json, User.class);

// 优点：可读性好，跨语言
// 缺点：体积较大，性能一般

2. Protobuf（推荐）

message User {
   string name = 1;
   int32 age = 2;
}
protoc --java_out=./src/main/java user.proto

// 序列化
UserProto.User user = UserProto.User.newBuilder().setName("Alice").setAge(25).build();
byte[] data = user.toByteArray();
// 反序列化
UserProto.User parsed = UserProto.User.parseFrom(data);

体积小，性能高；跨语言支持；向后兼容

Socket选项

TCP_NODELAY

禁用Nagle算法，减少延迟。

Socket socket = new Socket("localhost", 8080);
socket.setTcpNoDelay(true);  // 禁用Nagle算法

Nagle算法：将多个小数据包合并发送，减少网络开销
适用场景：实时性要求高的应用（如游戏、实时通信）

SO_REUSEADDR

允许端口复用，解决TIME_WAIT状态下的端口占用问题。

ServerSocket serverSocket = new ServerSocket();
serverSocket.setReuseAddress(true);  // 允许端口复用
serverSocket.bind(new InetSocketAddress(8080));

作用：允许在TIME_WAIT状态下重新绑定端口
适用场景：服务器重启时快速恢复

SO_KEEPALIVE

启用TCP保活机制，检测连接是否存活。

Socket socket = new Socket("localhost", 8080);
socket.setKeepAlive(true);  // 启用保活机制

保活机制：

• 默认2小时无数据后发送探测包
• 探测包失败后关闭连接
• 可检测到网络中断、主机崩溃等情况

SO_TIMEOUT

设置Socket读取超时时间。

Socket socket = new Socket("localhost", 8080);
socket.setSoTimeout(5000);  // 5秒超时

超时类型：

• 连接超时：connect()操作的超时
• 读取超时：read()操作的超时
• 写入超时：通常不超时，由操作系统控制

缓冲区大小设置

Socket socket = new Socket("localhost", 8080);
socket.setReceiveBufferSize(64 * 1024);  // 64KB接收缓冲区
socket.setSendBufferSize(64 * 1024);     // 64KB发送缓冲区

缓冲区大小：

• 默认值：通常8KB（系统相关）
• 建议值：根据网络延迟和带宽调整
• 过大：浪费内存，增加延迟
• 过小：频繁系统调用，降低性能

UDP编程

DatagramSocket和DatagramPacket

// UDP服务器
DatagramSocket socket = new DatagramSocket(8888);
byte[] buffer = new byte[1024];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

socket.receive(packet);  // 阻塞等待数据
String message = new String(packet.getData(), 0, packet.getLength());
System.out.println("Received: " + message);

// 发送响应
InetAddress clientAddress = packet.getAddress();
int clientPort = packet.getPort();
byte[] response = "Hello Client".getBytes();
DatagramPacket responsePacket = new DatagramPacket(
    response, response.length, clientAddress, clientPort);
socket.send(responsePacket);

UDP客户端

// UDP客户端
DatagramSocket socket = new DatagramSocket();
String message = "Hello Server";
byte[] data = message.getBytes();

InetAddress serverAddress = InetAddress.getByName("localhost");
DatagramPacket packet = new DatagramPacket(
    data, data.length, serverAddress, 8888);
socket.send(packet);

// 接收响应
byte[] buffer = new byte[1024];
DatagramPacket response = new DatagramPacket(buffer, buffer.length);
socket.receive(response);
String reply = new String(response.getData(), 0, response.getLength());
System.out.println("Reply: " + reply);

UDP广播和组播

// 广播
DatagramSocket socket = new DatagramSocket();
socket.setBroadcast(true);

InetAddress broadcastAddress = InetAddress.getByName("255.255.255.255");
byte[] data = "Broadcast Message".getBytes();
DatagramPacket packet = new DatagramPacket(
    data, data.length, broadcastAddress, 8888);
socket.send(packet);

// 组播（Multicast）
MulticastSocket multicastSocket = new MulticastSocket(8888);
InetAddress group = InetAddress.getByName("230.0.0.1");
multicastSocket.joinGroup(group);

byte[] buffer = new byte[1024];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
multicastSocket.receive(packet);

广播：服务发现、局域网通知；组播：视频直播、多人游戏

例子

https://docs.oracle.com/javase/tutorial/networking/sockets/examples/EchoClient.java

https://docs.oracle.com/javase/tutorial/networking/sockets/examples/EchoServer.java

https://tools.ietf.org/html/rfc862

回顾

多个客户端

But it's still blocking! And too many threads cause performance issues!

Thread Context Switch

https://en.wikipedia.org/wiki/Context_switch

http://tutorials.jenkov.com/java-concurrency/costs.html

https://eli.thegreenplace.net/2018/measuring-context-switching-and-memory-overheads-for-linux-threads/

https://blog.tsunanet.net/2010/11/how-long-does-it-take-to-make-context.html

Non-blocking I/O

With non-blocking I/O, we can use a single thread to handle multiple concurrent connections.

• Buffer
• Channel
• Selector

Selector

Java NIO has a class called "Selector" that allows a single thread to examine I/O events on multiple channels. That is, this selector can check the readiness of a channel for operations, such as reading and writing.

看个例子

https://medium.com/coderscorner/tale-of-client-server-and-socket-a6ef54a74763

https://github.com/arukshani/JavaIOAndNIO

Selector底层实现：操作系统IO多路复用

Java NIO的Selector底层使用操作系统提供的IO多路复用机制：

epoll工作原理

// Java NIO Selector
Selector selector = Selector.open();
channel.configureBlocking(false);
channel.register(selector, SelectionKey.OP_READ);

// 底层调用
// epoll_create() - 创建epoll实例
// epoll_ctl() - 注册/修改/删除事件
// epoll_wait() - 等待事件就绪

• O(1)时间复杂度：事件就绪时直接通知，无需遍历
• 支持大量连接：不受文件描述符限制
• 边缘触发模式：减少系统调用次数

Selector 使用中的挑战

• 事件处理逻辑复杂
  • 所有事件处理都在一个循环中
  • 业务逻辑与IO处理耦合，代码难以维护和扩展
• 单线程性能瓶颈
  • 如果某个连接处理耗时会阻塞其他连接，无法充分利用多核CPU
  • 需要引入线程池处理业务逻辑
• 事件分发不够清晰
  • 需要手动判断事件类型，缺少统一的事件分发机制
  • 错误处理分散在各处

Reactor 模式

https://en.wikipedia.org/wiki/Reactor_pattern

Reactor模式

单线程Reactor

Reactor线程
├── 监听连接事件
├── 分发事件到Handler
└── 处理IO读写

Handler
├── 处理连接
├── 处理读取
└── 处理写入

Reactor模式

核心组件：

• Selector：事件分发器，监听多个Channel
• ServerSocketChannel：监听连接事件
• SocketChannel：处理IO读写事件
• SelectionKey：事件类型（ACCEPT、READ、WRITE）

多线程Reactor

Main Reactor
├── 监听连接事件
└── 分发到Sub Reactor

Sub Reactor (多个)
├── 处理IO读写
└── 分发到Worker线程池

Worker线程池
└── 处理业务逻辑

Netty

Netty是基于NIO的高性能网络框架，实现了Reactor模式。

核心优势：

• 高性能：基于NIO，零拷贝
• 易用性：封装复杂API
• 可扩展：丰富的编解码器
• 稳定：Netty 4.x广泛使用

Netty Reactor

Netty核心组件

// 服务器启动
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

try {
    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                ch.pipeline().addLast(new StringDecoder()).addLast(new StringEncoder()).addLast(new ServerHandler());
            }
        })
        .option(ChannelOption.SO_BACKLOG, 128)
        .childOption(ChannelOption.SO_KEEPALIVE, true);
    
    ChannelFuture future = bootstrap.bind(8080).sync();
    future.channel().closeFuture().sync();
} finally {
    workerGroup.shutdownGracefully();
    bossGroup.shutdownGracefully();
}

ChannelHandler

public class ServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        String message = (String) msg;
        System.out.println("Received: " + message);
        
        // 响应
        ctx.writeAndFlush("Echo: " + message);
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

异步IO（AIO）

AIO概述

Java 7引入了AIO（Asynchronous I/O），提供真正的异步非阻塞IO。

AIO vs NIO：

• NIO：非阻塞IO，需要轮询检查就绪状态
• AIO：异步IO，操作系统完成后回调通知

AsynchronousServerSocketChannel

// AIO服务器
AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(8080));
// 异步接受连接
serverChannel.accept(null, new CompletionHandler<
    AsynchronousSocketChannel, Void>() {
    @Override
    public void completed(AsynchronousSocketChannel channel, Void attachment) {
        // 继续接受下一个连接
        serverChannel.accept(null, this);
        // 处理当前连接
        handleClient(channel);
    }
    @Override
    public void failed(Throwable exc, Void attachment) {
        exc.printStackTrace();
    }
});

AsynchronousSocketChannel

// 异步读取
ByteBuffer buffer = ByteBuffer.allocate(1024);
channel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
    @Override
    public void completed(Integer bytesRead, ByteBuffer buffer) {
        if (bytesRead > 0) {
            buffer.flip();
            byte[] data = new byte[buffer.remaining()];
            buffer.get(data);
            System.out.println("Received: " + new String(data));
            
            // 继续读取
            buffer.clear();
            channel.read(buffer, buffer, this);
        }
    }
    
    @Override
    public void failed(Throwable exc, ByteBuffer buffer) {
        exc.printStackTrace();
    }
});
---

### AsynchronousSocketChannel
```java
// 异步写入
ByteBuffer writeBuffer = ByteBuffer.wrap("Hello".getBytes());
channel.write(writeBuffer, null, new CompletionHandler<Integer, Void>() {
    @Override
    public void completed(Integer bytesWritten, Void attachment) {
        System.out.println("Written: " + bytesWritten + " bytes");
    }
    
    @Override
    public void failed(Throwable exc, Void attachment) {
        exc.printStackTrace();
    }
});

AIO适用场景

优点：

• 真正的异步，无需轮询
• 回调机制，代码更清晰
• 适合大量连接

缺点：

• Windows上实现不完善（使用线程池模拟）
• Linux上基于epoll，性能与NIO相当
• 代码复杂度较高

网络性能调优

系统参数调优

# Linux TCP参数调优
# /etc/sysctl.conf

# 增加TCP连接队列
net.core.somaxconn = 2048

# TCP快速回收TIME_WAIT
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 30

# 增加文件描述符限制
ulimit -n 65535

# TCP缓冲区大小
net.core.rmem_max = 16777216
net.core.wmem_max = 16777216
net.ipv4.tcp_rmem = 4096 87380 16777216
net.ipv4.tcp_wmem = 4096 65536 16777216

JVM网络参数

# JVM启动参数
-XX:+UseG1GC                    # 使用G1垃圾回收器
-XX:MaxDirectMemorySize=2g      # 直接内存大小
-Djava.net.preferIPv4Stack=true # 优先使用IPv4
-Djava.net.preferIPv6Addresses=false

关键参数：

• 直接内存：NIO使用直接内存，需要足够空间
• GC优化：网络应用产生大量临时对象，需要低延迟GC
• IPv4/IPv6：根据实际需求选择