c# - WCF 双工 : Send callback to a specific subscribed ASP. NET Webforms 客户端

Question

我正在开发连接到 WCF 4.0 双工服务以实现某种平面文件处理的 ASP.NET 4.0 Webforms 客户端应用程序。当用户在 Page_Load 事件上进入页面时，我为客户端订阅了双工服务，这是因为在某些情况下我需要通知所有客户端:

A) 启动进程的客户端必须在进程启动时得到通知。

B) 处理文件时必须通知启动进程的客户端。

C) 启动进程的客户必须在整个进程完成时得到通知。

D) 如果新客户(订阅者)在流程已经启动时进入，则必须收到特定通知。

E) 如果有多个客户端(订阅者)在其中一个启动进程时处于事件状态，则其他客户端必须收到特定通知。

我已经编写了这个逻辑，但是我在尝试完成特定订阅者通知时遇到了很多问题，WCF 似乎将 Web 应用程序的所有客户端/实例都识别为相同，我收到了所有启动进程的客户端中的通知，如果我打开其他浏览器并启动新 session (在 ASP.NET 上)，我会收到相同的通知，没有什么特别的。

在这里你可以看到我的代码的简化版本

WCF 服务接口(interface)

using System.ServiceModel;

namespace WcfService
{
    [ServiceContract(CallbackContract = typeof(IService1DuplexCallback))]
    public interface IService1
    {
        [OperationContract(IsOneWay = true)]
        void Subscribe(string idSesion);

        [OperationContract(IsOneWay = true)]
        void ProcessFiles(string idSesion);
    }

    public interface IService1DuplexCallback
    {
        [OperationContract(IsOneWay = true)]
        void NotifyProcessWorking();

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

        [OperationContract(IsOneWay = true)]
        void NotifyFileProcessed(int id);

        [OperationContract(IsOneWay = true)]
        void NotifyProcessFinished();
    }
}

WCF 服务实现

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

namespace WcfService
{
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Multiple)]
    public class Service1 : IService1
    {
        private static List<KeyValuePair<string, IService1DuplexCallback>> _clients = new List<KeyValuePair<string, IService1DuplexCallback>>();
        private static bool _isProcessStarted;
        private static string _sessionStarted = string.Empty;

        public void Subscribe(string idSesion)
        {
            lock (_clients)
            {
                if (!_clients.Any(x => string.Equals(x.Key, idSesion, StringComparison.InvariantCultureIgnoreCase)))
                {
                    var callback = OperationContext.Current.GetCallbackChannel<IService1DuplexCallback>();

                    if (callback != null)
                    {
                        var currentSubscriber = new KeyValuePair<string, IService1DuplexCallback>(idSesion, callback);
                        _clients.Add(currentSubscriber);
                    }
                }
            }

            if (_isProcessStarted)
            {
                NotifyProcessWorking(idSesion);
            }
        }

        public void ProcessFiles(string idSesion)
        {
            _isProcessStarted = true;
            _sessionStarted = idSesion;

            try
            {
                var mockFileCount = 23;
                var r = new Random();

                NotifyStarted();
                NotifyProcessWorking();

                Parallel.For(0, mockFileCount, (i) =>
                {
                    //Do a lot of specific validations... (time betweeen 5 secs and 2 minutes per file)
                    var time = r.Next(5000, 120000);

                    Thread.Sleep(time);

                    NotifyFileProcessed(i);
                });

                NotifyProcessFinished();
            }
            catch (Exception ex)
            {
                throw;
            }

            _isProcessStarted = false;
        }

        private static void NotifyStarted()
        {
            var c = _clients.FirstOrDefault(x => string.Equals(x.Key, _sessionStarted, StringComparison.InvariantCultureIgnoreCase));

            try
            {
                c.Value.NotifyProcessStarted();
            }
            catch (Exception ex)
            {
                lock (_clients)
                {
                    _clients.Remove(c);
                }
            }
        }

        private static void NotifyFileProcessed(int idFile)
        {
            var c = _clients.FirstOrDefault(x => string.Equals(x.Key, _sessionStarted, StringComparison.InvariantCultureIgnoreCase));

            try
            {
                c.Value.NotifyFileProcessed(idFile);
            }
            catch (Exception ex)
            {
                lock (_clients)
                {
                    _clients.Remove(c);
                }
            }
        }

        private static void NotifyProcessFinished()
        {
            foreach (var c in _clients)
            {
                try
                {
                    c.Value.NotifyProcessFinished();
                }
                catch (Exception ex)
                {
                    lock (_clients)
                    {
                        _clients.Remove(c);
                    }
                }
            }
        }

        private static void NotifyProcessWorking(string idSesion = "")
        {
            if (string.IsNullOrEmpty(idSesion))
            {
                foreach (var c in _clients)
                {
                    try
                    {
                        c.Value.NotifyProcessWorking();
                    }
                    catch (Exception ex)
                    {
                        lock (_clients)
                        {
                            _clients.Remove(c);
                        }
                    }
                }
            }
            else
            {
                var c = _clients.FirstOrDefault(x => string.Equals(x.Key, idSesion, StringComparison.InvariantCultureIgnoreCase));

                try
                {
                    c.Value.NotifyProcessWorking();
                }
                catch (Exception)
                {
                    lock (_clients)
                    {
                        _clients.Remove(c);
                    }
                }
            }
        }
    }
}

