Netty教程 / 第 160 节

第16-20章：实战项目合集

本文档将第16-20章的5个实战项目合并为精简版本，涵盖核心架构和关键代码。

第16章：实战项目1 - 即时通讯系统（IM）

16.1 系统架构

客户端 ←→ IM服务器 ←→ 消息存储
          ↓
    用户会话管理
    消息路由
    离线推送

16.2 核心功能实现

消息协议定义

/**
 * IM消息协议
 * +------+------+--------+----------+
 * | Type | From | To     | Content  |
 * | 1B   | 8B   | 8B     | N bytes  |
 * +------+------+--------+----------+
 */
public class IMMessage {
    public enum Type {
        LOGIN(1),           // 登录
        LOGOUT(2),          // 登出
        SINGLE_CHAT(3),     // 单聊
        GROUP_CHAT(4),      // 群聊
        HEARTBEAT(5);       // 心跳
        
        private final byte code;
        Type(int code) { this.code = (byte) code; }
        public byte getCode() { return code; }
    }
    
    private Type type;
    private long fromUserId;
    private long toUserId;      // 单聊：目标用户ID，群聊：群组ID
    private String content;
    private long timestamp;
    
    // getter/setter...
}

// 编码器
public class IMMessageEncoder extends MessageToByteEncoder<IMMessage> {
    @Override
    protected void encode(ChannelHandlerContext ctx, IMMessage msg, ByteBuf out) {
        out.writeByte(msg.getType().getCode());
        out.writeLong(msg.getFromUserId());
        out.writeLong(msg.getToUserId());
        out.writeLong(msg.getTimestamp());
        
        byte[] content = msg.getContent().getBytes(CharsetUtil.UTF_8);
        out.writeInt(content.length);
        out.writeBytes(content);
    }
}

// 解码器
public class IMMessageDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        if (in.readableBytes() < 25) return;  // 1+8+8+8 = 25
        
        in.markReaderIndex();
        byte typeCode = in.readByte();
        long fromUserId = in.readLong();
        long toUserId = in.readLong();
        long timestamp = in.readLong();
        
        if (in.readableBytes() < 4) {
            in.resetReaderIndex();
            return;
        }
        
        int contentLength = in.readInt();
        if (in.readableBytes() < contentLength) {
            in.resetReaderIndex();
            return;
        }
        
        byte[] content = new byte[contentLength];
        in.readBytes(content);
        
        IMMessage msg = new IMMessage();
        msg.setType(IMMessage.Type.values()[typeCode - 1]);
        msg.setFromUserId(fromUserId);
        msg.setToUserId(toUserId);
        msg.setTimestamp(timestamp);
        msg.setContent(new String(content, CharsetUtil.UTF_8));
        
        out.add(msg);
    }
}

IM服务器

@Component
public class IMServer {
    
    private final SessionManager sessionManager;
    private final MessageRouter messageRouter;
    
    @PostConstruct
    public void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        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 IMMessageDecoder())
                            .addLast(new IMMessageEncoder())
                            .addLast(new IdleStateHandler(60, 30, 0))
                            .addLast(new IMServerHandler(sessionManager, messageRouter));
                    }
                });
        
        bootstrap.bind(8080).sync();
        System.out.println("IM服务器启动成功");
    }
}

// 会话管理器
@Component
public class SessionManager {
    
    private final Map<Long, Channel> userChannels = new ConcurrentHashMap<>();
    private final Map<String, Long> channelUsers = new ConcurrentHashMap<>();
    
    public void addSession(long userId, Channel channel) {
        userChannels.put(userId, channel);
        channelUsers.put(channel.id().asLongText(), userId);
        
        channel.closeFuture().addListener(future -> {
            removeSession(userId);
        });
    }
    
    public void removeSession(long userId) {
        Channel channel = userChannels.remove(userId);
        if (channel != null) {
            channelUsers.remove(channel.id().asLongText());
        }
    }
    
    public Channel getChannel(long userId) {
        return userChannels.get(userId);
    }
    
    public boolean isOnline(long userId) {
        return userChannels.containsKey(userId);
    }
}

