Spring Boot+Netty实现远程过程调用(RPC)

# 目录结构
sp_netty_server
 ...
 - handler
  - NettyServerHandler.java
 - rpc
  - NettyServer.java
  - ServerChannelInitializer.java
 SpNettyServerApplication.java

依赖和配置文件:

    org.springframework.boot
    spring-boot-starter-web



    org.projectlombok
    lombok
    1.18.2
    true



    io.netty
    netty-all
    4.1.17.Final

server.port=8801

具体代码：

NettyServerHandler.java

@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("Channel active ...");
    }

    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("服务器收到消息: {}", msg.toString());
        ctx.write(msg.toString()+",Hello too !");
        ctx.flush();
    }

    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

NettyServer.java

@Slf4j
public class NettyServer {
    public void start() {
        InetSocketAddress socketAddress = new InetSocketAddress("127.0.0.1", 8082);
        //new 一个主线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        //new 一个工作线程组
        EventLoopGroup workGroup = new NioEventLoopGroup(200);
        ServerBootstrap bootstrap = new ServerBootstrap()
                .group(bossGroup, workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ServerChannelInitializer())
                .localAddress(socketAddress)
                //设置队列大小
                .option(ChannelOption.SO_BACKLOG, 1024)
                // 两小时内没有数据的通信时,TCP会自动发送一个活动探测数据报文
                .childOption(ChannelOption.SO_KEEPALIVE, true);
        //绑定端口,开始接收进来的连接
        try {
            ChannelFuture future = bootstrap.bind(socketAddress).sync();
            log.info("服务器启动开始监听端口: {}", socketAddress.getPort());
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("服务器开启失败", e);
        } finally {
            //关闭主线程组
            bossGroup.shutdownGracefully();
            //关闭工作线程组
            workGroup.shutdownGracefully();
        }
    }
}

ServerChannelInitializer.java

public class ServerChannelInitializer extends ChannelInitializer {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        //添加编解码
        socketChannel.pipeline().addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
        socketChannel.pipeline().addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
        socketChannel.pipeline().addLast(new NettyServerHandler());
    }
}

SpNettyServerApplication.java

@Slf4j
@SpringBootApplication
public class SpNettyServerApplication extends SpringBootServletInitializer {

    @Override
    protected SpringApplicationBuilder configure(SpringApplicationBuilder application) {
        return application.sources(SpNettyServerApplication.class);
    }

    public static void main(String[] args) {
        SpringApplication.run(SpNettyServerApplication.class, args);
        //开启Netty服务
        NettyServer nettyServer = new NettyServer();
        nettyServer.start();
        log.info("server is running ...");
    }

}

# 目录结构
sp_netty_client
 ...
 - controller
  - HelloController.java
 - data
  - ResponseResult.java
 - handler
  - NettyClientHandler.java
 - rpc
  - NettyClientUtil.java
 SpNettyClientApplication.java

具体代码：

SpNettyClientApplication.java

@SpringBootApplication
public class SpNettyClientApplication {
    public static void main(String[] args) {
        SpringApplication.run(SpNettyClientApplication.class, args);
    }
}

NettyClientUtil.java

@Slf4j
public class NettyClientUtil {

    public static ResponseResult helloNetty(String msg) {
        NettyClientHandler nettyClientHandler = new NettyClientHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap()
                .group(group)
                //该参数的作用就是禁止使用Nagle算法，使用于小数据即时传输
                .option(ChannelOption.TCP_NODELAY, true)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast("decoder", new StringDecoder());
                        socketChannel.pipeline().addLast("encoder", new StringEncoder());
                        socketChannel.pipeline().addLast(nettyClientHandler);
                    }
                });
        try {
            ChannelFuture future = bootstrap.connect("127.0.0.1", 8082).sync();
            log.info("客户端发送成功....");
            //发送消息
            future.channel().writeAndFlush(msg);
            // 等待连接被关闭
            future.channel().closeFuture().sync();
            return nettyClientHandler.getResponseResult();
        } catch (Exception e) {
            log.error("客户端Netty失败", e);
        } finally {
            //以一种优雅的方式进行线程退出
            group.shutdownGracefully();
        }
        return new ResponseResult(500, null, "server error");
    }
}

NettyClientHandler.java

@Slf4j
@Setter
@Getter
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    private ResponseResult responseResult;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("client is running ...");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("客户端收到消息: {}", msg.toString());
        this.responseResult = new ResponseResult(200, null, msg.toString());
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

ResponseResult.java

@Setter
@Getter
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class ResponseResult implements Serializable {

    private int code;

    private Object data;

    private String msg;

}

HelloController.java

@RestController
public class HelloController {

    @RequestMapping("/hello/{name}")
    public ResponseResult helloNetty(@PathVariable("name") String name) {
        return NettyClientUtil.helloNetty(name);
    }
}

参考：

https://xiyuan.blog.csdn.net/article/details/109248198