WCF 服务 Web.config

<?xml version="1.0"?>
<configuration>
  <appSettings/>
  <system.web>
    <compilation debug="true" targetFramework="4.0"/>
    <httpRuntime/>
  </system.web>
  <system.serviceModel>
    <services>
      <service name="WcfService.Service1">
        <endpoint address="" binding="wsDualHttpBinding" bindingConfiguration="FileProcessorDuplexBinding" 
                  name="FileProcessorDuplexEndPoint" contract="WcfService.IService1"/>
      </service>
    </services>
    <bindings>
      <wsDualHttpBinding>
        <binding name="FileProcessorDuplexBinding" closeTimeout="00:30:00" openTimeout="00:30:00" 
                 sendTimeout="00:30:00" receiveTimeout="00:30:00" maxBufferPoolSize="2147483647"
                 maxReceivedMessageSize="2147483647">
          <reliableSession inactivityTimeout="00:30:00"/>
          <readerQuotas maxDepth="2147483647" maxStringContentLength="2147483647" maxArrayLength="2147483647" 
                        maxBytesPerRead="2147483647" maxNameTableCharCount="2147483647"/>
          <security mode="None"/>
        </binding>
      </wsDualHttpBinding>
    </bindings>
    <behaviors>
      <serviceBehaviors>
        <behavior>
          <serviceMetadata httpGetEnabled="true" httpsGetEnabled="true"/>
          <serviceDebug includeExceptionDetailInFaults="false"/>
        </behavior>
      </serviceBehaviors>
    </behaviors>
    <serviceHostingEnvironment  multipleSiteBindingsEnabled="true"/>
  </system.serviceModel>
  <system.webServer>
    <modules runAllManagedModulesForAllRequests="true"/>
    <directoryBrowse enabled="true"/>
  </system.webServer>
</configuration>

ASP.NET WebForm 客户端用户界面

<%@ Page Language="C#" AutoEventWireup="true" CodeBehind="Default.aspx.cs" Inherits="WebApplication.Default" %>

<!DOCTYPE html>

<html xmlns="http://www.w3.org/1999/xhtml">
<head runat="server">
    <script src="Scripts/jquery-2.1.4.min.js"></script>
    <title></title>
</head>
<body>
    <form id="form1" runat="server">
        <div>
            <asp:Button ID="btnStart" runat="server" Text="Start Process" OnClientClick="Start();"/>
            <br/>
            <br/>
            <asp:Label ID="lblStatus" runat="server" Text="[Process Status]"></asp:Label>
        </div>
        <script>

            function Start() {

                var loc = window.location.href;
                var dataValue = "{}";

                $.ajax({
                    type: "POST",
                    url: loc + "/StartProcess",
                    contentType: 'application/json',
                    data: dataValue,
                    dataType: 'json',
                    error: function(XMLHttpRequest, textStatus, errorThrown) {
                        alert("Request: " + XMLHttpRequest.toString() + "\n\nStatus: " + textStatus + "\n\nError: " + errorThrown);
                    },
                    success: function(result) {
                    }
                });

            }

            setInterval(function () {

                var loc = window.location.href;
                var dataValue = "{ id: '1' }";

                $.ajax({
                    type: "POST",
                    url: loc + "/CheckMessage",
                    contentType: 'application/json',
                    data: dataValue,
                    dataType: 'json',
                    error: function(XMLHttpRequest, textStatus, errorThrown) {
                        alert("Request: " + XMLHttpRequest.toString() + "\n\nStatus: " + textStatus + "\n\nError: " + errorThrown);
                    },
                    success: function(result) {
                        processMessage(result.d);
                    }
                });


            }, 1000);

            function processMessage(msg) {

                if (msg) {
                    switch (msg) {
                    case "working":
                        alert("Process currently working");
                        $('[id$=lblStatus]').attr('disabled', true);
                        break;

                    case "started":
                        $('#<%=lblStatus.ClientID%>').html("Process started");
                        break;

                    case "finished":
                        $('#<%=lblStatus.ClientID%>').html("Process finished");
                        break;

                    default:
                        var data = msg.split(":");
                        $('#<%=lblStatus.ClientID%>').html("File Processed: " + data[1]);
                        break;
                    }
                }
            }
        </script>
    </form>
