如何将SynchronizationContext与WCF结合使用

Question

我正在阅读SynchronizationContext，并试图通过尝试将OperationContext流到所有线程(即使是在await调用之后)来确保不会搞砸任何事情。

我有一个SynchronizationContext类：

public class OperationContextSynchronizationContext : SynchronizationContext
{

    // Track the context to make sure that it flows through to the next thread.

    private readonly OperationContext _context;

    public OperationContextSynchronizationContext(OperationContext context)
    {
        _context = context;
    }

    public override void Post(SendOrPostCallback d, object state)
    {
        OperationContext.Current = _context;
        d(state);
    }
}

然后，在每个方法调用(使用Ninject IInterceptor)时都这样调用：

var original = SynchronizationContext.Current;
try
{
    // Make sure that the OperationContext flows across to the other threads,
    // since we need it for ContextStack.  (And also it's cool to have it.)
    SynchronizationContext.SetSynchronizationContext(new OperationContextSynchronizationContext(OperationContext.Current));

    // Process the method being called.
    invocation.Proceed();
}
finally
{
    SynchronizationContext.SetSynchronizationContext(original);
}

它看起来很有效(我可以根据需要使用OperationContext )，但这是正确的方法吗？我错过了什么可能会咬我的重要东西吗？

EDITed带着Stephen的一些评论：

public class OperationContextSynchronizationContext : SynchronizationContext, IDisposable
{

    // Track the context to make sure that it flows through to the next thread.

    private readonly OperationContext _context;
    private readonly SynchronizationContext _previous;

    public OperationContextSynchronizationContext(OperationContext context)
    {
        _context = context;
        _previous = SynchronizationContext.Current;
        SynchronizationContext.SetSynchronizationContext(this);
    }

    public override void Post(SendOrPostCallback d, object state)
    {
        OperationContext.Current = _context;
        d(state);
        //(_previous ?? new SynchronizationContext()).Post(d, state);
    }

    private bool _disposed = false;
    public void Dispose()
    {
        if (!_disposed)
        {
            SynchronizationContext.SetSynchronizationContext(_previous);
            _disposed = true;
        }
    }
}

最终

public class OperationContextSynchronizationContext : SynchronizationContext, IDisposable
{

    // Track the operation context to make sure that it flows through to the next call context.

    private readonly OperationContext _context;
    private readonly SynchronizationContext _previous;

    public OperationContextSynchronizationContext()
    {
        _context = OperationContext.Current;
        _previous = SynchronizationContext.Current;
        SynchronizationContext.SetSynchronizationContext(this);
    }

    public override void Post(SendOrPostCallback d, object state)
    {
        var context = _previous ?? new SynchronizationContext();
        context.Post(
            s =>
            {
                OperationContext.Current = _context;
                try
                {
                    d(s);
                }
                catch (Exception ex)
                {
                    // If we didn't have this, async void would be bad news bears.
                    // Since async void is "fire and forget," they happen separate
                    // from the main call stack.  We're logging this separately so
                    // that they don't affect the main call (and it just makes sense).

                    // log here
                }
            },
            state
        );
    }

    private bool _disposed = false;
    public void Dispose()
    {
        if (!_disposed)
        {
            // Return to the previous context.
            SynchronizationContext.SetSynchronizationContext(_previous);
            _disposed = true;
        }
    }
}

Answer 1

有几件事我觉得很突出。

首先，我不能建议为此使用SynchronizationContext。您正在用框架解决方案来解决应用程序问题。它会起作用的；我只是从架构的角度发现这是有问题的。不过，唯一的选择不是那么干净:也许最合适的方法是为Task编写一个扩展方法，该方法返回保存OperationContext的自定义侍者。

其次，OperationContextSynchronizationContext.Post的实现直接执行委托。这有几个问题:首先，应该异步执行委托(我怀疑.NET框架或TPL中有几个地方假定了这一点)。另一方面，这个SynchronizationContext有一个特定的实现；在我看来，如果自定义SyncCtx封装了一个现有的实现，效果会更好。一些SyncCtx有特定的线程需求，现在OperationContextSynchronizationContext正在替代这些线程，而不是作为补充。

第三，自定义SyncCtx在调用其委托时不会将自己设置为当前的SyncCtx。因此，如果在同一方法中有两个await，则不会工作。

Answer 2

注意:在假设这是正确的解决方案之前，请阅读Stephen Cleary's answer。在我的特定用例中，除了在Framework级别解决这个问题之外，我别无选择。