// 消息路由器
@Component
public class MessageRouter {
    
    private final SessionManager sessionManager;
    private final OfflineMessageService offlineMessageService;
    
    public void route(IMMessage message) {
        switch (message.getType()) {
            case SINGLE_CHAT:
                routeSingleChat(message);
                break;
            case GROUP_CHAT:
                routeGroupChat(message);
                break;
        }
    }
    
    private void routeSingleChat(IMMessage message) {
        Channel channel = sessionManager.getChannel(message.getToUserId());
        if (channel != null && channel.isActive()) {
            // 在线，直接发送
            channel.writeAndFlush(message);
        } else {
            // 离线，存储离线消息
            offlineMessageService.save(message);
        }
    }
    
    private void routeGroupChat(IMMessage message) {
        long groupId = message.getToUserId();
        List<Long> members = groupService.getGroupMembers(groupId);
        
        for (Long memberId : members) {
            if (memberId.equals(message.getFromUserId())) continue;
            
            Channel channel = sessionManager.getChannel(memberId);
            if (channel != null && channel.isActive()) {
                channel.writeAndFlush(message);
            } else {
                offlineMessageService.save(message);
            }
        }
    }
}

// 业务Handler
public class IMServerHandler extends SimpleChannelInboundHandler<IMMessage> {
    
    private final SessionManager sessionManager;
    private final MessageRouter messageRouter;
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, IMMessage msg) {
        switch (msg.getType()) {
            case LOGIN:
                handleLogin(ctx, msg);
                break;
            case LOGOUT:
                handleLogout(ctx, msg);
                break;
            case SINGLE_CHAT:
            case GROUP_CHAT:
                messageRouter.route(msg);
                break;
            case HEARTBEAT:
                handleHeartbeat(ctx);
                break;
        }
    }
    
    private void handleLogin(ChannelHandlerContext ctx, IMMessage msg) {
        long userId = msg.getFromUserId();
        sessionManager.addSession(userId, ctx.channel());
        
        // 发送登录成功响应
        IMMessage response = new IMMessage();
        response.setType(IMMessage.Type.LOGIN);
        response.setContent("登录成功");
        ctx.writeAndFlush(response);
        
        // 推送离线消息
        pushOfflineMessages(userId, ctx.channel());
    }
    
    private void handleLogout(ChannelHandlerContext ctx, IMMessage msg) {
        sessionManager.removeSession(msg.getFromUserId());
        ctx.close();
    }
    
    private void handleHeartbeat(ChannelHandlerContext ctx) {
        IMMessage response = new IMMessage();
        response.setType(IMMessage.Type.HEARTBEAT);
        ctx.writeAndFlush(response);
    }
}

第17章：实战项目2 - RPC 框架

17.1 RPC 架构

客户端 → 动态代理 → 序列化 → Netty客户端
                                  ↓
                            网络传输
                                  ↓
服务端 ← 反射调用 ← 反序列化 ← Netty服务端

17.2 核心实现

// RPC请求
public class RpcRequest {
    private String requestId;
    private String className;
    private String methodName;
    private Class<?>[] parameterTypes;
    private Object[] parameters;
    // getter/setter...
}

// RPC响应
public class RpcResponse {
    private String requestId;
    private Object result;
    private Throwable error;
    // getter/setter...
}

// RPC服务端
@Component
public class RpcServer {
    
    private final Map<String, Object> serviceMap = new ConcurrentHashMap<>();
    
    public void registerService(String serviceName, Object serviceImpl) {
        serviceMap.put(serviceName, serviceImpl);
    }
    
    @PostConstruct
    public void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        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 RpcDecoder(RpcRequest.class))
                            .addLast(new RpcEncoder(RpcResponse.class))
                            .addLast(new RpcServerHandler(serviceMap));
                    }
                });
        
        bootstrap.bind(8080).sync();
        System.out.println("RPC服务器启动成功");
    }
}

