WCF 双工内的 TPL 数据流块答案

【问题标题】：TPL dataflow blocks inside a WCF duplexWCF 双工内的 TPL 数据流块
【发布时间】：2013-03-21 12:31:09
【问题描述】：

我是 SO 的新作家，请多多包涵。

我有一个带有双工服务合同的 WCF 服务。该服务合同有一个操作联系人，假设进行长时间的数据处理。我被限制将并发数据处理的数量限制为最多 3 个。我的问题是，在数据处理之后，我需要返回到相同的服务实例上下文，所以我回调我的发起程序端点，传递数据处理结果。我需要提一下，由于各种原因，我受限于 TPL 数据流和 WCF 双工。

这是我目前所写内容的演示

在控制台库中我模拟 WCF 调用

class Program
{
    static void Main(string[] args)
    {
        // simulate service calls

        Enumerable.Range(0, 5).ToList().ForEach(x =>
        {
            new System.Threading.Thread(new ThreadStart(async () =>
            {
                var service = new Service();
                await service.Inc(x);
            })).Start();
        });
    }
}

这里是 WCF 服务

// service contract
public class Service
{
    static TransformBlock<Message<int>, Message<int>> transformBlock;

    static Service()
    {
        transformBlock = new TransformBlock<Message<int>, Message<int>>(x => Inc(x), new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 3
        });
    }

    static Message<int> Inc(Message<int> input)
    {
        System.Threading.Thread.Sleep(100);

        return new Message<int> { Token = input.Token, Data = input.Data + 1 };
    }

    // operation contract
    public async Task Inc(int id)
    {
        var token = Guid.NewGuid().ToString();

        transformBlock.Post(new Message<int> { Token = token, Data = id });

        while (await transformBlock.OutputAvailableAsync())
        {
            Message<int> message;
            if (transformBlock.TryReceive(m => m.Token == token, out message))
            {
                // do further processing using initiator service instance members
                // something like Callback.IncResult(m.Data);
                break;
            }
        }
    }
}

public class Message<T>
{
    public string Token { get; set; }

    public T Data { get; set; }
}

操作合约并不是真的需要异步，但我需要 OutputAvailableAsync 通知。

这是一个好方法还是有更好的解决方案适合我的场景？

提前致谢。

【问题讨论】：

“我受限于 TPL 数据流”你是什么意思？你必须使用数据流吗？为什么？这没有多大意义。
我对数据处理的要求是保证并发但限制并行。 TPL 数据流块似乎是一个不错的选择，而且它们是由技术要求而不是 Plinq 或其他东西强加的。
如果这真的是您使用数据流的全部，那么我认为这是一种矫枉过正的做法。您可以使用更简单的代码实现相同的效果（请参阅我的答案）。另外，我也不确定双工服务是否是正确的选择。即使没有，客户端和服务器都可以是异步的。
有人告诉我数据流可能会演变成一个块链，在这种情况下，我应该传播我想的 TCS。信号量看起来很优雅。关于复式我完全同意。谢谢

标签： c# .net wcf task-parallel-library tpl-dataflow

【解决方案1】：

首先，我认为您不应该像现在这样使用令牌。在进程之间进行通信时，唯一标识符很有用。但是，当您在单个进程中时，只需使用引用相等。

要真正回答您的问题，我认为（某种）忙循环不是一个好主意。

一个更简单的异步限制解决方案是使用SemaphoreSlim。比如：

static readonly SemaphoreSlim Semaphore = new SemaphoreSlim(3);

// operation contract
public async Task Inc(int id)
{
    await Semaphore.WaitAsync();

    try
    {
        Thread.Sleep(100);
        var result = id + 1;
        // do further processing using initiator service instance members
        // something like Callback.IncResult(result);
    }
    finally
    {
        Semaphore.Release();
    }
}

如果您真的想（或必须？）使用数据流，您可以使用TaskCompletionSource 在操作和块之间进行同步。该操作方法将等待TaskCompletionSource 的Task，并且该块将在它完成对该消息的计算时设置它：

private static readonly ActionBlock<Message<int>> Block =
    new ActionBlock<Message<int>>(
        x => Inc(x),
        new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 3
        });

static void Inc(Message<int> input)
{
    Thread.Sleep(100);

    input.TCS.SetResult(input.Data + 1);
}

// operation contract
public async Task Inc(int id)
{
    var tcs = new TaskCompletionSource<int>();

    Block.Post(new Message<int> { TCS = tcs, Data = id });

    int result = await tcs.Task;
    // do further processing using initiator service instance members
    // something like Callback.IncResult(result);
}

【讨论】：