WCF扩展之实现ZeroMQ绑定和protocolBuffer消息编码(二)实现IRequestChannel(2016-03-15 12:35)

时间:2023-03-09 13:05:16
WCF扩展之实现ZeroMQ绑定和protocolBuffer消息编码(二)实现IRequestChannel(2016-03-15 12:35)

这是这个系列的第二篇,其他的文章请点击下列目录

WCF扩展之实现ZeroMQ绑定和protocolBuffer消息编码(一)概要设计

WCF扩展之实现ZeroMQ绑定和protocolBuffer消息编码(二)实现IRequestChannel

WCF扩展之实现ZeroMQ绑定和protocolBuffer消息编码(三)实现ReplyChannel

从今天开始,一步步介绍我是如何实现自定义的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, , bufferManager);

            socket.Send(requestData.Array);

转换的方式是通过encoder对象获得的,encoder就是通过MessageEncodingBindingElement创建的。MessageEncodingBindingElement是WCF信道栈的一部分,但是Message并不是沿着每个信道穿过,MessageEncodingBindingElement就是这样的信道,它只是被TransportBindingElement使用,在TransportChannel需要编码和解码时,TransportBindingElement的encoder会站出来执行自己的职责。

接收服务端的回复时,是这样做的:

            byte[] replyData = bufferManager.TakeBuffer();

            int replaySize = socket.Receive(replyData);

            Message response = encoder.ReadMessage(
new ArraySegment<byte>(replyData, , 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, , 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。