// RPC服务端Handler
public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {
    
    private final Map<String, Object> serviceMap;
    private final ExecutorService executor = Executors.newFixedThreadPool(10);
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) {
        executor.submit(() -> {
            RpcResponse response = new RpcResponse();
            response.setRequestId(request.getRequestId());
            
            try {
                Object result = handle(request);
                response.setResult(result);
            } catch (Throwable t) {
                response.setError(t);
            }
            
            ctx.writeAndFlush(response);
        });
    }
    
    private Object handle(RpcRequest request) throws Exception {
        Object service = serviceMap.get(request.getClassName());
        if (service == null) {
            throw new RuntimeException("服务不存在: " + request.getClassName());
        }
        
        Method method = service.getClass().getMethod(
            request.getMethodName(),
            request.getParameterTypes()
        );
        
        return method.invoke(service, request.getParameters());
    }
}

// RPC客户端
public class RpcClient {
    
    private final String host;
    private final int port;
    private Channel channel;
    private final Map<String, RpcFuture> pendingRequests = new ConcurrentHashMap<>();
    
    public void connect() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline()
                            .addLast(new RpcEncoder(RpcRequest.class))
                            .addLast(new RpcDecoder(RpcResponse.class))
                            .addLast(new RpcClientHandler(pendingRequests));
                    }
                });
        
        channel = bootstrap.connect(host, port).sync().channel();
    }
    
    public Object send(RpcRequest request) throws Exception {
        RpcFuture future = new RpcFuture();
        pendingRequests.put(request.getRequestId(), future);
        
        channel.writeAndFlush(request);
        
        return future.get(5, TimeUnit.SECONDS);
    }
}

// 动态代理
public class RpcProxy {
    
    private final RpcClient client;
    
    @SuppressWarnings("unchecked")
    public <T> T create(Class<T> interfaceClass) {
        return (T) Proxy.newProxyInstance(
            interfaceClass.getClassLoader(),
            new Class<?>[]{interfaceClass},
            (proxy, method, args) -> {
                RpcRequest request = new RpcRequest();
                request.setRequestId(UUID.randomUUID().toString());
                request.setClassName(interfaceClass.getName());
                request.setMethodName(method.getName());
                request.setParameterTypes(method.getParameterTypes());
                request.setParameters(args);
                
                return client.send(request);
            }
        );
    }
}

第18章：实战项目3 - 网关服务器

18.1 网关架构

客户端 → 网关服务器 → 后端服务
         ├── 路由转发
         ├── 负载均衡
         ├── 限流熔断
         └── 监控日志

18.2 核心实现

// 网关服务器
@Component
public class GatewayServer {
    
    private final RouteManager routeManager;
    private final LoadBalancer loadBalancer;
    
    @PostConstruct
    public void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        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 HttpServerCodec())
                            .addLast(new HttpObjectAggregator(65536))
                            .addLast(new GatewayHandler(routeManager, loadBalancer));
                    }
                });
        
        bootstrap.bind(8080).sync();
        System.out.println("网关服务器启动成功");
    }
}

// 路由管理器
@Component
public class RouteManager {
    
    private final Map<String, List<String>> routes = new ConcurrentHashMap<>();
    
    public void addRoute(String path, String backend) {
        routes.computeIfAbsent(path, k -> new CopyOnWriteArrayList<>()).add(backend);
    }
    
    public List<String> getBackends(String path) {
        return routes.get(path);
    }
}

// 负载均衡器
@Component
public class LoadBalancer {
    
    private final AtomicInteger counter = new AtomicInteger(0);
    
    public String select(List<String> backends) {
        if (backends == null || backends.isEmpty()) {
            return null;
        }
        
        // 轮询
        int index = Math.abs(counter.getAndIncrement() % backends.size());
        return backends.get(index);
    }
}

