摘要：
稍后，我们将考虑通过netty制作一个非常简单的RPC框架。今天，我们将尝试通过正常的调用逻辑调用netty构建的nio服务器，并同步获取返回信息。
后面考虑通过netty做一个真正意义的简约版RPC框架，今天先尝试通过正常调用逻辑调用netty构建的nio服务端并同步获得返回信息。为后面做铺垫
服务端实现我们先完成服务端的逻辑，逻辑很简单，把客户端请求的内容加上服务器时间戳一并返回
public void run() throws InterruptedException { EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workGroup = new NioEventLoopGroup(); try{ ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup,workGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG,4096) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new LineBasedFrameDecoder(Integer.MAX_VALUE)); ch.pipeline().addLast(new StringDecoder()); ch.pipeline().addLast(new Handler()); } }); System.out.println("服务启动"+port); ChannelFuture channelFuture = serverBootstrap.bind(port).sync(); channelFuture.channel().closeFuture().sync(); }finally { workGroup.shutdownGracefully(); bossGroup.shutdownGracefully(); } }
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { String data = (String) msg; System.out.println("服务端接收数据：" + data); data = data + " now:"+System.currentTimeMillis(); ctx.writeAndFlush(Unpooled.copiedBuffer(data, CharsetUtil.UTF_8)); }
服务端用了LineBasedFrameDecoder，以防止半包读写问题，客户端需要进行配合
客户端实现这个案例客户端实现有两个难点：
1、客户端方法如何能请求到SimpleChannelInboundHandler，即把需要发送的消息传到handler中
2、如何能等待服务端响应同步返回
第一个问题其实是如何把客户端输入的参数传入handler，所以我们需要把参数以构造函数传参的方式以此传入ChannelInitializer、SimpleChannelInboundHandler，这样handler就可以拿到客户端输入的数据了
下面的Controller里需要提前把channel准备好，如果RPC框架需要考虑通道与服务的关系
@RestController public class Controller { static String ip = "127.0.0.1"; static int port = 9876; static Bootstrap bootstrap; static NioEventLoopGroup worker; static { bootstrap = new Bootstrap(); worker = new NioEventLoopGroup(); bootstrap.group(worker); bootstrap.channel(NioSocketChannel .class) .option(ChannelOption.TCP_NODELAY, true) .remoteAddress(new InetSocketAddress(ip, port)); } @GetMapping("/nio/netty/server") public String test(String param){ CustomerChannelInitializer customerChannelInitializer = new CustomerChannelInitializer(param); bootstrap.handler(customerChannelInitializer); ChannelFuture channelFuture= null; String msg = ""; try { channelFuture = bootstrap.connect().sync(); } catch (InterruptedException e) { e.printStackTrace(); } return customerChannelInitializer.getResponse(); } }
public class CustomerChannelInitializer extends ChannelInitializer<SocketChannel> { private String response; private CountDownLatch countDownLatch; private String param; private ClientChannelHandlerAdapter clientChannelHandlerAdapter; public CustomerChannelInitializer(String param) { this.param = param; } @Override protected void initChannel(SocketChannel ch) throws Exception { countDownLatch = new CountDownLatch(1); clientChannelHandlerAdapter = new ClientChannelHandlerAdapter(param, this); ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(clientChannelHandlerAdapter); } public String getResponse() { try { countDownLatch.await(); } catch (InterruptedException e) { e.printStackTrace(); } return response; } public void setResponse(String response) { this.response = response; countDownLatch.countDown(); } }
public class ClientChannelHandlerAdapter extends SimpleChannelInboundHandler<ByteBuf> { private String param; private CustomerChannelInitializer customerChannelInitializer; public ClientChannelHandlerAdapter(String param, CustomerChannelInitializer customerChannelInitializer) { this.param = param; this.customerChannelInitializer = customerChannelInitializer; } @Override protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { System.out.println("客户端收到返回数据：" + msg.toString(CharsetUtil.UTF_8)); customerChannelInitializer.setResponse(msg.toString(CharsetUtil.UTF_8)); } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("客户端准备发送数据"); ctx.writeAndFlush(Unpooled.copiedBuffer(param + System.getProperty("line.separator"), CharsetUtil.UTF_8)); System.out.println("客户端发送数据完成"); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println("发生异常"); cause.printStackTrace(); ctx.close(); } }
我们来看第二个问题，由于netty是异步的，所以无法等待到服务端响应后调用客户端的channelRead0方法，controller就已经返回了，导致了网页显示的返回结果一直是空
主线程通过CountDownLatch来锁住没有返回结果的线程，直到工作线程获得结果并解锁
@Override protected void initChannel(SocketChannel ch) throws Exception { countDownLatch = new CountDownLatch(1); ……
public String getResponse() { try { countDownLatch.await(); } catch (InterruptedException e) { e.printStackTrace(); } return response; } public void setResponse(String response) { this.response = response; countDownLatch.countDown(); }