java - 当服务器上有来自缓冲区的连续消息流时，如何检索字符串？

Question

我使用 Netty 实现了一个基本的客户端和服务器程序。客户端向服务器发送一组消息字符串，服务器接受并显示到终端。我可以对任意数量的字符串执行此操作。但有时字符串会被切成一半或任意大小并显示在服务器上。有什么办法可以消除这种情况吗？我们正在从客户端发送 QuickFix 字符串。

下面是我的示例 QuickFix 字符串:

8=FIX.4.29=0007935=A49=TTDS68AO56=RaviEx34=152=20170427-14:05:04.572108=60141=Y98=010=242

上面字符串中的字符“r”是一个SOH字符。现在我的要求是:当我从ClientHandler循环发送多个字符串到服务器时，一些字符串被自动剪切并显示在服务器端(可能是由于Netty的速度)。我想消除这个。每个字符串都从“8=”开始，以“10=xxx”结束。当所有字符串都插入到缓冲区中时，任何人都可以帮助我从连续缓冲区中检索字符串。

现在，当我运行下面的代码时，有时我准确地得到我的字符串，有时它显示异常。异常是由于半字符串造成的。在 FIXMESSAGEDECODER 类中，我编写了检索以“8=”开头并以“10=”结尾的字符串的逻辑。

任何人都可以帮助我如何从缓冲区中准确地检索消息字符串而不吐出消息的任何部分。

我的客户代码:

 public class EchoClient {

        private final String host;
        private final int port;

        public EchoClient(String host, int port) {
            this.host = host;
            this.port = port;
        }
    public void start() throws Exception{
    EventLoopGroup group = new NioEventLoopGroup();
    try{
     Bootstrap b = new Bootstrap();
     b.group(group).channel(NioSocketChannel.class)
             .remoteAddress(new InetSocketAddress(host, port))
             .handler(new ChannelInitializer<SocketChannel>(){

             @Override
             public void initChannel(SocketChannel ch) throws Exception{
             ch.pipeline().addLast(new EchoClientHandler());
             }
             });

                ChannelFuture future = b.connect().sync();
               future.channel().closeFuture().sync();
         }

            finally {
                group.shutdownGracefully().sync();
            }
        }

        public static void main (String [] args) throws Exception {
            new EchoClient("127.0.0.1", 11235).start();
        }
    }

我的 ClientHandler :

public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf>{

@Override
 public void channelActive(ChannelHandlerContext ctx){
   System.out.println("Connected");
   int i=0;
   while(i<100){
ctx.writeAndFlush(Unpooled.copiedBuffer("8=FIX.4.29=0007935=A49=TTDS68AO56=RaviEx34=152=20170427-14:05:04.572108=60141=Y98=010=242\n", 
CharsetUtil.UTF_8));

  i++;
  }
}

 @Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
   System.out.println("Client received: " + in.toString(CharsetUtil.UTF_8));
}

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

我的服务器:

public class EchoServer{

 private final int port;
 public EchoServer(int port) {
      this.port = port;
 }

 public void start() throws Exception {
     EventLoopGroup group = new NioEventLoopGroup();
     try {
           ServerBootstrap b = new ServerBootstrap();
           b.group(group)
            .channel(NioServerSocketChannel.class)
            .localAddress(new InetSocketAddress(port))
            .childHandler(new ChannelInitializer<SocketChannel>() {
             @Override
              public void initChannel(SocketChannel ch) throws Exception {
              System.out.println("New client connected: " + ch.localAddress());
              ch.pipeline().addLast(new FixMessageDecoder());
              }
            });

         ChannelFuture f = b.bind().sync();
         f.channel().closeFuture().sync();
     }
     finally {
                group.shutdownGracefully().sync();
            }
        }

        public static void main (String [] args) throws Exception {
        new EchoServer(11235).start();
        }
}

我的 FixMessage 解码器:

public class FixMessageDecoder extends MessageToMessageDecoder<ByteBuf> {

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception{
  String messg = in.toString(CharsetUtil.UTF_8);
  String str = messg.substring(messg.indexOf("8"), messg.lastIndexOf("10")+6);
  System.out.println(str);
 }
}   

Answer 1

不幸的是，不能保证您的字符串会作为一个整体立即到达。可能是当解码器第一次调用时，缓冲区将包含一段字符串，例如 [8=...]，第二次调用时将包含 [...10=XXX]。另一点是你可以获得你的字符串和下一个字符串的一部分，例如 [8=...10=XXX8=...]。你必须考虑比数字 8 和 10 更好的线条检测器。如果你确定在这个线条模式中，像 8= 和 10=XXX 只使用一次，然后使用它。

我可以建议您以测试驱动开发风格重写解码器。幸运的是，解码器非常容易测试。首先，您编写大量测试，在其中描述传入缓冲区的多种可能变体(完整字符串、部分字符串、一次两个字符串)。然后编写解码器来通过所有这些测试。