如何在Netty UDP通信中实现可靠握手机制？

DevOps 小满

2026-7-29

基于Netty实现UDP可靠请求-响应Handler

核心原理

UDP本身是无连接、不可靠的传输协议，要实现服务器注册这类需要可靠传输的逻辑，必须手动实现以下机制：

请求ID匹配：每个请求携带唯一ID，响应返回相同ID，用于关联请求和响应
超时重传：请求发送后启动超时定时器，未收到响应则重试，直到达到最大重试次数
响应去重：收到响应后从待处理请求表中移除对应ID，避免重复处理
非阻塞等待：利用Netty的Promise/Future实现异步等待，不阻塞NIO线程

代码实现

1. 自定义UDP消息协议

先定义统一的请求/响应消息结构，包含匹配ID、消息类型和业务 payload：

import java.io.*;

public class UdpRequestResponseMsg {
    private long requestId;
    private byte type; // 0=注册请求,1=注册响应,2=服务器列表请求,3=服务器列表响应
    private byte[] payload;

    // Getter & Setter
    public long getRequestId() { return requestId; }
    public void setRequestId(long requestId) { this.requestId = requestId; }
    public byte getType() { return type; }
    public void setType(byte type) { this.type = type; }
    public byte[] getPayload() { return payload; }
    public void setPayload(byte[] payload) { this.payload = payload; }

    // 序列化：转为二进制字节数组
    public byte[] serialize() throws IOException {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);
        dos.writeLong(requestId);
        dos.writeByte(type);
        dos.writeInt(payload.length);
        dos.write(payload);
        dos.close();
        return bos.toByteArray();
    }

    // 反序列化：从字节数组恢复消息
    public static UdpRequestResponseMsg deserialize(byte[] data) throws IOException {
        UdpRequestResponseMsg msg = new UdpRequestResponseMsg();
        ByteArrayInputStream bis = new ByteArrayInputStream(data);
        DataInputStream dis = new DataInputStream(bis);
        msg.setRequestId(dis.readLong());
        msg.setType(dis.readByte());
        int payloadLen = dis.readInt();
        byte[] payload = new byte[payloadLen];
        dis.read(payload);
        msg.setPayload(payload);
        dis.close();
        return msg;
    }
}

2. 客户端请求-响应Handler

负责发送请求、管理待处理请求、处理超时重传和响应匹配：