// 网关Handler
public class GatewayHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    
    private final RouteManager routeManager;
    private final LoadBalancer loadBalancer;
    private final RateLimiter rateLimiter = RateLimiter.create(1000);  // 1000 QPS
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
        // 限流
        if (!rateLimiter.tryAcquire()) {
            sendError(ctx, HttpResponseStatus.TOO_MANY_REQUESTS, "请求过于频繁");
            return;
        }
        
        // 路由
        String uri = request.uri();
        List<String> backends = routeManager.getBackends(uri);
        if (backends == null || backends.isEmpty()) {
            sendError(ctx, HttpResponseStatus.NOT_FOUND, "路由不存在");
            return;
        }
        
        // 负载均衡
        String backend = loadBalancer.select(backends);
        
        // 转发请求
        forwardRequest(ctx, request, backend);
    }
    
    private void forwardRequest(ChannelHandlerContext ctx, FullHttpRequest request, String backend) {
        // 创建后端连接
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(ctx.channel().eventLoop())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline()
                            .addLast(new HttpClientCodec())
                            .addLast(new HttpObjectAggregator(65536))
                            .addLast(new SimpleChannelInboundHandler<FullHttpResponse>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx2, FullHttpResponse response) {
                                    // 返回响应给客户端
                                    ctx.writeAndFlush(response.retain());
                                }
                            });
                    }
                });
        
        // 连接后端服务
        String[] parts = backend.split(":");
        bootstrap.connect(parts[0], Integer.parseInt(parts[1]))
                .addListener((ChannelFutureListener) future -> {
                    if (future.isSuccess()) {
                        future.channel().writeAndFlush(request.retain());
                    } else {
                        sendError(ctx, HttpResponseStatus.BAD_GATEWAY, "后端服务连接失败");
                    }
                });
    }
    
    private void sendError(ChannelHandlerContext ctx, HttpResponseStatus status, String message) {
        FullHttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1,
            status,
            Unpooled.copiedBuffer(message, CharsetUtil.UTF_8)
        );
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }
}

第19章：实战项目4 - 游戏服务器

19.1 游戏服务器架构

客户端 → 游戏服务器 → 游戏逻辑
         ├── 房间管理
         ├── 状态同步
         └── 断线重连

19.2 核心实现

// 游戏消息
public class GameMessage {
    public enum Type {
        JOIN_ROOM,      // 加入房间
        LEAVE_ROOM,     // 离开房间
        GAME_ACTION,    // 游戏操作
        STATE_SYNC      // 状态同步
    }
    
    private Type type;
    private long playerId;
    private long roomId;
    private String action;
    private Map<String, Object> data;
    // getter/setter...
}

// 房间管理器
@Component
public class RoomManager {
    
    private final Map<Long, GameRoom> rooms = new ConcurrentHashMap<>();
    private final AtomicLong roomIdGenerator = new AtomicLong(1);
    
    public GameRoom createRoom(int maxPlayers) {
        long roomId = roomIdGenerator.getAndIncrement();
        GameRoom room = new GameRoom(roomId, maxPlayers);
        rooms.put(roomId, room);
        return room;
    }
    
    public GameRoom getRoom(long roomId) {
        return rooms.get(roomId);
    }
    
    public void removeRoom(long roomId) {
        rooms.remove(roomId);
    }
}

// 游戏房间
public class GameRoom {
    
    private final long roomId;
    private final int maxPlayers;
    private final Map<Long, Player> players = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    public GameRoom(long roomId, int maxPlayers) {
        this.roomId = roomId;
        this.maxPlayers = maxPlayers;
        
        // 启动状态同步任务（每100ms同步一次）
        scheduler.scheduleAtFixedRate(this::syncState, 100, 100, TimeUnit.MILLISECONDS);
    }
    
    public boolean addPlayer(Player player) {
        if (players.size() >= maxPlayers) {
            return false;
        }
        players.put(player.getId(), player);
        return true;
    }
    
    public void removePlayer(long playerId) {
        players.remove(playerId);
        if (players.isEmpty()) {
            scheduler.shutdown();
        }
    }
    
    public void broadcast(GameMessage message) {
        players.values().forEach(player -> {
            if (player.getChannel() != null && player.getChannel().isActive()) {
                player.getChannel().writeAndFlush(message);
            }
        });
    }
    
