由于JAVA NIO编写服务器的过程过于复杂且不易掌控,所以我们选择Netty框架进行开发。
/**
* @author monkjavaer
* @date 2019/7/18 14:56
*/
public class NettyServer {
private static Logger LOGGER = LoggerFactory.getLogger(NettyServer.class);
public static int PORT = 8080;
public static void connect(){
//配置两个服务端的NIO线程组,一个用于接收客服端的链接,另一个用于进行SocketChannel的网络读写。
//NioEventLoopGroup是一个处理I/O操作的多线程事件循环
//"boss":接收一个传入连接
EventLoopGroup boss = new NioEventLoopGroup();
//"worker" : 当boss接收连接并把接收的连接注册给worker,work就开始处理
EventLoopGroup worker = new NioEventLoopGroup();
try {
//ServerBootstrap是一个帮助类,可以设置服务器
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(boss,worker)
//NioServerSocketChannel用于实例化新通道来接收传入的连接
.channel(NioServerSocketChannel.class)
//配置日志
.handler(new LoggingHandler(LogLevel.INFO))
//ChannelInitializer用于配置新通道
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//通过ChannelPipeline添加处理类ChannelHandler
//通常有很多处理类,可以将这个内部类new ChannelInitializer提为一个独立的类
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new NettyServerHandler());
}
})
//ChannelOption和ChannelConfig可以设置各种参数
.option(ChannelOption.SO_BACKLOG, 128)
//option()用于接受传入连接的NioServerSocketChannel,childOption()用于父ServerChannel接受的通道
.childOption(ChannelOption.SO_KEEPALIVE, true);
// Bind and start to accept incoming connections.
//异步地绑定服务器;调用 sync()方法阻塞等待直到绑定完成
ChannelFuture f = bootstrap.bind(PORT).sync();
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
}
}
public static void main(String[] args) {
NettyServer.connect();
}
}
客户端的处理类和服务器类似。
/**
* ChannelHandler.Sharable 标注一个channel handler可以被多个channel安全地共享
* ChannelInboundHandlerAdapter实现了ChannelInboundHandler
* 回调事件处理类
*
* @author monkjavaer
* @date 2019/7/18 15:36
*/
@ChannelHandler.Sharable
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
private static Logger LOGGER = LoggerFactory.getLogger(NettyServerHandler.class);
/**
* 新的连接被建立时调用
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LOGGER.info("client {} connected.", ctx.channel().remoteAddress());
ctx.writeAndFlush(Unpooled.copiedBuffer("hello client!", CharsetUtil.UTF_8));
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf byteBuf = (ByteBuf) msg;
//获取缓冲区可读字节数
int readableBytes = byteBuf.readableBytes();
byte[] bytes = new byte[readableBytes];
byteBuf.readBytes(bytes);
LOGGER.info("readableBytes is{},server received message:{}", readableBytes, new String(bytes, StandardCharsets.UTF_8));
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
// ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
// .addListener(ChannelFutureListener.CLOSE);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
LOGGER.error("server exceptionCaught,{}",cause.getMessage());
ctx.close();
}
}
/**
* @author monkjavaer
* @date 2019/7/18 17:17
*/
public class NettyClient {
private static Logger LOGGER = LoggerFactory.getLogger(NettyClient.class);
public static String IP = "127.0.0.1";
public static int PORT = 8080;
public static void main(String[] args) {
EventLoopGroup client = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(client)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY,true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new NettyClientHandler());
}
});
ChannelFuture f = bootstrap.connect(IP,PORT).sync();
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally {
client.shutdownGracefully();
}
}
}
/**
* @author monkjavaer
* @date 2019/7/18 17:26
*/
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
private static Logger LOGGER = LoggerFactory.getLogger(NettyServerHandler.class);
/**
* 新的连接被建立时调用
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LOGGER.info("server {} connected.", ctx.channel().remoteAddress());
ctx.writeAndFlush(Unpooled.copiedBuffer("hello server!", CharsetUtil.UTF_8));
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf byteBuf = (ByteBuf) msg;
//获取缓冲区可读字节数
int readableBytes = byteBuf.readableBytes();
byte[] bytes = new byte[readableBytes];
byteBuf.readBytes(bytes);
LOGGER.info("readableBytes is{},client received message:{}", readableBytes, new String(bytes, StandardCharsets.UTF_8));
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
LOGGER.error("server exceptionCaught,{}",cause.getMessage());
ctx.close();
}
}
原文:https://www.cnblogs.com/monkjavaer/p/11210369.html