网络编程之基于nio的Netty框架Demo

首选引入netty jar包

/**

 * @author dlj2018年1月2日

 * netty服务端，实现异步非阻塞处理消息

 */

public class Server {

public static void main(String[] args) throws Exception {
//1 第一个线程组 是用于接收Client端连接的
EventLoopGroup bossGroup = new NioEventLoopGroup();
//2 第二个线程组 是用于实际的业务处理操作的
EventLoopGroup workerGroup = new NioEventLoopGroup();

//3 创建一个辅助类Bootstrap，就是对我们的Server进行一系列的配置
ServerBootstrap b = new ServerBootstrap(); 
//把俩个工作线程组加入进来
b.group(bossGroup, workerGroup)
//我要指定使用NioServerSocketChannel这种类型的通道(服务端通道类型)
.channel(NioServerSocketChannel.class)
<
4000
span style="white-space:pre;">//一定要使用 childHandler 去绑定具体的 事件处理器
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(new ServerHandler());
}
});

//绑定指定的端口 进行监听
ChannelFuture f = b.bind(8765).sync(); 

//Thread.sleep(1000000);
f.channel().closeFuture().sync();//主线程阻塞

bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
 

}

}

----------------------------------------------------------------------------------------------------------

//非阻塞服务端业务处理器

public class ServerHandler  extends ChannelHandlerAdapter {

/* 
* 当客户端简历连接后做writeAndFlush后 自动读取客户端数据
* msg客户端发来的消息
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

//do something msg
ByteBuf buf = (ByteBuf)msg; //ByteBuf：netty封装的缓冲区
byte[] data = new byte[buf.readableBytes()];  //readableBytes:缓冲区大小
buf.readBytes(data); //将缓冲区数据转到data中
String request = new String(data, "utf-8");
System.out.println("Server: " + request);
//写给客户端
String response = "我是反馈的信息";
//返回数据方式1
ctx.writeAndFlush(Unpooled.copiedBuffer( response.getBytes()));//Unpooled.copiedBuffer将bytep[]转成buf
//返回数据方式2
// ctx.write(Unpooled.copiedBuffer( response.getBytes()));
// ctx.write(Unpooled.copiedBuffer( "hahhaha".getBytes()));
// ctx.flush();

//writeAndFlush.addListener(ChannelFutureListener.CLOSE);//ChannelFuture添加一个监听器，向客户端写完数据后关闭连接(故netty可以实现短连接与长连接)

}

//发生异常后自动调用
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}

}

------------------------------------------------------------------------------------------------

/**

 * @author dlj2018年1月2日

 * netty 客户的实现异步非阻塞接受返回消息

 */

public class Client {

public static void main(String[] args) throws Exception {

EventLoopGroup workgroup = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();
b.group(workgroup)
.channel(NioSocketChannel.class)//NioSocketChannel客户端通道类型
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(new ClientHandler());
}
});

ChannelFuture cf1 = b.connect("127.0.0.1", 8765).sync();

//发送数据方式1 用netty的bytebuf来存放数据
cf1.channel().writeAndFlush(Unpooled.copiedBuffer("777".getBytes()));
//发送数据方式2
// ctx.write(Unpooled.copiedBuffer( response.getBytes()));
// ctx.write(Unpooled.copiedBuffer( "hahhaha".getBytes()));
// ctx.flush();

cf1.channel().closeFuture().sync();
workgroup.shutdownGracefully();

}

}

----------------------------------------------------------------

//非阻塞客户端业务处理器

public class ClientHandler extends ChannelHandlerAdapter {

/* 读取服务端数据
* msg服务端端发来的消息
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try {
//do something msg
ByteBuf buf = (ByteBuf)msg; //ByteBuf：netty封装的缓冲区
byte[] data = new byte[buf.readableBytes()];  //readableBytes:缓冲区大小
buf.readBytes(data); //将缓冲区数据转到data中
String request = new String(data, "utf-8");
System.out.println("Client: " + request);

} finally {
ReferenceCountUtil.release(msg); //客户端要显示释放消息，服务端由于做了write操作里面netty内部帮我们释放了消息
}
}

//发生异常时回调方法
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}

}