1
public interface Serialize {
2
    /**
3
     * 序列化
4
     *
5
     * @param obj 要序列化的对象
6
     * @return 字节数组
7
     */
8
    byte[] serialize(Object obj);
9

10
    /**
11
     * 反序列化
12
     *
13
     * @param bytes 序列化后的字节数组
14
     * @param clazz 目标类
15
     * @param <T>   类的类型。举个例子,  {@code String.class} 的类型是 {@code Class<String>}.
16
     *              如果不知道类的类型的话，使用 {@code Class<?>}
17
     * @return 反序列化的对象
18
     */
19
    <T> T deserialize(byte[] bytes, Class<T> clazz);
20
}

1
<dependency>
2
  <groupId>com.esotericsoftware</groupId>
3
  <artifactId>kryo</artifactId>
4
  <version>4.0.2</version>
5
</dependency>

1
@Slf4j
2
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
3
    @Override
4
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
5
        super.channelRead(ctx, msg);
6
        try {
7
            RpcRequest rpcRequest = (RpcRequest) msg;
8
            log.info("服务端接收到信息: [{}] ", rpcRequest);
9
            RpcResponse messageFromServer = RpcResponse.builder().message("message from server").build();
10
            ChannelFuture f = ctx.writeAndFlush(messageFromServer);
11
            f.addListener(ChannelFutureListener.CLOSE);
12
        } finally {
13
            ReferenceCountUtil.release(msg);
14
        }
15
    }
16
}

1
@Slf4j
2
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
3

4
    @Override
5
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
6
        super.channelRead(ctx, msg);
7
        try {
8
            RpcResponse rpcResponse = (RpcResponse) msg;
9
            log.info("服务端返回数据: [{}]", rpcResponse.toString());
10
            AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
11
            ctx.channel().attr(key).set(rpcResponse);
12
            ctx.channel().close();
13
        } finally {
14
            ReferenceCountUtil.release(msg);
15
        }
16
    }
17
}

1
@Slf4j
2
@RequiredArgsConstructor
3
public class NettyServer {
4

5
    private final int port;
6

7
    public void run(){
8
        EventLoopGroup bossGroup = new NioEventLoopGroup();
9
        EventLoopGroup workerGroup = new NioEventLoopGroup();
10
        KryoSerializer kryoSerializer = new KryoSerializer();
11
        try {
12
            ServerBootstrap b = new ServerBootstrap();
13
            b.group(bossGroup, workerGroup)
14
                    .channel(NioServerSocketChannel.class)
15
                    .childOption(ChannelOption.TCP_NODELAY, true)
16
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
17
                    .option(ChannelOption.SO_BACKLOG, 128)
18
                    .handler(new LoggingHandler(LogLevel.INFO))
19
                    .childHandler(new ChannelInitializer<SocketChannel>() {
20
                        @Override
21
                        protected void initChannel(SocketChannel ch) {
22
                            ch.pipeline().addLast(new NettyDecoder(kryoSerializer, RpcRequest.class));
23
                            ch.pipeline().addLast(new NettyEncoder(kryoSerializer, RpcResponse.class));
24
                            ch.pipeline().addLast(new NettyServerHandler());
25
                        }
26
                    });
27

28
            ChannelFuture f = b.bind(port).sync();
29
            f.channel().closeFuture().sync();
30
        } catch (InterruptedException e) {
31
            log.error("无法启动服务:", e);
32
        } finally {
33
            bossGroup.shutdownGracefully();
34
            workerGroup.shutdownGracefully();
35
        }
36
    }
37
}

1
@Slf4j
2
@RequiredArgsConstructor
3
public class NettyClient {
4
    private final String host;
5
    private final int port;
6
    private static final Bootstrap b;
7

8
    static {
9
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
10
        b = new Bootstrap();
11
        KryoSerializer kryoSerializer = new KryoSerializer();
12
        b.group(eventLoopGroup)
13
                .channel(NioSocketChannel.class)
14
                .handler(new LoggingHandler(LogLevel.INFO))
15
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
16
                .handler(new ChannelInitializer<SocketChannel>() {
17
                    @Override
18
                    protected void initChannel(SocketChannel ch) {
19
                        ch.pipeline().addLast(new NettyDecoder(kryoSerializer, RpcResponse.class));
20
                        ch.pipeline().addLast(new NettyEncoder(kryoSerializer, RpcRequest.class));
21
                        ch.pipeline().addLast(new NettyClientHandler());
22
                    }
23
                });
24
    }
25

26
    public RpcResponse send(RpcRequest request) {
27
        try {
28
            ChannelFuture f = b.connect(host, port).sync();
29
            log.info("客户端链接  {}", host + ":" + port);
30
            Channel futureChannel = f.channel();
31
            if (futureChannel != null) {
32
                futureChannel.writeAndFlush(request).addListener(future -> {
33
                    if (future.isSuccess()) {
34
                        log.info("客户端发送信息: [{}]", request.toString());
35
                    } else {
36
                        log.error("发送失败:", future.cause());
37
                    }
38
                });
39
                futureChannel.closeFuture().sync();
40
                AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
41
                return futureChannel.attr(key).get();
42
            }
43
        } catch (InterruptedException e) {
44
            log.error("无法链接服务端:", e);
45
        }
46
        return null;
47
    }
48
}