netty系列之:netty對marshalling的支援
簡介
在之前的文章中我們講過了，jboss marshalling是一種非常優秀的java對象序列化的方式，它可以兼容JDK自帶的序列化，同時也提供了性能和使用上的優化。
那麼這麼優秀的序列化工具可不可以用在netty中作為消息傳遞的方式呢？
答案當然是肯定的，在netty中一切皆有可能。
netty中的marshalling provider
回顧一下jboss marshalling的常用用法，我們需要從MarshallerFactory中創建出Marshaller，因為mashaller有不同的實現，所以需要指定具體的實現來創建MarshallerFactory,如下所示：
MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("river");
這個MarshallerFactory實際上就是一個MarshallerProvider。
netty中定義了這樣的一個介面：
public interface MarshallerProvider {
Marshaller getMarshaller(ChannelHandlerContext ctx) throws Exception;
}
MarshallerProvider實際上就做了和MarshallerFactory等同的工作。
既然MarshallerProvider是一個介面，那麼它有哪些實現呢？
在netty中它有兩個實現類，分別是DefaultMarshallerProvider和ThreadLocalMarshallerProvider。
兩者有什麼區別呢？
先來看一下DefaultMarshallerProvider：
public class DefaultMarshallerProvider implements MarshallerProvider {
private final MarshallerFactory factory;
private final MarshallingConfiguration config;
public DefaultMarshallerProvider(MarshallerFactory factory, MarshallingConfiguration config) {
this.factory = factory;
this.config = config;
}
public Marshaller getMarshaller(ChannelHandlerContext ctx) throws Exception {
return factory.createMarshaller(config);
}
}
顧名思義，DefaultMarshallerProvider就是marshallerProvider的默認實現，從具體的實現程式碼中，我們可以看出，DefaultMarshallerProvider實際上需要傳入MarshallerFactory和MarshallingConfiguration作為參數，然後使用傳入的MarshallerFactory來創建具體的marshaller Provider，和我們手動創建marshaller的方式是一致的。
但是上面的實現中每次getMarshaller都需要重新從factory中創建一個新的，性能上可能會有問題。所以netty又實現了一個新的ThreadLocalMarshallerProvider:
public class ThreadLocalMarshallerProvider implements MarshallerProvider {
private final FastThreadLocal<Marshaller> marshallers = new FastThreadLocal<Marshaller>();
private final MarshallerFactory factory;
private final MarshallingConfiguration config;
public ThreadLocalMarshallerProvider(MarshallerFactory factory, MarshallingConfiguration config) {
this.factory = factory;
this.config = config;
}
@Override
public Marshaller getMarshaller(ChannelHandlerContext ctx) throws Exception {
Marshaller marshaller = marshallers.get();
if (marshaller == null) {
marshaller = factory.createMarshaller(config);
marshallers.set(marshaller);
}
return marshaller;
}
}
ThreadLocalMarshallerProvider和DefaultMarshallerProvider的不同之處在於，ThreadLocalMarshallerProvider中保存了一個FastThreadLocal的對象，FastThreadLocal是JDK中ThreadLocal的優化版本，比ThreadLocal更快。
在getMarshaller方法中，先從FastThreadLocal中get出Marshaller對象，如果Marshaller對象不存在，才從factory中創建出一個Marshaller對象，最後將Marshaller對象放到ThreadLocal中。
有MarshallerProvider就有和他對應的UnMarshallerProvider：
public interface UnmarshallerProvider {
Unmarshaller getUnmarshaller(ChannelHandlerContext ctx) throws Exception;
}
netty中的UnmarshallerProvider有三個實現類，分別是DefaultUnmarshallerProvider,ThreadLocalUnmarshallerProvider和ContextBoundUnmarshallerProvider.
前面的兩個DefaultUnmarshallerProvider,ThreadLocalUnmarshallerProvider跟marshaller的是實現是一樣的，這裡就不重複講解了。
我們主要來看一下ContextBoundUnmarshallerProvider的實現。
從名字上我們可以看出，這個unmarshaller是和ChannelHandlerContext相關的。
ChannelHandlerContext表示的是channel的上下文環境，它裡面有一個方法叫做attr，可以保存和channel相關的屬性：
<T> Attribute<T> attr(AttributeKey<T> key);
ContextBoundUnmarshallerProvider的做法就是將Unmarshaller存放到context中，每次使用的時候先從context中獲取，如果沒有取到再從factroy中獲取。
我們來看下ContextBoundUnmarshallerProvider的實現：
public class ContextBoundUnmarshallerProvider extends DefaultUnmarshallerProvider {
private static final AttributeKey<Unmarshaller> UNMARSHALLER = AttributeKey.valueOf(
ContextBoundUnmarshallerProvider.class, "UNMARSHALLER");
public ContextBoundUnmarshallerProvider(MarshallerFactory factory, MarshallingConfiguration config) {
super(factory, config);
}
@Override
public Unmarshaller getUnmarshaller(ChannelHandlerContext ctx) throws Exception {
Attribute<Unmarshaller> attr = ctx.channel().attr(UNMARSHALLER);
Unmarshaller unmarshaller = attr.get();
if (unmarshaller == null) {
unmarshaller = super.getUnmarshaller(ctx);
attr.set(unmarshaller);
}
return unmarshaller;
}
}
ContextBoundUnmarshallerProvider繼承自DefaultUnmarshallerProvider，在getUnmarshaller方法首先從ctx取出unmarshaller,如果沒有的話，則調用DefaultUnmarshallerProvider中的getUnmarshaller方法取出unmarshaller。
Marshalling編碼器
上面的章節中我們獲取到了marshaller,接下來看一下如何使用marshaller來進行編碼和解碼操作。
首先來看一下編碼器MarshallingEncoder，MarshallingEncoder繼承自MessageToByteEncoder,接收的泛型是Object:
public class MarshallingEncoder extends MessageToByteEncoder<Object>
是將Object對象編碼成為ByteBuf。回顧一下之前我們講到的通常對象的編碼都需要用到一個對象長度的欄位，用來分割對象的數據，同樣的MarshallingEncoder也提供了一個4個位元組的LENGTH_PLACEHOLDER，用來存儲對象的長度。
具體的看一下它的encode方法：
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
Marshaller marshaller = provider.getMarshaller(ctx);
int lengthPos = out.writerIndex();
out.writeBytes(LENGTH_PLACEHOLDER);
ChannelBufferByteOutput output = new ChannelBufferByteOutput(out);
marshaller.start(output);
marshaller.writeObject(msg);
marshaller.finish();
marshaller.close();
out.setInt(lengthPos, out.writerIndex() - lengthPos - 4);
}
encode的邏輯很簡單，首先從provider中拿到marshaller對象，然後先向out中寫入4個位元組的LENGTH_PLACEHOLDER，接著使用marshaller向
out中寫入編碼的對象，最後根據寫入對象長度填充out，得到最後的輸出。
因為encode的數據保存的有長度數據，所以decode的時候就需要用到一個frame decoder叫做LengthFieldBasedFrameDecoder。
通常有兩種方式來使用LengthFieldBasedFrameDecoder，一種是將LengthFieldBasedFrameDecoder加入到pipline handler中，decoder只需要處理經過frame decoder處理過後的對象即可。
還有一種方法就是這個decoder本身就是一個LengthFieldBasedFrameDecoder。
這裡netty選擇的是第二種方法，我們看下MarshallingDecoder的定義：
public class MarshallingDecoder extends LengthFieldBasedFrameDecoder
首先需要在構造函數中指定LengthFieldBasedFrameDecoder的欄位長度，這裡調用了super方法來實現：
public MarshallingDecoder(UnmarshallerProvider provider, int maxObjectSize) {
super(maxObjectSize, 0, 4, 0, 4);
this.provider = provider;
}
並且重寫了extractFrame方法：
protected ByteBuf extractFrame(ChannelHandlerContext ctx, ByteBuf buffer, int index, int length) {
return buffer.slice(index, length);
}
最後再看下decode方法：
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
ByteBuf frame = (ByteBuf) super.decode(ctx, in);
if (frame == null) {
return null;
}
Unmarshaller unmarshaller = provider.getUnmarshaller(ctx);
ByteInput input = new ChannelBufferByteInput(frame);
try {
unmarshaller.start(input);
Object obj = unmarshaller.readObject();
unmarshaller.finish();
return obj;
} finally {
unmarshaller.close();
}
}
decode的邏輯也很簡單，首先調用super方法decode出frame ByteBuf。然後再調用unmarshaller實現對象的讀取，最後將改對象返回。
Marshalling編碼的另外一種實現
上面我們講到對對象的編碼使用的是LengthFieldBasedFrameDecoder,根據對象實際數據之前的一個length欄位來確定欄位的長度，從而讀取真實的數據。
那麼可不可以不指定對象長度也能夠準確的讀取對象呢？
其實也是可以的，我們可以不斷的嘗試讀取數據，直到找到合適的對象數據為止。
看過我之前文章的朋友可能就想到了，ReplayingDecoder不就是做這個事情的嗎？在ReplayingDecoder中會不斷的重試，直到找到符合條件的消息為止。
於是netty基於ReplayingDecoder也有一個marshalling編碼解碼的實現，叫做CompatibleMarshallingEncoder和CompatibleMarshallingDecoder。
CompatibleMarshallingEncoder很簡單，因為不需要對象的實際長度，所以直接使用marshalling編碼即可。
public class CompatibleMarshallingEncoder extends MessageToByteEncoder<Object> {
private final MarshallerProvider provider;
public CompatibleMarshallingEncoder(MarshallerProvider provider) {
this.provider = provider;
}
@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
Marshaller marshaller = provider.getMarshaller(ctx);
marshaller.start(new ChannelBufferByteOutput(out));
marshaller.writeObject(msg);
marshaller.finish();
marshaller.close();
}
}
CompatibleMarshallingDecoder繼承了ReplayingDecoder:
public class CompatibleMarshallingDecoder extends ReplayingDecoder<Void>
它的decode方法的核心就是調用unmarshaller的方法：
Unmarshaller unmarshaller = provider.getUnmarshaller(ctx);
ByteInput input = new ChannelBufferByteInput(buffer);
if (maxObjectSize != Integer.MAX_VALUE) {
input = new LimitingByteInput(input, maxObjectSize);
}
try {
unmarshaller.start(input);
Object obj = unmarshaller.readObject();
unmarshaller.finish();
out.add(obj);
} catch (LimitingByteInput.TooBigObjectException ignored) {
discardingTooLongFrame = true;
throw new TooLongFrameException();
} finally {
unmarshaller.close();
}
注意，這裡解碼的時候會有兩種異常，第一種異常就是unmarshaller.readObject時候的異常，這種異常會被ReplayingDecoder捕獲從而重試。
還有一種就是欄位太長的異常，這種異常無法處理只能放棄：
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
if (cause instanceof TooLongFrameException) {
ctx.close();
} else {
super.exceptionCaught(ctx, cause);
}
}
總結
以上就是在netty中使用marshalling進行編碼解碼的實現。原理和對象編碼解碼是很類似的，大家可以對比分析一下。
本文已收錄於 //www.flydean.com/17-1-netty-marshalling/
最通俗的解讀，最深刻的乾貨，最簡潔的教程，眾多你不知道的小技巧等你來發現！
歡迎關注我的公眾號:「程式那些事」,懂技術，更懂你！
