实现一个双工服务（使用 WCF）--- WCF 初学者教程

引言

在本文的开头，我们先来谈谈 WCF 中的消息传输模式。

消息传输模式

• 请求-回复 - 默认情况下，所有 WCF 都会以请求-回复模式运行。这意味着当客户端向 WCF 服务发出请求时，客户端会等待从服务接收响应（直到收到超时，默认值为 1 分钟，可以通过 ReceivedTimeOut 对象更改）。如果服务在接收超时时间内没有响应，客户端将收到 TimeOutException。

• 单向 - 在单向传输模式下，客户端将请求发送到服务器，而不关心服务执行是否成功或失败。服务器端没有返回值，是单向通信。客户端仅在将其调用分派到服务时会被阻塞一小段时间。服务抛出的任何异常都不会到达客户端。

     ​     客户端在向服务器发出单向调用后，将继续执行其语句，无需等待服务器执行完毕。但有时，当单向调用到达服务时，它们可能不会一次分派，具体取决于服务的配置的 并发 模式行为。如果排队的消息数量超过了队列的容量，即使发出了单向调用，客户端也会被阻塞。

​[ServiceContract]
public interface IMyService
{
    [OperationContract(IsOneWay=true)]
    void MyMethod(string parameter);
}

您还可以将单向通信与会话性服务一起使用。

[ServiceContract(SessionMode = SessionMode.Required)]
interface IMyContract
{
    [OperationContract(IsOneWay = true)]
    void MyMethod();
}

• 回调服务 - 到目前为止，我们看到所有客户端都会调用服务来完成任务。但 WCF 也提供了服务调用客户端的功能。因此，我们称之为双工传输模式。

回调服务

• HTTP 协议是无连接的，因此不支持回调操作。因此，BasicHttpBinding 和 WSHttpBinding 不能用于回调服务。
• WSDualHttpBinding 是为回调操作而设计的。
• 所有这些绑定都包括 TCP 和 IPC 协议支持双工通信。

步骤 1：配置回调契约

实际上，它定义了客户端应该具有哪些操作。

//IPushCallback.cs
​namespace WCFTraining.DuplexStreamingService
{
    public interface IPushCallback
    {
        [OperationContract(IsOneWay = true)]
        void ReceiveData(string data);

        [OperationContract(IsOneWay = true)]
        void ReceiveStream(Stream stream);

        [OperationContract(IsOneWay = true)]
        void ReceiveLog(List<string> log);
    }
}

步骤 2：定义服务及其实现

//IDuplexService.cs
namespace IISHost.DuplexStreamingService
{
    [ServiceContract(CallbackContract = typeof(IPushCallback))]
    public interface IDuplexService
    {
        //Duplex operation
        [OperationContract(IsOneWay = true)]
        void StartPushingData();

        [OperationContract(IsOneWay = true)]
        void StopPushingData();

        [OperationContract(IsOneWay = true)]
        void StartPushingStream();

        [OperationContract(IsOneWay = true)]
        void StopPushingStream();

        //Logging
        [OperationContract(IsOneWay = true)]
        void GetLog();
    }
}

//DuplexService.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.ServiceModel;
using System.Threading;

namespace WCFTrrainning.DuplexStreamingService
{
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Reentrant)]
    public class DuplexService : IDuplexService
    {
        List<string> log = new List<string>();
        static bool continuePushingData;

        #region Duplex operation
        public void StartPushingData()
        {
            log.Add("StartPushingData");
            continuePushingData = true;
            IPushCallback pushCallPackChannel = OperationContext.Current.GetCallbackChannel<IPushCallback>();
            ThreadPool.QueueUserWorkItem(new WaitCallback(PushData), pushCallPackChannel);
        }

        public void StopPushingData()
        {
            log.Add("StopPushingData");
            continuePushingData = false;
        }

        public void StartPushingStream()
        {
            log.Add("StartPushingStream");
            IPushCallback pushCallbackChannel = OperationContext.Current.GetCallbackChannel<IPushCallback>();
            ThreadPool.QueueUserWorkItem(new WaitCallback(PushStream), pushCallbackChannel);
        }

        public void StopPushingStream()
        {
            log.Add("StopPushingStream");
            localStream.StopStreaming = true;
        }

        public void GetLog()
        {
            IPushCallback pushCallbackChannel = OperationContext.Current.GetCallbackChannel<IPushCallback>();
            pushCallbackChannel.ReceiveLog(log);
        }

        private static void PushData(object state)
        {
            IPushCallback pushCallPackChannel = state as IPushCallback;

            do
            {
                pushCallPackChannel.ReceiveData(CreateInterestingString(rand.Next(4, 256)));
            } while (continuePushingData);

            pushCallPackChannel.ReceiveData("Last Message");
        }

        void PushStream(object state)
        {
            IPushCallback pushCallbackChannel = state as IPushCallback;
            localStream = new FlowControlledStream();
            localStream.ReadThrottle = TimeSpan.FromMilliseconds(800);

            pushCallbackChannel.ReceiveStream(localStream);
        }

        #endregion
    }
}

该服务将托管在 IIS 中。因此，我们需要编辑 web.config

