Netty入门3

1.7. 流数据的传输处理

 

1.7.1. Socket Buffer的缺陷

对于例如TCP/IP这种基于流的传输协议实现，接收到的数据会被存储在socket的接受缓冲区内。不幸的是，这种基于流的传输缓冲区并不是一个包队列，而是一个字节队列。这意味着，即使你以两个数据包的形式发送了两条消息，操作系统却不会把它们看成是两条消息，而仅仅是一个批次的字节序 列。因此，在这种情况下我们就无法保证收到的数据恰好就是远程节点所发送的数据。例如，让我们假设一个操作系统的TCP/IP堆栈收到了三个数据包：

 

+-----+-----+-----+
| ABC | DEF | GHI |
+-----+-----+-----+

 

由于这种流传输协议的普遍性质，在你的应用中有较高的可能会把这些数据读取为另外一种形式：

 

+----+-------+---+---+
| AB | CDEFG | H | I |
+----+-------+---+---+

 

因此对于数据的接收方，不管是服务端还是客户端，应当重构这些接收到的数据，让其变成一种可让你的应用逻辑易于理解的更有意义的数据结构。在上面所述的这个例子中，接收到的数据应当重构为下面的形式：

 

+-----+-----+-----+
| ABC | DEF | GHI |
+-----+-----+-----+

 

1.7.2. 第一种方案

现在让我们回到时间协议服务客户端的例子中。我们在这里遇到了同样的问题。一个32位的整数是一个非常小的数据量，因此它常常不会被切分在不同的数据段内。然而，问题是它确实可以被切分在不同的数据段内，并且这种可能性随着流量的增加而提高。

 

最简单的方案是在程序内部创建一个可准确接收4字节数据的累积性缓冲。下面的代码是修复了这个问题后的TimeClientHandler实现。

package org.jboss.netty.example.time;  
 
import static org.jboss.netty.buffer.ChannelBuffers.*;  
 
import java.util.Date;  
 
@ChannelPipelineCoverage("one")  
public class TimeClientHandler extends SimpleChannelHandler {  
 
    private final ChannelBuffer buf = dynamicBuffer();  
 
    @Override 
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {  
        ChannelBuffer m = (ChannelBuffer) e.getMessage();  
        buf.writeBytes(m);  
         
        if (buf.readableBytes() >= 4) {  
            long currentTimeMillis = buf.readInt() * 1000L;  
            System.out.println(new Date(currentTimeMillis));  
            e.getChannel().close();  
        }  
    }  
 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {  
        e.getCause().printStackTrace();  
        e.getChannel().close();  
    }  
} 

 

 代码说明 

1) 这一次我们使用“one”做为ChannelPipelineCoverage的注解值。这是由于这个修改后的TimeClientHandler不在不 在内部保持一个buffer缓冲，因此这个TimeClientHandler实例不可以再被多个Channel通道或ChannelPipeline共 享。否则这个内部的buffer缓冲将无法缓冲正确的数据内容。

2) 动态的buffer缓冲也是ChannelBuffer的一种实现，其拥有动态增加缓冲容量的能力。当你无法预估消息的数据长度时，动态的buffer缓冲是一种很有用的缓冲结构。

3) 首先，所有的数据将会被累积的缓冲至buf容器。

4) 之后，这个处理器将会检查是否收到了足够的数据然后再进行真实的业务逻辑处理，在这个例子中需要接收4字节数据。否则，Netty将重复调用messageReceived方法，直至4字节数据接收完成。

 

这里还有另一个地方需要进行修改。你是否还记得我们把TimeClientHandler实例添加到了这个ClientBootstrap实例的默 认ChannelPipeline管道里？这意味着同一个TimeClientHandler实例将被多个Channel通道共享，因此接受的数据也将受 到破坏。为了给每一个Channel通道创建一个新的TimeClientHandler实例，我们需要实现一个 ChannelPipelineFactory管道工厂：

package org.jboss.netty.example.time;  
 
public class TimeClientPipelineFactory implements ChannelPipelineFactory {  
 
    public ChannelPipeline getPipeline() {  
        ChannelPipeline pipeline = Channels.pipeline();  
        pipeline.addLast("handler", new TimeClientHandler());  
        return pipeline;  
    }  
}

 现在，我们需要把TimeClient下面的代码片段：

 

TimeClientHandler handler = new TimeClientHandler();  
bootstrap.getPipeline().addLast("handler", handler); 

 替换为：

bootstrap.setPipelineFactory(new TimeClientPipelineFactory()); 

 虽然这看上去有些复杂，并且由于在TimeClient内部我们只创建了一个连接（connection），因此我们在这里确实没必要引入TimeClientPipelineFactory实例。

然而，当你的应用变得越来越复杂，你就总会需要实现自己的ChannelPipelineFactory，这个管道工厂将会令你的管道配置变得更加具有灵活性。

1.7.3. 第二种方案

 

虽然第二种方案解决了时间协议客户端遇到的问题，但是这个修改后的处理器实现看上去却不再那么简洁。设想一种更为复杂的，由多个可变长度字段组成的协议。你的ChannelHandler实现将变得越来越难以维护。

正如你已注意到的，你可以为一个ChannelPipeline添加多个ChannelHandler，因此，为了减小应用的复杂性，你可以把这个 臃肿的ChannelHandler切分为多个独立的模块单元。例如，你可以把TimeClientHandler切分为两个独立的处理器：

 TimeDecoder，解决数据分段的问题。

 TimeClientHandler，原始版本的实现。

幸运的是，Netty提供了一个可扩展的类，这个类可以直接拿过来使用帮你完成TimeDecoder的开发：

package org.jboss.netty.example.time;  
 
 
public class TimeDecoder extends FrameDecoder {  
 
    @Override 
    protected Object decode(  
            ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer)  {  
             
        if (buffer.readableBytes() < 4) {  
            return null;   
        }  
         
        return buffer.readBytes(4);  
    }  
} 

 代码说明

1) 这里不再需要使用ChannelPipelineCoverage的注解，因为FrameDecoder总是被注解为“one”。

2) 当接收到新的数据后，FrameDecoder会调用decode方法，同时传入一个FrameDecoder内部持有的累积型buffer缓冲。

3) 如果decode返回null值，这意味着还没有接收到足够的数据。当有足够数量的数据后FrameDecoder会再次调用decode方法。

4) 如果decode方法返回一个非空值，这意味着decode方法已经成功完成一条信息的解码。FrameDecoder将丢弃这个内部的累计型缓冲。请注 意你不需要对多条消息进行解码，FrameDecoder将保持对decode方法的调用，直到decode方法返回非空对象。

如果你是一个勇于尝试的人，你或许应当使用ReplayingDecoder，ReplayingDecoder更加简化了解码的过程。为此你需要查看API手册获得更多的帮助信息。

package org.jboss.netty.example.time;  
 
public class TimeDecoder extends ReplayingDecoder<VoidEnum> {  
 
    @Override 
    protected Object decode(  
            ChannelHandlerContext ctx, Channel channel,  
            ChannelBuffer buffer, VoidEnum state) {  
             
        return buffer.readBytes(4);  
    }  
} 

 此外，Netty还为你提供了一些可以直接使用的decoder实现，这些decoder实现不仅可以让你非常容易的实现大多数协议，并且还会帮你避免某些臃肿、难以维护的处理器实现。请参考下面的代码包获得更加详细的实例：

org.jboss.netty.example.factorial for a binary protocol, and

 org.jboss.netty.example.telnet for a text line-based protocol