从今天开始，一步步介绍我是如何实现自定义的ZeroMQ绑定和protocolBuffer消息编码的。
本系列的想法主要来源于蒋金楠Artech的系列博客-WCF后续之旅，和WCF开发团队成员carlosfigueira的WCF Extension系列博客。
首先要明确的是，通信和编码的实现是分离的。WCF也是这样设计的。自定义Binding包含BindingElementCollection集合，这个集合可以添加多个ElementBinding。但是有两个ElementBinding必不可少，TransportBindingElement、MessageEncodingBindingElement。而ZeroMQ只是提供了通信技术，不涉及具体的编码，它的传输数据是指定长度的字符串，至于字符串以用何种方式编码，ZeroMQ是不知道的。因此使得ZeroMQ作为WCF的自定义Binding中的TransportBindingElement的实现成为可能。我们先不管编码，先考虑传输。
WCF的通道栈在上述两个系列博客中已经有详细的介绍，我就不想重复了，直接说说我是如何将WCF和ZeroMQ衔接的。
实现IRequestChannel
新建类型ZMQRequestChannel，继承于ChannelBase，ChannelBase是WCF提供的通道基类。
首先要创建ZeroMQ的socket，类型是REQ。
socket = this.zmqContext.CreateSocket(SocketType.REQ); socket.Connect(zmqAddress);
实现IRequestChannel的方法
public Message Request(Message message, TimeSpan timeout)
Request负责发送请求，阻塞当前线程，直到接收到服务器的回复。发送请求时，将Message转换为byte[]，交给ZeroMQ发送
ArraySegment<byte> requestData = encoder.WriteMessage(message, 1000, bufferManager);
socket.Send(requestData.Array);
转换的方式是通过encoder对象获得的，encoder就是通过MessageEncodingBindingElement创建的。MessageEncodingBindingElement是WCF信道栈的一部分，但是Message并不是沿着每个信道穿过，MessageEncodingBindingElement就是这样的信道，它只是被TransportBindingElement使用，在TransportChannel需要编码和解码时，TransportBindingElement的encoder会站出来执行自己的职责。
接收服务端的回复时，是这样做的：
byte[] replyData = bufferManager.TakeBuffer(1000);
int replaySize = socket.Receive(replyData);
Message response = encoder.ReadMessage(
new ArraySegment<byte>(replyData, 0, replaySize), bufferManager);
bufferManager.ReturnBuffer(replyData);
这里使用了BufferManager提供一个接收缓存，缓存的大小暂时设为1000。使用缓存的目的在于分配时的效率更高。当接收完成时，记得回收。
IRequestChannel还需要实现异步的发送方法
IAsyncResult BeginRequest(Message message, AsyncCallback callback, object state);
IAsyncResult BeginRequest(Message message, TimeSpan timeout, AsyncCallback callback, object state);
Message EndRequest(IAsyncResult result);
参考刚才的同步方法，发送分为编码、发送。其中编码在本地完成，不会超时，发送由于依赖网络环境和服务器状态，需要将这部分移到异步过程中。
由于发送异步方法在ReplyChannel中也会使用，因此我创建了ZMQChannelBase基类，派生于ChannelBase，ZMQRequestChannel继承于新的ZMQChannelBase基类。把异步发送方法放在ZMQChannelBase中。
public IAsyncResult BeginRequest(Message message, AsyncCallback callback, object state)
{
return base.BeginSendMessage(message, this.DefaultSendTimeout, callback, state);
}
public IAsyncResult BeginSendMessage(Message message, TimeSpan timeout, AsyncCallback callback, object state)
{
base.ThrowIfDisposedOrNotOpen();
ArraySegment<byte> requestData = encoder.WriteMessage(message, 1000, bufferManager);
return this.BeginWriteData(requestData, timeout, callback, state);
}
BeginWriteData方法返回一个AsyncResult对象，实现异步接口IAsyncResult
class SocketSendAsyncResult : AsyncResult
{
ZMQChannelBase channel;
ArraySegment<byte> buffer;
public SocketSendAsyncResult(ArraySegment<byte> buffer, ZMQChannelBase channel, AsyncCallback callback, object state)
: base(callback, state)
{
this.channel = channel;
this.buffer = buffer;
this.StartSending();
}
void StartSending()
{
SocketSendHandle handler = new SocketSendHandle(delegate(ZmqSocket socket, byte[] data)
{
socket.Send(data);
serviceHanledDone.Set();
});
IAsyncResult sendResult = handler.BeginInvoke(channel.socket, buffer.Array, OnSend, null);
}
static void OnSend(IAsyncResult result)
{
if (result.CompletedSynchronously)
{
return;
}
SocketSendAsyncResult thisPtr = (SocketSendAsyncResult)result.AsyncState;
Exception completionException = null;
bool shouldComplete = false;
if (shouldComplete)
{
thisPtr.Complete(false, completionException);
}
}
public static void End(IAsyncResult result)
{
AsyncResult.End<SocketSendAsyncResult>(result);
}
}
至此，RequestChannel就完成了，相比ReplyChannel，RequestChannel比较简单。下一篇介绍ReplyChannel。