<version="1.0" encoding="utf-8"?>
<configuration>
  <appSettings>
    <add key="aspnet:UseTaskFriendlySynchronizationContext" value="true" />
  </appSettings>
  <system.web>
    <compilation debug="true">
    </compilation>
  </system.web>
    <system.serviceModel>
      <services>
        <service name="IISHost.DuplexStreamingService.DuplexService">
          <endpoint address="" binding="netTcpBinding" contract="IISHost.DuplexStreamingService.IDuplexService"></endpoint>
          <endpoint address="mex" binding="mexTcpBinding" contract="IMetadataExchange"></endpoint>
        </service>
      </services>
      <behaviors>
        <serviceBehaviors>
          <behavior name="">
            <serviceMetadata httpGetEnabled="true" httpsGetEnabled="true" />
            <serviceDebug includeExceptionDetailInFaults="false"/>
          </behavior>
        </serviceBehaviors>
      </behaviors>
    </system.serviceModel>
    <system.webServer>
        <directoryBrowse enabled="true" />
    </system.webServer>
</configuration>

客户端

步骤 3：实现回调契约

//ClientReceiver.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using DuplexStreamingService;

namespace DuplexStreamingClient
{
    public class ClientReceiver : IPushCallback
    {
        public ManualResetEvent LogReceived { get; set; }
        public ManualResetEvent ReceiveDataInvoked { get; set; }
        public ManualResetEvent ReceiveDataCompleted { get; set; }
        public ManualResetEvent ReceiveStreamInvoked { get; set; }
        public ManualResetEvent ReceiveStreamCompleted { get; set; }
        public string Name { get; set; }

        public List<string> ServerLog { get; set; }

        public ClientReceiver()
        {
            LogReceived = new ManualResetEvent(false);
            ReceiveDataInvoked = new ManualResetEvent(false);
            ReceiveDataCompleted = new ManualResetEvent(false);
            ReceiveStreamInvoked = new ManualResetEvent(false);
            ReceiveStreamCompleted = new ManualResetEvent(false);
            Name = "ClientReceiver_" + DateTime.Now;
        }

        public void ReceiveData(string data)
        {
            Console.WriteLine("[{0}] ReceiveData received the following:", Name);
            Console.WriteLine(data);
            ReceiveDataInvoked.Set();
            if (data == "LastMessage")
            {
                ReceiveDataCompleted.Set();
            }
        }

        public void ReceiveStream(Stream stream)
        {
            ReceiveStreamInvoked.Set();

            int readResult;
            byte[] buffer = new byte[1000];
            do
            {
                try
                {
                    readResult = stream.Read(buffer, 0, buffer.Length);
                    Console.WriteLine("[{0}] just read {1} bytes.{2}stream.Position = {3}", this.Name, readResult, Environment.NewLine, stream.Position);
                }
                catch (Exception ex)
                {
                    Console.WriteLine("Caught exception when trying to read: {0}", ex);
                    throw;
                }
            }
            while (readResult != 0);

            stream.Close();

            Console.WriteLine("[{0}] ReceiveStream invoked.", Name);
            ReceiveStreamCompleted.Set();
        }

        public void ReceiveLog(List<string> log)
        {
            ServerLog = log;
            LogReceived.Set();
        }
    }
}

步骤 4：生成一个客户端来调用服务。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.ServiceModel;
using System.ServiceModel.Channels;

namespace DuplexStreamingClient
{
    class Program
    {
        static void Main(string[] args)
        {
            RunTests();
        }

        private static void RunTests()
        {
            string address = "net.tcp:///IISHost/DuplexStreamService/DuplexService.svc";
            NetTcpBinding binding = new NetTcpBinding();
            ClientReceiver clientReceiver = new ClientReceiver();

            DuplexChannelFactory<IDuplexService> channelFactory = new DuplexChannelFactory<IDuplexService>(new InstanceContext(clientReceiver), binding, address);
            IDuplexService client = channelFactory.CreateChannel();

            Console.WriteLine(client.DownloadData());

            // Test Duplex Operation
            Console.WriteLine("Invoking StartPushingStream - get the server to push a stream to the client.");
            client.StartPushingStream();

            Console.WriteLine("Waiting on ReceiveStreamInvoked from the ClientReceiver.");
            clientReceiver.ReceiveStreamInvoked.WaitOne();
            clientReceiver.ReceiveStreamInvoked.Reset();

            Console.WriteLine("Invoking StopPushingStream");
            client.StopPushingStream();
            Console.WriteLine("Waiting on ReceiveStreamCompleted from the ClientReceiver.");
            clientReceiver.ReceiveStreamCompleted.WaitOne();
            clientReceiver.ReceiveStreamCompleted.Reset();

            Console.WriteLine("Getting results from server via callback.");
            client.GetLog();
            clientReceiver.LogReceived.WaitOne();

            Console.WriteLine("----Following are the logs from the server:-----");
            foreach (string serverLogItem in clientReceiver.ServerLog)
            {
                Console.WriteLine(serverLogItem);
            }
            Console.WriteLine("---------------- End server log. ---------------");

            ((IChannel)client).Close();

            Console.WriteLine("Test passed.");
        }
    }
}