服务端主要有以下几步:
- 新建服务netty boot
- 实现出站数据分包规则
- 实现数据入站组包规则
- 实现自定义对象转换为bytes编码器
- 实现bytes转换为自定义对象解码器
- 对接收到的对象处理.
- 第一步很简单,直接new就可以.
ServerBootstrap nettyBoot =new ServerBootstrap();
- 第二步,第三步,netty也实现了,只需传参自定义即可.
LengthFieldPrepender prepender = new LengthFieldPrepender(4, true);
LengthFieldBasedFrameDecoder frameDecoder = new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, -4, 4);
- 第四步,新建对象编码字符串类.这里写一个简单点的StringToBytes编码类.
/**
* StringToBytesEncoder
* <p>
* 释义: 字符串转换为bytes编码器
*
* @author: xinyi.pan
* @create: 2018-11-28 09:42
**/
@ChannelHandler.Sharable
public class StringToBytesEncoder extends MessageToByteEncoder<String> {
@Override
protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
//编码
out.writeCharSequence(msg, StandardCharsets.UTF_8);
}
}
- 第五步,新建一个bytes解码类
/**
* BytesToStringDecoder
* <p>
* 释义: bytes -> string 解码器
*
* @author: xinyi.pan
* @create: 2018-11-28 09:44
**/
@ChannelHandler.Sharable
public class BytesToStringDecoder extends MessageToMessageDecoder<ByteBuf> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
//解码
String str = msg.toString(StandardCharsets.UTF_8);
out.add(str);
}
}
- 第六步,接收到的消息处理类,这个类负责处理,并且消息不再转发到下一级.
/**
* MsgHandler
* <p>
* 释义:
* 数据转发,得到的string 可以经由该处理器转发出来
*
* @author: xinyi.pan
* @create: 2018-11-28 09:50
**/
@ChannelHandler.Sharable
public class MsgClientHandler extends ChannelInboundHandlerAdapter {
String name ;
public MsgClientHandler(String name) {
this.name = name;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String str = (String) msg;
System.out.println("客户端收到:" + str);
/**
* do any thing ....
*/
//释放msg
ReferenceCountUtil.release(msg);
}
}
/**
* MsgHandler
* <p>
* 释义:
* 数据转发,得到的string 可以经由该处理器转发出来
*
* @author: xinyi.pan
* @create: 2018-11-28 09:50
**/
@ChannelHandler.Sharable
public class MsgServerHandler extends ChannelInboundHandlerAdapter {
String name ;
public MsgServerHandler(String name) {
this.name = name;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String str = (String) msg;
System.out.println("服务端收到:" + str);
ctx.writeAndFlush("服务端返回数据:" + RandomStringUtils.randomAlphabetic(10));
//释放msg
ReferenceCountUtil.release(msg);
}
}
- 剩下的就是启动netty.服务端启动
//用于接收网络I/O请求线程池,异步IO
private static EventLoopGroup bossEventLoop = new NioEventLoopGroup();
//处理channel的线程池
private static EventLoopGroup workerEventLoop = new NioEventLoopGroup();
public static void nettySart() {
//
ServerBootstrap nettyBoot = new ServerBootstrap();
//出站 最后一步,数据分包
final LengthFieldPrepender prepender = new LengthFieldPrepender(4, true);
//出站,第一步,将自定义对象:String转化为bytes
final StringToBytesEncoder encoder = new StringToBytesEncoder();
//入站第一步:网络包组包,不过这里加入last的时候要new出来,不然回报异常.
LengthFieldBasedFrameDecoder frameDecoder = new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, -4, 4);
//入站,第二步,bytes 转化为自定义的对象:String
final BytesToStringDecoder decoder = new BytesToStringDecoder();
//入战,第三步,消息处理,亦可以转发.
final MsgServerHandler msgServerHandler = new MsgServerHandler("服务端");
//配置netty服务
nettyBoot.group(bossEventLoop, workerEventLoop)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 100)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) {
ChannelPipeline pipeline = ch.pipeline();
//入站,1,2,3出站5,4
pipeline.addLast(new LoggingHandler(LogLevel.INFO)) //日志打印
.addLast("4", prepender)//4 分包,bytes -> http数据包
.addLast("5", encoder)//5 //msg -> bytes
.addLast("1", new LengthFieldBasedFrameDecoder(1000000, 0, 4, -4, 4))//1,http网络数据包,组包
.addLast("2", decoder)//2 //bytes -> msg ,自定义解码
.addLast("3", msgServerHandler);//3 //msg -> msg,数据转换,转发数据
}
});
try {
//启动netty并监听指定端口,这里是main执行用同步阻塞.也可以用异步.
ChannelFuture channelFuture = nettyBoot.bind(1765).await();
} catch (Throwable ex) {
throw new RuntimeException(ex);
}
}
@PreDestroy
public void close() {
bossEventLoop.shutdownGracefully();
workerEventLoop.shutdownGracefully();
}
7.客户端发送数据
//用于连接和处理网络请求I/O的线程池
private static NioEventLoopGroup eventLoopGroup;
public static void startConnect() {
//关闭检查
if (eventLoopGroup != null) {
try {
eventLoopGroup.shutdownGracefully().sync();
eventLoopGroup = null;
} catch (InterruptedException e) {
}
}
//新建
eventLoopGroup = new NioEventLoopGroup();
//出站 最后一步,数据分包
final LengthFieldPrepender prepender = new LengthFieldPrepender(4, true);
//出站,第一步,将自定义对象:String转化为bytes
final StringToBytesEncoder encoder = new StringToBytesEncoder();
//入站,第二步,bytes 转化为自定义的对象:String
final BytesToStringDecoder decoder = new BytesToStringDecoder();
//入战,第三步,消息处理,亦可以转发.
final MsgClientHandler msgClientHandler = new MsgClientHandler("客户端");
//服务配置
Bootstrap clientBoot = new Bootstrap();
clientBoot.group(eventLoopGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 15 * 1000) // 超时时间
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//请求日志
pipeline.addLast(new LoggingHandler(LogLevel.INFO)) //日志打印
.addLast("4", prepender)//out 4
.addLast("5", encoder)//out 5
.addLast("1", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, -4, 4))//in 1
.addLast("2", decoder)//in 2
.addLast("3", msgClientHandler)//in 3
;
}
});
try {
//链接
ChannelFuture channelFuture = clientBoot.connect("127.0.0.1", 1765).sync();
Channel channel = channelFuture.channel();
channel.writeAndFlush("啊还是大所大撒所大");
// 服务器返回结果在 MsgHandler 里面处理
} catch (Throwable ex) {
}
}
@PreDestroy
public void close() {
eventLoopGroup.shutdownGracefully();
}
- 测试,先启动服务端
public static void main(String[] args) {
nettySart();
}
- 然后启动客户端
public static void main(String[] args) {
startConnect();
try {
Thread.sleep(100000000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
}
System.out.println("OK");
}
10, 查看日志
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 24 e6 9c 8d e5 8a a1 e7 ab af e8 bf 94 |...$............|
|00000010| e5 9b 9e e6 95 b0 e6 8d ae 3a 42 69 52 41 4f 67 |.........:BiRAOg|
|00000020| 63 45 70 7a |cEpz |
+--------+-------------------------------------------------+----------------+
客户端收到:服务端返回数据:BiRAOgcEpz
L:/127.0.0.1:1765 - R:/127.0.0.1:52226] READ: 31B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 1f e5 95 8a e8 bf 98 e6 98 af e5 a4 a7 |................|
|00000010| e6 89 80 e5 a4 a7 e6 92 92 e6 89 80 e5 a4 a7 |............... |
+--------+-------------------------------------------------+----------------+
服务端收到:啊还是大所大撒所大
11:34:19.663 [nioEventLoopGroup-3-1] INFO io.netty.handler.logging.LoggingHandler - [id: 0x29ed7f41, L:/127.0.0.1:1765 - R:/127.0.0.1:52226] WRITE: 4B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 24 |...$ |
+--------+-------------------------------------------------+----------------+
另外:附一个自己的基github地址:于netty开发的rpc
