- 案例代码下载
- 初学者的话推荐直接套用 all-in-one 的 jar 包，若熟悉 Netty 可以根据需求添加不同的 jar 包
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.22.Final</version>
</dependency>
服务端
- 首先编写 Handler 类，继承自 SimpleChannelInboundHandler，SimpleChannelInboundHandler 能处理特定类型的消息，覆盖其几个方法：
- handlerAdded()：每当从服务端收到新的客户端连接时，将客户端的 Channel 存入 ChannelGroup 列表中，并通知列表中的其他客户端 Channel
- handlerRemoved()：每当从服务端收到客户端断开时，客户端的 Channel 自动从 ChannelGroup 列表中移除了，并通知列表中的其他客户端 Channel
- channelRead0()：每当从服务端读到客户端写入信息时，将信息转发给其他客户端的 Channel
- channelActive()：服务端监听到客户端活动
- channelInactive()：服务端监听到客户端不活动
- exceptionCaught()：出现异常时回调该方法
public class SimpleChatServerHandler extends SimpleChannelInboundHandler<String> {
public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception { // (2)
Channel incoming = ctx.channel();
channels.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 加入\n");
channels.add(ctx.channel());
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { // (3)
Channel incoming = ctx.channel();
channels.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 离开\n");
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception { // (4)
Channel incoming = ctx.channel();
for (Channel channel : channels) {
if (channel != incoming){
channel.writeAndFlush("[" + incoming.remoteAddress() + "]" + s + "\n");
} else {
channel.writeAndFlush("[you]" + s + "\n");
}
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception { // (5)
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient:"+incoming.remoteAddress()+"在线");
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception { // (6)
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient:"+incoming.remoteAddress()+"掉线");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (7)
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient:"+incoming.remoteAddress()+"异常");
// 当出现异常就关闭连接
cause.printStackTrace();
ctx.close();
}
}
- 编写 ChannelInitializer 用于配置客户端连接的 Channel，这里的几个 Handler 作用如下：
- DelimiterBasedFrameDecoder：允许你指定的一个或多个分隔符来分割接收的 ByteBuf，它的构造方法参数意义如下：
- maxFrameLength：解码的帧的最大长度
- stripDelimiter：解码时是否去掉分隔符
- failFast：为 true 时当 frame 长度超过 maxFrameLength 立即报 TooLongFrameException 异常，为 false 读取完整个帧再报异常
- delimiter：分隔符
- StringDecoder：将入站的 ByteBuf 解码为字符串
- StringEncoder：将出站的 ByteBuf 编码为字符串
public class SimpleChatServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()))
.addLast("decoder", new StringDecoder())
.addLast("encoder", new StringEncoder())
.addLast("handler", new SimpleChatServerHandler());
System.out.println("SimpleChatClient:"+ch.remoteAddress() +"连接上");
}
}
public class SimpleChatServer {
private int port;
public SimpleChatServer(int port) {
this.port = port;
}
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // 处理客户端连接事件的线程池
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 处理连接后所有事件的线程池
try {
ServerBootstrap b = new ServerBootstrap(); // NIO 服务的辅助启动类
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // 指定连接该服务器的 Channel 类型为 NioServerSocketChannel
.childHandler(new SimpleChatServerInitializer()) // 指定需要执行的 Handler
.option(ChannelOption.SO_BACKLOG, 128) // 设置 bossGroup 的相关参数
.childOption(ChannelOption.SO_KEEPALIVE, true); // 设置 workerGroup 相关参数
System.out.println("SimpleChatServer 启动了");
ChannelFuture f = b.bind(port).sync(); // 绑定端口，调用 ChannelFuture 的 sync() 阻塞方法等待绑定完成
// 调用 closeFuture() 方法返回此通道关闭时的 ChannelFuture
// 调用 ChannelFuture 的 sync() 阻塞方法直到服务端关闭链路之后才退出 main() 函数
f.channel().closeFuture().sync();
} finally {
// 优雅退出机制。。。退出线程池（该方法源码没读过，也不知怎么个优雅方式）
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
System.out.println("SimpleChatServer 关闭了");
}
}
public static void main(String[] args) throws Exception {
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 8080;
}
new SimpleChatServer(port).run();
}
}
客户端
public class SimpleChatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(msg);
}
}
- 编写 ChannelInitializer 用于配置客户端连接的 Channel
public class SimpleChatClientInitializer extends ChannelInitializer {
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("framer",new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()))
.addLast("decoder", new StringDecoder())
.addLast("encoder", new StringEncoder())
.addLast("handler",new SimpleChatClientHandler());
}
}
public class SimpleChatClient {
private final String host;
private final int port;
public SimpleChatClient(String host, int port) {
this.host = host;
this.port = port;
}
public void run(){
EventLoopGroup workerGroup = new NioEventLoopGroup();
try{
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(workerGroup)
.channel(NioSocketChannel.class)
.handler(new SimpleChatClientInitializer());
Channel channel = bootstrap.connect(host,port).sync().channel(); // 连接服务端
// 接收控制台输入的内容并发送给服务端
BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
while (true){
channel.writeAndFlush(br.readLine()+"\r\n");
}
}catch (Exception e){
e.printStackTrace();
}finally {
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
new SimpleChatClient("localhost",8080).run();
}
}
网友评论