标签: task-parallel-library

我正在尝试以编程方式将C#4中的异步操作链接起来,例如写入给定的Stream对象.我最初是"手动"执行此操作,将回调挂钩从一个操作到下一个操作,但我想我会尝试使用.NET 4任务并行库来省去重新发明并发轮的麻烦.

首先,我将异步调用包装在Tasks中,如下所示:

public static Task CreateWriteTask(Stream stream, byte[] data)
{
    return Task.Factory.FromAsync(stream.BeginWrite, stream.EndWrite, data, 0, data.Length, null);
}

Continuations使链接同步操作变得非常容易(如果您原谅不幸的方法名称):

public static Task ChainFlush(Stream stream, Task precedingTask)
{
    return precedingTask.ContinueWith(x => stream.Flush());
}

但是没有任何版本的Task.ContinueWith方法可以像接受同样的方式接受异步操作TaskFactory.FromAsync.

所以,假设我坚持使用TPL,我正在寻找这种方法的正确实现:

public static Task ChainWrite(Stream stream, byte[] data, Task precedingTask)
{
    //?
}

我很开心System.Threading.Tasks.但是,我看到的许多代码示例都是这样的:

Dim lcTask = Task.Factory.StartNew(Sub() DoSomeWork())
Dim lcTaskLong = Task.Factory.StartNew(Sub() DoSomeWork(), TaskCreationOptions.LongRunning)
Task.WaitAll(lcTask, lcTaskLong)

这就是样本的范围.
任务实现IDisposable,显然我应该处理它们,但如果我只是想要"消防和遗忘"怎么办？

如果我不处理,我会泄漏线程/句柄/记忆/业力吗？我使用"错误"的任务吗？(应该只使用代理并单独留下任务吗？)

我可以处理ContinueWith()吗？(这好像在玩俄罗斯轮盘赌.)

我发现命名线程在调试时非常有用.

我看不到使用参数命名线程的方法 Task.Factory.StartNew()

那么在任务中明确命名线程是否可以接受？例如:

private void MyFunc()
{
    Task.Factory.StartNew(() =>
    {
        Thread.CurrentThread.Name = "Foobulizer";
        Foobulize();
    });
}

但是,我知道线程可能会被重用于不同的任务,所以我需要在任务结束时显式重置线程名称吗？这感觉非常hacky所以我认为这可能是一个坏主意,或者有一个正确的方法来做到这一点？

我在前几天读到,对于长期运行的任务,我最好的办法是手动创建线程,而不是使用.NET的线程池或任务并行.当我正在学习c#线程时,我真的希望有人能够启发我,特别是对于长时间运行的IO任务.先感谢您.

这是我注意到的非常奇怪的事情.

我正在编写CRM 2011 Silverlight扩展,而且,在我的本地开发实例上一切正常.该应用程序使用OData进行通信,并使用System.Threading.Tasks.Task很多来在后台执行所有操作(FromAsync是一种祝福).

但是,我决定在CRM 2011 Online中测试我的应用程序,并且令我惊讶地发现它将不再起作用; 结束检索任务时,我会收到安全异常.

使用Fiddler,我发现CRM正试图将我重定向到Live登录页面,考虑到我已经登录,这个页面没有多大意义.

经过多次尝试后,我发现错误是因为我从与UI线程不同的线程访问该服务.