    private void syncState() {
        GameMessage message = new GameMessage();
        message.setType(GameMessage.Type.STATE_SYNC);
        message.setRoomId(roomId);
        
        // 收集所有玩家状态
        Map<String, Object> state = new HashMap<>();
        players.values().forEach(player -> {
            state.put(String.valueOf(player.getId()), player.getState());
        });
        message.setData(state);
        
        broadcast(message);
    }
}

// 玩家
public class Player {
    private long id;
    private String name;
    private Channel channel;
    private Map<String, Object> state = new ConcurrentHashMap<>();
    // getter/setter...
}

第20章：实战项目5 - 物联网数据采集平台

20.1 IoT 平台架构

IoT设备 → 数据采集服务器 → 数据处理
          ├── 设备认证
          ├── 数据解析
          ├── 数据存储
          └── 实时监控

20.2 核心实现

// IoT消息
public class IoTMessage {
    private String deviceId;
    private String messageType;  // DATA, HEARTBEAT, COMMAND
    private Map<String, Object> payload;
    private long timestamp;
    // getter/setter...
}

// IoT服务器
@Component
public class IoTServer {
    
    private final DeviceManager deviceManager;
    private final DataProcessor dataProcessor;
    
    @PostConstruct
    public void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        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 IoTMessageDecoder())
                            .addLast(new IoTMessageEncoder())
                            .addLast(new IdleStateHandler(120, 60, 0))
                            .addLast(new IoTServerHandler(deviceManager, dataProcessor));
                    }
                });
        
        bootstrap.bind(8080).sync();
        System.out.println("IoT服务器启动成功");
    }
}

// 设备管理器
@Component
public class DeviceManager {
    
    private final Map<String, Device> devices = new ConcurrentHashMap<>();
    
    public boolean authenticate(String deviceId, String token) {
        // 设备认证逻辑
        return true;
    }
    
    public void registerDevice(String deviceId, Channel channel) {
        Device device = new Device(deviceId, channel);
        devices.put(deviceId, device);
        
        channel.closeFuture().addListener(future -> {
            devices.remove(deviceId);
            System.out.println("设备离线: " + deviceId);
        });
    }
    
    public Device getDevice(String deviceId) {
        return devices.get(deviceId);
    }
}

// 数据处理器
@Component
public class DataProcessor {
    
    private final ExecutorService executor = Executors.newFixedThreadPool(10);
    
    public void process(IoTMessage message) {
        executor.submit(() -> {
            try {
                // 数据验证
                validate(message);
                
                // 数据转换
                Map<String, Object> data = transform(message);
                
                // 数据存储
                store(data);
                
                // 实时分析
                analyze(data);
                
            } catch (Exception e) {
                System.err.println("数据处理失败: " + e.getMessage());
            }
        });
    }
    
    private void validate(IoTMessage message) {
        // 数据验证逻辑
    }
    
    private Map<String, Object> transform(IoTMessage message) {
        // 数据转换逻辑
        return message.getPayload();
    }
    
    private void store(Map<String, Object> data) {
        // 存储到数据库或时序数据库
    }
    
    private void analyze(Map<String, Object> data) {
        // 实时分析，触发告警等
    }
}

本章小结

实战项目总结

✅ 第16章：即时通讯系统 - 用户会话、消息路由、离线推送
✅ 第17章：RPC框架 - 动态代理、序列化、服务调用
✅ 第18章：网关服务器 - 路由转发、负载均衡、限流
✅ 第19章：游戏服务器 - 房间管理、状态同步
✅ 第20章：物联网平台 - 设备管理、数据采集、实时处理

关键技术点

协议设计 - 自定义二进制协议
会话管理 - 用户/设备连接管理
消息路由 - 点对点、广播
负载均衡 - 轮询、随机、一致性哈希
数据处理 - 异步处理、批量处理

🎉 第四部分完成！

恭喜！您已经完成了**第四部分：实战项目（16-20章）**的学习！

下一部分预告

**第五部分：性能调优（21-23章）**将学习：

Netty 性能调优实战
JVM 调优与监控
压力测试与性能分析

上一章：第15章：Netty与Spring Boot集成
下一章：第21章：Netty性能调优实战