所以，为了把我的实现加入其中.我需要在等待关键字之后修复OperationContext.Current和Thread.CurrentUICulture流到线程，我发现有几种情况下，您的解决方案不能正常工作(TDD for the win!)。

这是一个新的SynchronisationContext，它将促进某些自定义状态的捕获和恢复：

public class CustomFlowingSynchronizationContext : SynchronizationContext
{
    private readonly SynchronizationContext _previous;
    private readonly ICustomContextFlowHandler _customContextFlowHandler;

    public CustomFlowingSynchronizationContext(ICustomContextFlowHandler customContextFlowHandler, SynchronizationContext synchronizationContext = null)
    {
        this._previous = synchronizationContext ?? SynchronizationContext.Current;
        this._customContextFlowHandler = customContextFlowHandler;
    }

    public override void Send(SendOrPostCallback d, object state)
    {
        var callback = this.CreateWrappedSendOrPostCallback(d);
        if (this._previous != null) this._previous.Send(callback, state);
        else base.Send(callback, state);
    }

    public override void OperationStarted()
    {
        this._customContextFlowHandler.Capture();
        if (this._previous != null) this._previous.OperationStarted();
        else base.OperationStarted();
    }

    public override void OperationCompleted()
    {
        if (this._previous != null) this._previous.OperationCompleted();
        else base.OperationCompleted();
    }

    public override int Wait(IntPtr[] waitHandles, bool waitAll, int millisecondsTimeout)
    {
        if (this._previous != null) return this._previous.Wait(waitHandles, waitAll, millisecondsTimeout);
        return base.Wait(waitHandles, waitAll, millisecondsTimeout);
    }

    public override void Post(SendOrPostCallback d, object state)
    {
        var callback = this.CreateWrappedSendOrPostCallback(d);
        if (this._previous != null) this._previous.Post( callback, state);
        else base.Post( callback, state);
    }
    private SendOrPostCallback CreateWrappedSendOrPostCallback(SendOrPostCallback d)
    {
        return s =>
        {
            var previousSyncCtx = SynchronizationContext.Current;
            var previousContext = this._customContextFlowHandler.CreateNewCapturedContext();
            SynchronizationContext.SetSynchronizationContext(this);
            this._customContextFlowHandler.Restore();
            try
            {
                d(s);
            }
            catch (Exception ex)
            {
                // If we didn't have this, async void would be bad news bears.
                // Since async void is "fire and forget", they happen separate
                // from the main call stack.  We're logging this separately so
                // that they don't affect the main call (and it just makes sense).
            }
            finally
            {
                this._customContextFlowHandler.Capture();
                // Let's get this thread back to where it was before
                SynchronizationContext.SetSynchronizationContext(previousSyncCtx);
                previousContext.Restore();
            }
        };
    }

    public override SynchronizationContext CreateCopy()
    {
        var synchronizationContext = this._previous != null ? this._previous.CreateCopy() : null;
        return new CustomFlowingSynchronizationContext(this._customContextFlowHandler, synchronizationContext);
    }

    public override string ToString()
    {
        return string.Format("{0}({1})->{2}", base.ToString(), this._customContextFlowHandler, this._previous);
    }
}

ICustomContextFlowHandler接口如下所示：

public interface ICustomContextFlowHandler
{
    void Capture();
    void Restore();
    ICustomContextFlowHandler CreateNewCapturedContext();
}

WCF中用于我的用例的这个ICustomContextFlowHandler的实现如下：

public class WcfContextFlowHandler : ICustomContextFlowHandler
{
    private CultureInfo _currentCulture;
    private CultureInfo _currentUiCulture;
    private OperationContext _operationContext;

    public WcfContextFlowHandler()
    {
        this.Capture();
    }

    public void Capture()
    {
        this._operationContext = OperationContext.Current;
        this._currentCulture = Thread.CurrentThread.CurrentCulture;
        this._currentUiCulture = Thread.CurrentThread.CurrentUICulture;
    }

    public void Restore()
    {
        Thread.CurrentThread.CurrentUICulture = this._currentUiCulture;
        Thread.CurrentThread.CurrentCulture = this._currentCulture;
        OperationContext.Current = this._operationContext;
    }

    public ICustomContextFlowHandler CreateNewCapturedContext()
    {
        return new WcfContextFlowHandler();
    }
}

