客户端(springmvc)调用netty构建的nio服务端,获得响应后返回页面(同步响应)

时间:2024-11-21 20:34:13

后面考虑通过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();
}