1. 编写服务端代码
编写业务逻辑：读取到客户端的消息时候，打印客户端消息，并给客户端回复一条消息
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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 DemoNettyServer {
public void bind(int port) throws Exception {
// 主线程组
EventLoopGroup bossGroup = new NioEventLoopGroup();
// 从线程组
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
// netty服务器启动类
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup) //绑定两个线程组
// 用于构造socketchannel工厂
.channel(NioServerSocketChannel.class) //指定NIO的模式
.childHandler(new ChannelInitializer<SocketChannel>() { // 子处理器，用于处理workerGroup
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new DemoNettyServerHandler());
}
});
// 启动server，绑定端口
ChannelFuture channelFuture = serverBootstrap.bind(8088).sync();
System.out.println("server start");
// 监听关闭的channel，等待服务器 socket 关闭 。设置位同步方式
channelFuture.channel().closeFuture().sync();
System.out.println("server close");
} finally {
//退出线程组
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
int port = 8080;
new DemoNettyServer().bind(port);
}
}
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
public class DemoNettyServerHandler extends ChannelInboundHandlerAdapter {
/**
* 本方法用于读取客户端发送的信息
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("收到来自客服端的一条消息");
ByteBuf result = (ByteBuf) msg;
byte[] bytesMsg = new byte[result.readableBytes()];
// msg中存储的是ByteBuf类型的数据，把数据读取到byte[]中
result.readBytes(bytesMsg);
String resultStr = new String(bytesMsg);
// 接收并打印客户端的信息
System.out.println("客服端内容:" + resultStr);
// 释放资源，这行很关键
result.release();
// 向客户端发送消息
String response = "我是server，我已经收到你的消息： " + resultStr;
// 在当前场景下，发送的数据必须转换成ByteBuf数组
ByteBuf encoded = ctx.alloc().buffer(4 * response.length());
encoded.writeBytes(response.getBytes());
ctx.write(encoded);
ctx.flush();
}
/**
* 本方法用作处理异常
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// 当出现异常就关闭连接
cause.printStackTrace();
ctx.close();
}
/**
* 信息获取完毕后操作
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
2. 编写客户端代码
编写业务逻辑：获取用户输入，连接服务端，发送消息，读取服务端消息，关闭连接。
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;
import java.util.Scanner;
public class DemoNettyClient {
public void connect(String host, int port) throws Exception {
EventLoopGroup worker = new NioEventLoopGroup();
try {
// 客户端启动类程序
Bootstrap bootstrap = new Bootstrap();
/**
*EventLoop的组
*/
bootstrap.group(worker);
/**
* 用于构造socketchannel工厂
*/
bootstrap.channel(NioSocketChannel.class);
/**设置选项
* 参数：Socket的标准参数（key，value），可自行百度
保持呼吸，不要断气！
* */
bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
/**
* 自定义客户端Handle（客户端在这里搞事情）
*/
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addFirst(new DemoNettyClientHandler());
}
});
/** 开启客户端监听，连接到远程节点，阻塞等待直到连接完成*/
ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
/**阻塞等待数据，直到channel关闭(客户端关闭)*/
channelFuture.channel().closeFuture().sync();
} finally {
worker.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
while (true){
Scanner content = new Scanner(System.in);
System.out.print("请输入您要发送的内容: ");
Meesage.CLIENT_MESSAGE = content.nextLine();
DemoNettyClient client = new DemoNettyClient();
client.connect("127.0.0.1", 8088);
}
}
}
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
public class DemoNettyClientHandler extends ChannelInboundHandlerAdapter {
/**
* 服务端发过来消息时调用
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("收到来自服务端的一条消息");
ByteBuf result = (ByteBuf) msg;
byte[] result1 = new byte[result.readableBytes()];
result.readBytes(result1);
System.out.println(new String(result1));
result.release();
//关闭连接
ctx.close();
}
/**
* 异常时调用
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
/**
* 连接到服务器调用
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
String msg = Meesage.CLIENT_MESSAGE;
ByteBuf encoded = ctx.alloc().buffer(4 * msg.length());
encoded.writeBytes(msg.getBytes());
ctx.write(encoded);
ctx.flush();
}
}
public class Meesage {
public static String CLIENT_MESSAGE = "";
public static String SERVER_MESSAGE = "";
}
3. 结果分析