</body>
</html>

ASP.NET WebForm 客户端代码隐藏

using System;
using System.Collections.Concurrent;
using System.ServiceModel;
using System.Web.Services;
using System.Web.UI;
using WebApplication.ServiceReference1;

namespace WebApplication
{
    [CallbackBehavior(ConcurrencyMode = ConcurrencyMode.Multiple, UseSynchronizationContext = false)]
    public partial class Default : Page, IService1Callback
    {
        private static ConcurrentQueue<string> _serviceReceivedMessages = new ConcurrentQueue<string>();
        private static string _sessionId = string.Empty;

        protected void Page_Load(object sender, EventArgs e)
        {
            if (!IsPostBack)
            {
                _sessionId = Session.SessionID;

                var proxyDuplex = new Service1Client(new InstanceContext(new Default()));
                proxyDuplex.Subscribe(_sessionId);
            }
        }

        [WebMethod]
        public static void StartProcess()
        {
            var proxyDuplex = new Service1Client(new InstanceContext(new Default()));
            proxyDuplex.ProcessFiles(_sessionId);
        }

        [WebMethod]
        public static string CheckMessage(string id)
        {
            var message = string.Empty;

            _serviceReceivedMessages.TryDequeue(out message);

            return message ?? (message = string.Empty);
        }

        public void NotifyProcessWorking()
        {
            _serviceReceivedMessages.Enqueue("working");
        }

        public void NotifyProcessStarted()
        {
            _serviceReceivedMessages.Enqueue("started");
        }

        public void NotifyFileProcessed(int id)
        {
            _serviceReceivedMessages.Enqueue("processed:"+id);
        }

        public void NotifyProcessFinished()
        {
            _serviceReceivedMessages.Enqueue("finished");
        }
    }
}

ASP.NET WebForm 客户端 Web.config

<?xml version="1.0"?>
<configuration>
  <system.web>
    <compilation debug="true" targetFramework="4.0"/>
    <httpRuntime/>
  </system.web>
  <system.serviceModel>
    <bindings>
      <wsDualHttpBinding>
        <binding name="FileProcessorDuplexBinding" 
                 closeTimeout="00:30:00" openTimeout="00:30:00" receiveTimeout="00:30:00"
                 sendTimeout="00:30:00" maxBufferPoolSize="2147483647" maxReceivedMessageSize="2147483647"
                 clientBaseAddress="http://localhost:62778/TempUri">
          <reliableSession inactivityTimeout="00:30:00" />
          <readerQuotas maxDepth="2147483647" maxStringContentLength="2147483647"
            maxArrayLength="2147483647" maxBytesPerRead="2147483647" maxNameTableCharCount="2147483647" />
          <security mode="None" />
        </binding>
      </wsDualHttpBinding>
    </bindings>
    <client>
      <endpoint address="http://localhost:62778/Service1.svc" binding="wsDualHttpBinding"
        bindingConfiguration="FileProcessorDuplexBinding" contract="ServiceReference1.IService1"
        name="FileProcessorDuplexEndPoint" />
    </client>
  </system.serviceModel>
</configuration>

在这里你可以download包含完整代码的 Visual Studio 2015 解决方案。

我想知道我的代码有什么问题，我认为这种行为是可能的，但不明白为什么 WCF 没有通知特定的客户端。

谢谢

更新 1

我做了@JuanK 建议我(当时)的所有更改，但没有成功，行为继续相同，我添加了一个新的控制台项目来测试相同的服务，并且在该项目中工作正常

但在 ASP.NET 项目中错误继续存在，第二个客户端收到所有通知

在这里你可以download VS 解决方案已更新(此时)

Answer 1

