当服务器上的缓冲区中有连续的消息流时，如何检索字符串? [英] How to retrieve a String when there is a continuous flow of the messages at the server from a buffer?
问题描述
我使用Netty实现了一个基本的客户端和服务器程序.客户端将一组消息字符串发送到服务器，服务器接受并显示给终端.我可以对任意数量的字符串执行此操作.但是有时字符串会被切成一半或任何大小，并显示在服务器上.有什么办法可以消除这种情况?我们正在从客户端发送QuickFix字符串.
I implemented a basic Client and Server program using Netty. Client sends group of message strings to the server and the server accepts and displays it to the terminal. I can do this for any number of strings. But sometimes the strings are cut to half or any size and are displayed at the server. Is there any way to eliminate this? We are sending QuickFix Strings from Client.
下面是我的示例QuickFix字符串:
Below is my sample QuickFix string:
8 = FIX.4.29 = 0007935 = A49 = TTDS68AO56 = RaviEx34 = 152 = 20170427-14:05:04.572108 = 60141 = Y98 = 010 = 242
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"结尾.当所有字符串都插入缓冲区时，谁能帮我从连续缓冲区中检索字符串.
The character "r" in the above string is an SOH character. Now my requirement is: When I send multiple strings in a loop from ClientHandler to the Server,some strings are cut automatically and displayed at the Server side(May be due to the speed of Netty). I want to eliminate this. Every string starts from "8=" and ends with "10=xxx". Can anyone help me in retrieving the string from a continuous buffer as all the strings are inserted into a buffer.
现在，当我运行我的以下代码时，有时我会准确地获取我的字符串，有时它会显示Exception.该异常是由于Half Strings引起的.在FIXMESSAGEDECODER类中，我编写了用于检索以"8 ="开头和以"10 ="结尾的字符串的逻辑.
Now when I run my Below code , Sometimes I get my string exactly and sometimes it shows Exception.The exception is due to the Half Strings. In the FIXMESSAGEDECODER class I wrote the logic for retrieving the string starting with "8=" and ending with "10=".
任何人都可以帮助我如何准确地从缓冲区中检索消息字符串而不会呕吐消息的任何部分.
Can anyone help me how to retrieve message string exactly from the buffer without vomiting any part of the messages.
我的客户代码:
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);
}
}
推荐答案
不幸的是，不能保证您的字符串会一次出现.可能是当解码器第一次调用时，缓冲区将包含一段字符串，例如[8 = ...]，而在第二次调用时，将有[... 10 = XXX].另一点是，您可以获取字符串和下一个字符串的一部分，例如[8 = ... 10 = XXX8 = ...].而且，您必须考虑比数字8和10更好的线检测器.如果您确定像8=和10=XXX这样的线模式仅使用一次，请使用它.
Unfortunately, there is no guarantee that your string will come at once as a whole. It could be that when the decoder first called, the buffer will contain a piece of string such as [8=...], and on the second call there will be [...10=XXX]. Another point is that you can get your string and a part of next string like [8=...10=XXX8=...]. And you have to think of better line detectors than just numbers 8 and 10. If you are sure that at this line pattern like 8= and 10=XXX is used only once, then use it.
我建议您以测试驱动的开发样式重写解码器.幸运的是，解码器非常容易测试.首先，您编写了很多测试，在此您可以想到的是描述传入缓冲区的许多可能变体(完整字符串，一部分字符串，一次两个字符串).然后，编写您的解码器以通过所有这些测试.
I can suggest you to rewrite your decoder in Test-driven development style. Fortunately, decoders are very easy to test. First you write a lot of tests, where you describe so many possible variants of the incoming buffer (full string, part of string, two strings at once) as you can think of. Then you write your decoder to pass all this tests.
这篇关于当服务器上的缓冲区中有连续的消息流时，如何检索字符串?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