这是一个简单的例子:

    //this will work
    private void button1_Click(object sender, RoutedEventArgs e)
    {
        var query = ctx.AccountSet;
        query.BeginExecute((result) =>
        {
            textBox1.Text = query.EndExecute(result).First().Name;
        }, null);
    }

    //this will fail
    private void button2_Click(object sender, RoutedEventArgs e)
    {
        System.Threading.Tasks.Task.Factory.StartNew(RestAsync);
    }

    void RestAsync()
    {
        var query = ctx.AccountSet;
        var async = query.BeginExecute(null, null);
        var task = System.Threading.Tasks.Task.Factory.FromAsync<Account>(async, (result) =>
        {
            return query.EndExecute(result).First(); // <- Exception thrown here
        });
        textBox1.Dispatcher.BeginInvoke(() =>
        { …

图表1:一些代码将Async(不是async!)网络调用包装成一个Task

public static Task<byte[]> GetAsync(IConnection connection, uint id)
{
    ReadDataJob jobRDO = new ReadDataJob();

    //No overload of FromAsync takes 4 extra parameters, so we have to wrap
    // Begin in a Func so that it looks like it takes no parameters except 
    // callback and state
    Func<AsyncCallback, object, IAsyncResult> wrapped = (callback, state) =>
                jobRDO.Begin(connection, 0, 0, id, callback, state);

    return Task<byte[]>.Factory.FromAsync(wrapped, ar =>
    {
        ErrorCode errorCode;
        UInt32 sError;
        UInt32 attribute;
        byte[] data = new byte[10];
        jobRDO.End(out …

相关简介信息:

AFAIK,并发堆栈,队列和包类在内部使用链接列表实现.
而且我知道争用少得多,因为每个线程都负责自己的链表.无论如何,我的问题是关于ConcurrentDictionary<,>

但我正在测试这段代码:(单线程)

Stopwatch sw = new Stopwatch();
sw.Start();

    var d = new ConcurrentDictionary < int,  int > ();
    for(int i = 0; i < 1000000; i++) d[i] = 123;
    for(int i = 1000000; i < 2000000; i++) d[i] = 123;
    for(int i = 2000000; i < 3000000; i++) d[i] = 123;
    Console.WriteLine("baseline = " + sw.Elapsed);

sw.Restart();

    var d2 = new Dictionary < int, int > ();
    for(int i = 0; i < 1000000; i++)         lock (d2) …

这段代码怎么可能

TaskManager.RunSynchronously<MyObject>(fileMananager.BackupItems, package);

导致编译错误

The call is ambiguous between the following methods or properties:
'TaskManager.RunSynchronously<MyObject>(System.Action<MyObject>, MyObject)' and
'TaskManager.RunSynchronously<MyObject>(System.Func<MyObject, bool>, MyObject)'

当行动的签名是

public void BackupItems(MyObject package)

和"暧昧"的方法是

static class TaskManager
{
    public static void RunSynchronously<TInput>(Action<TInput> task, TInput param)
    {
        Task.Factory.StartNew(() => task(param));
    }

    public static bool RunSynchronously<TInput>(Func<TInput, bool> task, TInput param)
    {
        return Task.Factory.StartNew(() => task(param)).Result;
    }
}

在我看来,这些方法之间存在很大差异.我在这里错过了什么？

编辑:

除了接受的答案,我刚刚遇到了类似问题的解决方案.这是链接.

我的理解是这return Task.FromResult(foo)是一个简单的简写:

var tcs = new TaskCompletionSource<TFoo>();
tcs.SetResult(foo);
return tcs.Task;

是否有一些等效的任务返回异常状态？

var tcs = new TaskCompletionSource<TFoo>();
tcs.SetException(new NotSupportedException()); // or whatever is appropriate
return tcs.Task;

我没有看到任何类似的东西Task.FromException.或者在不返回任务的情况下抛出异常会更合适吗？

我正在尝试为System.Net.WebSocket编写一个扩展方法,它将使用Reactive Extensions(Rx.NET)将其转换为IObserver.你可以看到下面的代码:

public static IObserver<T> ToObserver<T>(this WebSocket webSocket, IWebSocketMessageSerializer<T> webSocketMessageSerializer)
{
    // Wrap the web socket in an interface that's a little easier to manage
    var webSocketMessageStream = new WebSocketMessageStream(webSocket);

    // Create the output stream to the client
    return Observer.Create<T>(
        onNext:      async message => await webSocketMessageStream.WriteMessageAsync(webSocketMessageSerializer.SerializeMessage(message)),
        onError:     async error   => await webSocketMessageStream.CloseAsync(WebSocketCloseStatus.InternalServerError, string.Format("{0}: {1}", error.GetType(), error.Message)),
        onCompleted: async ()      => await webSocketMessageStream.CloseAsync(WebSocketCloseStatus.NormalClosure, "Server disconnected")
    );
}

这段代码有效,但我担心在onNext,onError和onCompleted lambdas中使用async/await.我知道这会返回一个async void lambda,这是不赞成的(有时候会引起我已经遇到的问题).

我一直在阅读Rx.NET文档以及互联网上的博客文章,我似乎无法找到在IObserver中使用async/await方法的正确方法(如果有的话).有没有正确的方法来做到这一点？如果没有,那么我该如何解决这个问题呢？