java - Netty - 阻塞操作在完成之前不接受客户端请求

Question

[问题]:我有一个 Netty 服务器，它将通过 TCP 监听客户端，处理请求并发回响应。问题是处理部分需要时间，因为它需要应用程序用户的验证。客户端接连发送多个请求，而服务器似乎在前一个请求完成之前才接受客户端的请求。

我已经复制了这样的场景:客户端将一个接一个地发送几个请求，服务器将花费一些时间来接受请求并进行处理。通过将阻塞操作传递给另一个线程并完成剩余的操作，我可以很好地处理单个请求阻塞请求。但这仍然不接受来自客户端的第二个请求。它仍然会等待阻塞操作完成，然后再接受来自客户端的另一个请求。

客户端代码

int count =0;
long currentTime = System.currentTimeMillis();
long trigger = currentTime + 5000;
long stop_loop = currentTime + 130000;
while(count<2){
  boolean loop = true;
  while(loop){
    if(System.currentTimeMillis()>trigger){
      trigger += 5000;
      count++;
      loop = false;
    }
  }
  System.out.println("SENDING FOR COUNTER - "+count);
  content = Unpooled.copiedBuffer(b);
  System.out.println("Channel Active Method Called: "+ctx);
  ctx.write(content);
  ctx.flush();
}

服务器代码:

public void channelRead(ChannelHandlerContext ctx, Object msg) { 
final ExecutorService blockingThreadPool = Executors.newFixedThreadPool(10);
  try {
  System.out.println("server receive orde:"+body+"the counter is:" + ++counter);
  ctx.executor().execute(new Runnable(){
     public void run(){
        try{
            new MyBusinessLogicHandler(blockingThreadPool).channelRead(ctx , b_buf);
        }
        catch(Exception e ){
           logger.error(e.getMessage());
        }
     }
  }
  //ByteBuf resp = Unpooled.copiedBuffer(currentTime,CharsetUtil.UTF_8);
  //ByteBuf resp = Unpooled.wrappedBuffer(response.getBytes());
  //ctx.writeAndFlush(resp);
 // ctx.write(resp);
  //ctx.flush();
}

我的业务逻辑

 @Override
    public void channelRead(ChannelHandlerContext ctx, Object arg1)
                throws Exception{
         System.out.println(" INSIDE BUSINESS HANDLER - OBJ");
         System.out.println(arg1);

         String x = ctx.channel().attr(DiscardServerHandler.CHECKSUMKEY).get();
         System.out.println("X -"+x);

         ChannelHandlerContext temp = ctx;

         resp = Unpooled.wrappedBuffer(x.getBytes());
         ctx.write(resp);
         ctx.flush();
         ByteBuf buf = (ByteBuf) arg1;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body = new String(req,CharsetUtil.UTF_8);
            System.out.println(new Date());
            System.out.println("Business logic receive order : " + body);

            System.out.println("SERVER LISTENER READ");
            boolean loop = true;
            long currentTime = System.currentTimeMillis();
            long trigger = currentTime + 3000;
            long stop_loop = currentTime + 20000;
            int count = 0;
                System.out.println("Server READ");
//              String response = "Server Complete Response ";

                while(loop){
                    long now = System.currentTimeMillis();

                    if(now<stop_loop){

                        if(now>trigger){
                            count++;
                            System.out.println("Now is triggered @" + count + "-- "+(count*5)+"secs passed");
                            trigger += 5000;
                        }

                    }
                    else if(now>stop_loop){
                        loop = false;
                        System.out.println("Loop Completes");
                    }
                }
            String re = "BUSINESS LOGIC RESPONSE";  
            resp = Unpooled.wrappedBuffer(re.getBytes());

            temp.write(resp);
            temp.flush();

     }

我想要的只是接受所有客户端请求，无论之前的请求是否已完成/未完成。

Answer 1

这是一种经典模式，您需要从同步服务切换到异步(“异步”)服务。异步服务的想法是，您接收来自客户端的请求，然后验证它并将其存储在服务器上的某些存储中(通常是队列)中的“待办事项”列表中。然后返回响应(在本例中称为确认(“ack”))。在您的响应中，您向客户返回其请求的 ID 以及返回答案的媒介。 (通常是一个队列)。这样客户就会知道在哪里收听结果。同时，您现在可以在服务器上处理收到的请求，而无需阻止客户端。您还可以拥有多个“工作人员”实例，它们从存储中获取传入请求并进行处理。处理请求后，您的工作进程会使用您在“ack”中返回给客户端的相同 ID 将其发布到您在“ack”中返回给客户端的介质(队列)。欲了解更多信息，请看这里:Asynchronous vs synchronous execution, what does it really mean?