一、前言
Netty是一個基于Java的高性能、事件驅動的網絡應用框架,廣泛應用于各種網絡通信場景。本文將介紹如何使用Netty構建一個簡單的網絡聊天室,并擴展序列化算法來提高數據傳輸效率和靈活性。
二、Netty網絡聊天室的實現
1. 項目結構
我們將使用Maven構建項目,項目結構如下:
netty-chatroom/
├── src/
│ ├── main/
│ │ ├── java/
│ │ │ ├── server/
│ │ │ │ ├── ChatServer.java
│ │ │ │ ├── ChatServerInitializer.java
│ │ │ │ ├── ChatServerHandler.java
│ │ │ ├── client/
│ │ │ │ ├── ChatClient.java
│ │ │ │ ├── ChatClientInitializer.java
│ │ │ │ ├── ChatClientHandler.java
│ │ ├── resources/
│ │ ├── log4j.properties
├── pom.xml
?
2. 服務器端實現
ChatServer.java
package server;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;public class ChatServer {private final int port;public ChatServer(int port) {this.port = port;}public void start() throws Exception {NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChatServerInitializer()).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);ChannelFuture f = b.bind(port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {new ChatServer(8080).start();}
}
?
ChatServerInitializer.java
package server;import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;public class ChatServerInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(new StringEncoder());ch.pipeline().addLast(new ChatServerHandler());}
}
?
ChatServerHandler.java
package server;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;public class ChatServerHandler extends SimpleChannelInboundHandler<String> {private static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {channels.add(ctx.channel());}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {channels.remove(ctx.channel());}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {for (var channel : channels) {if (channel != ctx.channel()) {channel.writeAndFlush("[Client] " + ctx.channel().remoteAddress() + " says: " + msg + "\n");} else {channel.writeAndFlush("[You] " + msg + "\n");}}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}
?
3. 客戶端實現
ChatClient.java
package client;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;public class ChatClient {private final String host;private final int port;public ChatClient(String host, int port) {this.host = host;this.port = port;}public void start() throws Exception {NioEventLoopGroup group = new NioEventLoopGroup();try {Bootstrap b = new Bootstrap();b.group(group).channel(NioSocketChannel.class).handler(new ChatClientInitializer());ChannelFuture f = b.connect(host, port).sync();f.channel().closeFuture().sync();} finally {group.shutdownGracefully();}}public static void main(String[] args) throws Exception {new ChatClient("localhost", 8080).start();}
}
?
ChatClientInitializer.java
package client;import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;public class ChatClientInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(new StringEncoder());ch.pipeline().addLast(new ChatClientHandler());}
}
?
ChatClientHandler.java
package client;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;public class ChatClientHandler extends SimpleChannelInboundHandler<String> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {System.out.println(msg);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}
?
三、擴展序列化算法
為了提高數據傳輸效率,我們可以擴展Netty的序列化算法。Netty默認提供的序列化算法包括Java序列化、JSON、Protobuf等。下面介紹如何使用Protobuf進行序列化。
1. 配置Protobuf
首先,在?pom.xml
中添加Protobuf依賴:
<dependency><groupId>com.google.protobuf</groupId><artifactId>protobuf-java</artifactId><version>3.19.1</version>
</dependency>
?
2. 定義Protobuf消息
創建一個?chat.proto
文件:
syntax = "proto3";package chat;message ChatMessage {string from = 1;string to = 2;string content = 3;
}
?
編譯Protobuf文件生成Java類:
protoc --java_out=src/main/java src/main/proto/chat.proto
?
3. 修改服務器端處理器
在服務器端,使用Protobuf進行消息的序列化和反序列化:
package server;import chat.ChatMessage;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;public class ChatServerHandler extends SimpleChannelInboundHandler<ChatMessage> {private static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {channels.add(ctx.channel());}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {channels.remove(ctx.channel());}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, ChatMessage msg) throws Exception {for (var channel : channels) {if (channel != ctx.channel()) {channel.writeAndFlush(msg.toBuilder().setContent("[Client] " + ctx.channel().remoteAddress() + " says: " + msg.getContent()).build());} else {channel.writeAndFlush(msg.toBuilder().setContent("[You] " + msg.getContent()).build());}}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}
?
4. 修改客戶端處理器
在客戶端,同樣使用Protobuf進行消息的序列化和反序列化:
package client;import chat.ChatMessage;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;public class ChatClientHandler extends SimpleChannelInboundHandler<ChatMessage> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, ChatMessage msg) throws Exception {System.out.println(msg.getContent());}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}