这是WCF行为(所有这些都捆绑在一起以使事情变得更简单)，您需要添加到您的配置中，以连接新的SynchronisationContext：(在AfterReceiveRequest方法中发生了奇迹)

    public class WcfSynchronisationContextBehavior : BehaviorExtensionElement, IServiceBehavior, IDispatchMessageInspector
{
    #region Implementation of IServiceBehavior

    /// <summary>
    /// Provides the ability to change run-time property values or insert custom extension objects such as error handlers, message or parameter interceptors, security extensions, and other custom extension objects.
    /// </summary>
    /// <param name="serviceDescription">The service description.</param><param name="serviceHostBase">The host that is currently being built.</param>
    public void ApplyDispatchBehavior(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase)
    {
        foreach (ChannelDispatcher channelDispatcher in serviceHostBase.ChannelDispatchers)
        {
            foreach (EndpointDispatcher endpointDispatcher in channelDispatcher.Endpoints)
            {
                if (IsValidContractForBehavior(endpointDispatcher.ContractName))
                {
                    endpointDispatcher.DispatchRuntime.MessageInspectors.Add(this);
                }
            }
        }
    }

    /// <summary>
    /// Provides the ability to inspect the service host and the service description to confirm that the service can run successfully.
    /// </summary>
    /// <param name="serviceDescription">The service description.</param><param name="serviceHostBase">The service host that is currently being constructed.</param>
    public void Validate(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase)
    {
        // No implementation
    }

    /// <summary>
    /// Provides the ability to pass custom data to binding elements to support the contract implementation.
    /// </summary>
    /// <param name="serviceDescription">The service description of the service.</param>
    /// <param name="serviceHostBase">The host of the service.</param><param name="endpoints">The service endpoints.</param>
    /// <param name="bindingParameters">Custom objects to which binding elements have access.</param>
    public void AddBindingParameters(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase,
        Collection<ServiceEndpoint> endpoints, BindingParameterCollection bindingParameters)
    {
        // No implementation
    }


    #endregion

    #region Implementation of IDispatchMessageInspector

    /// <summary>
    /// Called after an inbound message has been received but before the message is dispatched to the intended operation.
    /// </summary>
    /// <returns>
    /// The object used to correlate state. This object is passed back in the <see cref="M:System.ServiceModel.Dispatcher.IDispatchMessageInspector.BeforeSendReply(System.ServiceModel.Channels.Message@,System.Object)"/> method.
    /// </returns>
    /// <param name="request">The request message.</param><param name="channel">The incoming channel.</param><param name="instanceContext">The current service instance.</param>
    public object AfterReceiveRequest(ref Message request, IClientChannel channel, InstanceContext instanceContext)
    {
        var customContextFlowHandler = new WcfContextFlowHandler();
        customContextFlowHandler.Capture();
        var synchronizationContext = new CustomFlowingSynchronizationContext(customContextFlowHandler);
        SynchronizationContext.SetSynchronizationContext(synchronizationContext);
        return null;
    }

    /// <summary>
    /// Called after the operation has returned but before the reply message is sent.
    /// </summary>
    /// <param name="reply">The reply message. This value is null if the operation is one way.</param><param name="correlationState">The correlation object returned from the <see cref="M:System.ServiceModel.Dispatcher.IDispatchMessageInspector.AfterReceiveRequest(System.ServiceModel.Channels.Message@,System.ServiceModel.IClientChannel,System.ServiceModel.InstanceContext)"/> method.</param>
    public void BeforeSendReply(ref Message reply, object correlationState)
    {
        // No implementation
    }

    #endregion

    #region Helpers

    /// <summary>
    /// Filters out metadata contracts.
    /// </summary>
    /// <param name="contractName">The contract name to validate.</param>
    /// <returns>true if not a metadata contract, false otherwise</returns>
    private static bool IsValidContractForBehavior(string contractName)
    {
        return !(contractName.Equals("IMetadataExchange") || contractName.Equals("IHttpGetHelpPageAndMetadataContract"));
    }

    #endregion Helpers

    #region Overrides of BehaviorExtensionElement

    /// <summary>
    /// Creates a behavior extension based on the current configuration settings.
    /// </summary>
    /// <returns>
    /// The behavior extension.
    /// </returns>
    protected override object CreateBehavior()
    {
        return new WcfSynchronisationContextBehavior();
    }

    /// <summary>
    /// Gets the type of behavior.
    /// </summary>
    /// <returns>
    /// A <see cref="T:System.Type"/>.
    /// </returns>
    public override Type BehaviorType
    {
        get { return typeof(WcfSynchronisationContextBehavior); }
    }

    #endregion
}