WCF 不会通知特定的客户端，因为您已编码该服务必须向每个人发送通知。

private static void NotifyProcessFinished()
{
    //FOR EACH CLIENT
    foreach (var c in _clients)
    {
        try
        {
            c.Value.NotifyProcessFinished();
        }
        catch (Exception ex)
        {
            lock (_clients)
            {
                _clients.Remove(c);
            }
        }
    }
}

另一方面，有一些ServiceBehavior和我已在此处修复的静态字段问题:

[ServiceBehavior(InstanceContextMode = InstanceContextMode.PerSession, ConcurrencyMode = ConcurrencyMode.Multiple)]
public class Service1 : IService1
{
    private static List<KeyValuePair<string, IService1DuplexCallback>> _clients = new List<KeyValuePair<string, IService1DuplexCallback>>();
    private static bool _isProcessStarted;
    private string _sessionStarted = string.Empty;

    public void Subscribe(string idSesion)
    {
        lock (_clients)
        {
            if (!_clients.Any(x => string.Equals(x.Key, idSesion, StringComparison.InvariantCultureIgnoreCase)))
            {
                var callback = OperationContext.Current.GetCallbackChannel<IService1DuplexCallback>();

                if (callback != null)
                {
                    var currentSubscriber = new KeyValuePair<string, IService1DuplexCallback>(idSesion, callback);
                    _clients.Add(currentSubscriber);
                }
            }
        }

        if (_isProcessStarted)
        {
            NotifyProcessWorking(idSesion);
        }
    }

    public void ProcessFiles(string idSesion)
    {
        _isProcessStarted = true;
        _sessionStarted = idSesion;

        try
        {
            var mockFileCount = 2;
            var r = new Random();

            NotifyStarted();
            NotifyProcessWorking();

            Parallel.For(0, mockFileCount, (i) =>
            {
                //Do a lot of specific validations... (time betweeen 5 secs and 2 minutes per file)
                var time = 5000;//r.Next(5000, 120000);

                Thread.Sleep(time);

                NotifyFileProcessed(i);
            });

            NotifyProcessFinished();
        }
        catch (Exception ex)
        {
            throw;
        }

        _isProcessStarted = false;
    }

    private void NotifyStarted()
    {
        var c = _clients.FirstOrDefault(x => string.Equals(x.Key, _sessionStarted, StringComparison.InvariantCultureIgnoreCase));

        try
        {
            c.Value.NotifyProcessStarted();
        }
        catch (Exception ex)
        {
            lock (_clients)
            {
                _clients.Remove(c);
            }
        }
    }

    private void NotifyFileProcessed(int idFile)
    {
        var c = _clients.FirstOrDefault(
            x => string.Equals(x.Key, _sessionStarted,
            StringComparison.InvariantCultureIgnoreCase)
            );

        try
        {
            c.Value.NotifyFileProcessed(idFile);
        }
        catch (Exception ex)
        {
            lock (_clients)
            {
                _clients.Remove(c);
            }
        }
    }

    private void NotifyProcessFinished()
    {
        //STILL SAME YOU HAVE IT. JUST IN CASE
        foreach (var c in _clients)
        {
            try
            {
                c.Value.NotifyProcessFinished();
            }
            catch (Exception ex)
            {
                lock (_clients)
                {
                    _clients.Remove(c);
                }
            }
        }
    }

    private static void NotifyProcessWorking(string idSesion = "")
    {
        if (string.IsNullOrEmpty(idSesion))
        {
            foreach (var c in _clients)
            {
                try
                {
                    c.Value.NotifyProcessWorking();
                }
                catch (Exception ex)
                {
                    lock (_clients)
                    {
                        _clients.Remove(c);
                    }
                }
            }
        }
        else
        {
            var c = _clients.FirstOrDefault(x => string.Equals(x.Key, idSesion, StringComparison.InvariantCultureIgnoreCase));

            try
            {
                c.Value.NotifyProcessWorking();
            }
            catch (Exception)
            {
                lock (_clients)
                {
                    _clients.Remove(c);
                }
            }
        }
    }
}

如果你有一个单一的实例行为和一个静态字段:静态字段在所有连接之间共享，所以例如 _sessionStarted始终获取上次连接的状态。

所以我将服务行为更改为 PerSession允许每个特定 session /连接保留非静态字段，而像 List<KeyValuePair<string, IService1DuplexCallback>> _clients 这样的静态字段仍然在他们之间共享。这也意味着一些方法现在是非静态方法。