Netty入门（二）时间服务器及客户端

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

在这个例子中，我在服务器和客户端连接被创立时发送一个消息，然后在客户端解析收到的消息并输出。并且，在这个项目中我使用 POJO 代替 ByteBuf 来作为传输对象。

一、服务器实现

1. 首先我们自定义传输数据对象

package com.coder.client; import java.util.Date; /**
* 自定义时间数据类
* @author Coder
*
*/
public class Time {
private final long value; public Time() {
// 除以1000是为了使时间精确到秒
this(System.currentTimeMillis() / 1000L);
} public Time(long value) {
this.value = value;
} public long value() {
return value;
} @Override
public String toString() {
return new Date((value()) * 1000L).toString();
}
}

2. 然后我们需要自定义服务器数据编码类

package com.coder.server; import com.coder.client.Time; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder; /**
* 服务器数据编码类
* @author Coder
*
*/
public class TimeEncoderPOJO extends MessageToByteEncoder<Time> { // 发送数据时调用
@Override
protected void encode(ChannelHandlerContext ctx, Time msg, ByteBuf out) throws Exception {
// 只传输当前时间，精确到秒
out.writeInt((int)msg.value());
} }

3. 也需要自定义服务器的业务逻辑类，如下：

package com.coder.server; import com.coder.client.Time; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter; /**
* 服务器解码器
* 连接建立时发送当前时间
* @author Coder
*
*/
public class TimeServerHandlerPOJO extends ChannelInboundHandlerAdapter {
/**
* 连接建立的时候并且准备进行通信时被调用
*/
@Override
public void channelActive(final ChannelHandlerContext ctx) throws Exception {
// 发送当前时间信息
ChannelFuture f = ctx.writeAndFlush(new Time());
// 发送完毕之后关闭 Channel
f.addListener(ChannelFutureListener.CLOSE);
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

4. 有了上面的代码，我们就可以实现服务器程序了，如下：

package com.coder.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.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel; public class TimeServerPOJO {
private int port; public TimeServerPOJO(int port) {
this.port = port;
} public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用来接收进来的连接
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用来处理已经被接收的连接
System.out.println("准备运行端口：" + port); try {
ServerBootstrap b = new ServerBootstrap(); // 启动NIO服务的辅助启动类
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // 这里告诉Channel如何接收新的连接
.childHandler( new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 自定义处理类
// 注意添加顺序
ch.pipeline().addLast(new TimeEncoderPOJO(),new TimeServerHandlerPOJO());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true); // 绑定端口，开始接收进来的连接
ChannelFuture f = b.bind(port).sync(); // 等待服务器socket关闭
f.channel().closeFuture().sync();
} catch (Exception e) {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
} public static void main(String[] args) throws Exception {
int port = 8080;
new TimeServer(port).run();
}
}

执行代码后如下：

这时候服务器在等待客户端的连接（非阻塞）。

二、客户端实现

客户端的实现与服务器类似。

1. 自定义客户端数据解码类

package com.coder.client; import java.util.List; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder; public class TimeDecoderPOJO extends ByteToMessageDecoder {
/**
* 有新数据接收时调用
* 为防止分包现象，先将数据存入内部缓存，到达满足条件之后再进行解码
*/
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if(in.readableBytes() < 4) {
return;
} // out添加对象则表示解码成功
out.add(new Time(in.readUnsignedInt()));
}
}

2. 自定义客户端业务逻辑类

package com.coder.client; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter; /**
* 客户端数据处理类
* @author Coder
*
*/
public class TimeClientHandlerPOJO extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 直接将信息转换成Time类型输出即可
Time time = (Time)msg;
System.out.println(time);
ctx.close();
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

3. 客户端程序实现

Netty 客户端的通信步骤大致为：

创建一个 NIO 线程组，用于处理服务器与客户端的连接，客户端不需要用到 boss worker。
创建一个 Bootstrap 对象，配置 Netty 的一系列参数，由于客户端 SocketChannel 没有父亲，所以不需要使用 childoption。
创建一个用于实际处理数据的类ChannelInitializer，进行初始化的准备工作，比如设置接受传出数据的字符集、格式以及实际处理数据的接口。
配置服务器 IP 和端口号，建立与服务器的连接。

package com.coder.client; import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel; public class TimeClientPOJO {
public static void main(String[] args) throws Exception{
String host = "127.0.0.1"; // ip
int port = 8080; // 端口
EventLoopGroup workerGroup = new NioEventLoopGroup(); try {
Bootstrap b = new Bootstrap(); // 与ServerBootstrap类似
b.group(workerGroup); // 客户端不需要boss worker
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true); // 客户端的socketChannel没有父亲
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// POJO
ch.pipeline().addLast(new TimeDecoderPOJO() ,new TimeClientHandlerPOJO());
}
}); // 启动客户端，客户端用connect连接
ChannelFuture f = b.connect(host, port).sync(); // 等待连接关闭
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
}