import io.netty.channel.*;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.concurrent.Promise;
import io.netty.util.internal.ThreadLocalRandom;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class UdpClientRequestHandler extends SimpleChannelInboundHandler<DatagramPacket> {
    // 待处理请求映射：requestId -> 请求条目
    private final Map<Long, PendingRequest> pendingRequests = new ConcurrentHashMap<>();
    // 最大重试次数
    private static final int MAX_RETRY = 3;
    // 请求超时时间（毫秒）
    private static final long TIMEOUT_MS = 2000;

    // 待处理请求条目
    private static class PendingRequest {
        final Promise<UdpRequestResponseMsg> promise;
        final UdpRequestResponseMsg requestMsg;
        final Channel channel;
        int retryCount;

        PendingRequest(Promise<UdpRequestResponseMsg> promise, UdpRequestResponseMsg requestMsg, Channel channel) {
            this.promise = promise;
            this.requestMsg = requestMsg;
            this.channel = channel;
            this.retryCount = 0;
        }
    }

    // 对外暴露的发送请求方法，返回Future异步结果
    public ChannelFuture<UdpRequestResponseMsg> sendRequest(UdpRequestResponseMsg requestMsg, InetSocketAddress brokerAddr) {
        // 生成唯一请求ID（用ThreadLocalRandom避免全局锁）
        long requestId = ThreadLocalRandom.current().nextLong();
        requestMsg.setRequestId(requestId);

        Channel channel = ctx().channel();
        Promise<UdpRequestResponseMsg> promise = channel.eventLoop().newPromise();
        PendingRequest pendingReq = new PendingRequest(promise, requestMsg, channel);
        pendingRequests.put(requestId, pendingReq);

        // 发送请求数据包
        sendDatagram(requestMsg, brokerAddr, channel);
        // 启动超时检测任务
        scheduleTimeoutCheck(requestId);

        return promise;
    }

    private void sendDatagram(UdpRequestResponseMsg msg, InetSocketAddress targetAddr, Channel channel) {
        try {
            byte[] data = msg.serialize();
            DatagramPacket packet = new DatagramPacket(Unpooled.wrappedBuffer(data), targetAddr);
            channel.writeAndFlush(packet);
        } catch (IOException e) {
            failPendingRequest(msg.getRequestId(), e);
        }
    }

    private void scheduleTimeoutCheck(long requestId) {
        ctx().channel().eventLoop().schedule(() -> {
            PendingRequest req = pendingRequests.get(requestId);
            if (req == null) return;

            if (req.retryCount >= MAX_RETRY) {
                // 重试耗尽，标记请求失败
                failPendingRequest(requestId, new TimeoutException("Request timed out after " + MAX_RETRY + " retries"));
                return;
            }

            // 重试发送
            req.retryCount++;
            sendDatagram(req.requestMsg, brokerAddr, req.channel);
            // 重新调度超时检测
            scheduleTimeoutCheck(requestId);
        }, TIMEOUT_MS, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
        // 解析响应消息
        byte[] data = new byte[packet.content().readableBytes()];
        packet.content().readBytes(data);
        UdpRequestResponseMsg response = UdpRequestResponseMsg.deserialize(data);

        // 匹配并完成对应的请求Promise
        PendingRequest req = pendingRequests.remove(response.getRequestId());
        if (req != null && !req.promise.isDone()) {
            req.promise.setSuccess(response);
        }
        // 无匹配请求：可能是超时后收到的重复响应，直接忽略
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 通道异常时，所有待处理请求标记失败
        pendingRequests.values().forEach(req -> {
            if (!req.promise.isDone()) {
                req.promise.setFailure(cause);
            }
        });
        pendingRequests.clear();
        ctx.close();
    }

    private void failPendingRequest(long requestId, Throwable cause) {
        PendingRequest req = pendingRequests.remove(requestId);
        if (req != null && !req.promise.isDone()) {
            req.promise.setFailure(cause);
        }
    }

    private ChannelHandlerContext ctx() {
        return context();
    }
}

3. Broker端响应Handler

负责接收客户端请求、处理业务逻辑并返回带相同请求ID的响应：

import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.Unpooled;

import java.net.InetSocketAddress;
import java.io.IOException;

public class UdpBrokerResponseHandler extends SimpleChannelInboundHandler<DatagramPacket> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
        // 解析客户端请求
        byte[] data = new byte[packet.content().readableBytes()];
        packet.content().readBytes(data);
        UdpRequestResponseMsg request = UdpRequestResponseMsg.deserialize(data);

        InetSocketAddress clientAddr = packet.sender();
        UdpRequestResponseMsg response = new UdpRequestResponseMsg();
        response.setRequestId(request.getRequestId()); // 复用请求ID做匹配

        // 根据消息类型处理业务
        switch (request.getType()) {
            case 0: // 服务器注册请求
                byte[] registerResp = handleServerRegister(request.getPayload());
                response.setType((byte) 1);
                response.setPayload(registerResp);
                break;
            case 2: // 获取服务器列表请求
                byte[] serverList = getServerList();
                response.setType((byte) 3);
                response.setPayload(serverList);
                break;
            default:
                // 未知消息类型，返回错误响应
                response.setType((byte) -1);
                response.setPayload("UNKNOWN_TYPE".getBytes());
                break;
        }

        // 发送响应数据包
        sendResponse(response, clientAddr, ctx);
    }

    // 服务器注册业务逻辑示例
    private byte[] handleServerRegister(byte[] payload) {
        // 解析服务器信息（比如IP、端口、负载），保存到内存/数据库
        return "REGISTER_SUCCESS".getBytes();
    }

    // 获取服务器列表业务逻辑示例
    private byte[] getServerList() {
        // 从存储中读取可用服务器列表，序列化后返回
        return "192.168.1.100:9000,192.168.1.101:9001".getBytes();
    }

    private void sendResponse(UdpRequestResponseMsg response, InetSocketAddress targetAddr, ChannelHandlerContext ctx) throws IOException {
        byte[] respData = response.serialize();
        DatagramPacket respPacket = new DatagramPacket(Unpooled.wrappedBuffer(respData), targetAddr);
        ctx.writeAndFlush(respPacket);
    }
}

关键注意事项

非阻塞设计：所有超时任务、IO操作都在Netty的EventLoop线程中执行，不会阻塞业务线程，符合NIO模型
可靠性增强：通过请求ID匹配、超时重传确保关键请求（如服务器注册）的可靠性，重试次数和超时时间可根据游戏场景调整
协议优化：示例中用了简单的二进制序列化，实际生产环境推荐用Protobuf/FlatBuffers等高效序列化框架，减少数据包大小，降低传输延迟
线程安全：用ConcurrentHashMap存储待处理请求，确保多线程环境下的安全访问

内容的提问来源于stack exchange，提问作者TraversingOnTarget

如何在Netty UDP通信中实现可靠握手机制？

核心原理 #

代码实现 #

1. 自定义UDP消息协议 #

2. 客户端请求-响应Handler #

3. Broker端响应Handler #