相关疑难解决方法(0)

在玩F#代理后,我尝试使用它们来减少地图.

我使用的基本结构是:

• 地图主管,它将在其状态下完成的所有工作排队,并接收来自地图工作人员的工作请求
• reduce supervisor与map supervisor做同样的事情来减少工作量
• 一堆地图并减少映射和减少的工人,如果一个工作失败,它会将其发送回相应的主管进行重新处理.

我想知道的问题是:

• 与使用PSeq的更传统(但非常好)的地图减少(http://tomasp.net/blog/fsharp-parallel-aggregate.aspx)相比,这有什么意义吗？
• 我实施地图和减少工人的方式看起来很难看是否有更好的方法？
• 看来我可以创建一个1000万的地图工人和1000万减少工人lol,我应该如何选择这些数字,越多越好？

非常感谢,

type Agent<'T> = MailboxProcessor<'T>

//This is the response the supervisor
//gives to the worker request for work
type 'work SupervisorResponse =
| Work of 'work //a piece of work
| NoWork//no work left to do 

//This is the message to the supervisor
type 'work WorkMsg = 
| ToDo of 'work //piles up work in the Supervisor queue
| WorkReq of   AsyncReplyChannel<SupervisorResponse<'work>> //' 

//The supervisor agent can be interacted …

我目前正在优化现有的，非常慢的和超时的生产应用程序。没有选择来重写它。

简而言之，这是一个WCF服务，当前依次调用其他四个“工作者” WCF服务。任何一个工人服务都不依赖于另一个的结果。因此，我们希望一次全部调用它们（而不是顺序调用）。我要重申的是，我们没有重写的奢望。

优化涉及使其立即调用所有工作者服务。这是想到异步的地方。

我在异步编程方面的经验有限，但是对于我的解决方案，我已经就该主题进行了尽可能多的阅读。

问题是，在测试中，它可以工作，但使我的CPU耗尽。谢谢您的帮助

以下是主要WCF服务中基本代码的简化版本

// The service operation belonging to main WCF Service
public void ProcessAllPendingWork()
{
    var workerTasks = new List<Task<bool>>();
    foreach(var workerService in _workerServices)
    {
        //DoWorkAsync is the worker method with the following signature:
        // Task<bool> DoWorkAsync()

        var workerTask = workerService.DoWorkAsync()
        workerTasks.Add(workerTask);
    }

    var task = Task.Run(async ()=>
    {
        await RunWorkerTasks(workerTasks);
    });
    task.Wait();


}

private async RunWorkerTasks(IEnumerable<Tast<bool>> workerTasks)
{
    using(var semaphore = new SemaphoreSlim(initialCount:3))
    {

        foreach (var workerTask in workerTasks)
        { …

如果您需要并行运行多个异步I/O任务但需要确保不会同时运行多个XI/O进程,该怎么办？I/O处理前后的任务不应该有这样的限制.

这是一个场景 - 假设有1000个任务; 每个人都接受一个文本字符串作为输入参数; 转换该文本(预I/O处理)然后将转换后的文本写入文件.目标是使预处理逻辑利用100%的CPU /内核和任务的I/O部分以最大10度并行性运行(同时打开最多10个用于一次写入文件).

你能提供一个示例代码如何使用C#/ .NET 4.5吗？

http://blogs.msdn.com/b/csharpfaq/archive/2012/01/23/using-async-for-file-access-alan-berman.aspx

如果可能，我想为并行启动的任务创建一个异步枚举器。所以第一个完成的是枚举的第一个元素，第二个完成的是枚举的第二个元素，依此类推。

public static async IAsyncEnumerable<T> ParallelEnumerateAsync(this IEnumerable<Task<T>> coldAsyncTasks)
{
    // ... 
}

我打赌有一种使用ContinueWith和 a 的方法Queue<T>，但我并不完全相信自己会实现它。

我目前正以10秒的间隔使用System.Threading.Timer.每次计时器触发时,我都会添加一小段代码写入文件,而且大多数情况下它会按时触发,有时候(大概是应用程序的其余部分是买的),无法触发30秒或40秒,并连续快速地反复射击.

我可以在.NET 3.5中使用更可靠的计时器吗？

计时器设置如下

Timer someTimer = new Timer(new TimerCallback(SomeMethod), null, 0, 10000);

...回调是:

private static void SomeMethod(object state)

但是,提供比这更多的代码是很困难的,因为Timer通常会正确触发.当它嵌入一个大型应用程序(约100,000行左右)时,多个线程被射击,左,右和中心,你慢慢开始看到计时器间歇性地发射.我看过几篇帖子暗示ThreadPool可能已经用尽了,所以我现在正在寻找这个可能是我正在经历的内容.

我有一个IEnumerable<Task<T>>whereT代表某个事件（事件的自然语言类型，而不是event事件的类型）。

我想异步处理这些，因为它们是 IO 绑定的，并限制并发量，因为处理事件的数据库无法处理超过少数（例如 6）个并发处理请求（它们相当重）什么是这样做的正确策略是什么？

如果我有

private Task processeventasync(T someevent) {
  ...
}

foreach(t in tasks) {
  await processeventsasync(await t)
}

我没有并发。

如果我用信号量保护事物，我实际上是在保护线程并用锁保护它们，而不是异步等待它们。

https://msdn.microsoft.com/en-us/library/system.threading.tasks.taskscheduler(v=vs.110).aspxLimitedConcurrencyLevelTaskScheduler上的示例也是基于线程/锁的方法

我考虑过维护一个最多 6 个任务的队列，并WhenAny围绕它进行循环，但这感觉就像重新发明方轮。

private List<Task> running = new List<Task>();

foreach(Task<T> task in tasks) {
  var inner = TaskExtensions.Unwrap(t.ContinueWith(tt => processeventasync(tt.Result)));
  running.Add(inner);
  if (running.Count >= 6) {
    var resulttask = await Task.WhenAny(running); 
    running.Remove(resulttask);
    await resulttask;
    //not sure if this await will schedule the next iteration
    //of the …

当Async/Await发出http请求时(HttpClient例如使用),默认情况下是否内置了任何限制？

这样的问题这意味着连接无限数量的将被打开.我的应用程序在一个循环中执行一批h​​ttp请求,似乎保持在大约50个TCP连接上限.

我最初担心我需要添加SemaphoreSlim限制,但.NET似乎为我做这个.任何人都可以对此有所了解吗？

我想知道如果要完成的任务数量很大,我们是否应该限制异步任务.假设您有1000个URL,您是否一次触发所有请求并等待所有请求:

var tasks = urlList.Select(url => downloadAsync(url));
await Task.WhenAll(tasks);

或者您批量处理请求并逐个处理:

foreach (var urlBatch in urlList.BatchEnumerable(BatchSize)){
    var tasks = urlBatch.Select(url => downloadAsync(url));
    await Task.WhenAll(tasks);
}

我认为批处理是没有必要的,因为第一种方法(立即触发所有请求)将创建由其安排的任务ThreadPool,因此我们应该ThreadPool决定何时执行每项任务.但是,有人告诉我,实际上只有在任务是计算任务时才有效.当任务涉及网络请求时,第一种方法可能导致主机挂起??? 这是为什么 ？

我有一个 Parallel.ForEach 循环，它循环遍历一个集合。在内部，我进行了多次网络 I/O 调用的循环。我使用了 Task.ContinueWith 并嵌套了后续的 async-await 调用。处理的顺序无关紧要，但每个异步调用的数据都应该以同步方式处理。含义 - 对于每次迭代，从第一个异步调用中检索到的数据应该传递给第二个异步调用。在第二个异步调用完成后，来自两个异步调用的数据应该一起处理。

Parallel.ForEach(someCollection, parallelOptions, async (item, state) =>
{
    Task<Country> countryTask = Task.Run(() => GetCountry(item.ID));

    //this is my first async call
    await countryTask.ContinueWith((countryData) =>
    {
        countries.Add(countryData.Result);

        Task<State> stateTask = Task.Run(() => GetState(countryData.Result.CountryID));

        //based on the data I receive in 'stateTask', I make another async call
        stateTask.ContinueWith((stateData) =>
        {
            states.Add(stateData.Result);

            // use data from both the async calls pass it to below function for some calculation
            // in a synchronized way (for …

private static async Task FuncAsync(DataTable dt, DataRow dr)
{
    try
    {
        await Task.Delay(3000); //assume this is an async http post request that takes 3 seconds to respond
        Thread.Sleep(1000) //assume this is some synchronous code that takes 2 second
    }
    catch (Exception e)
    {
        Thread.Sleep(1000); //assume this is synchronous code that takes 1 second
    }
}

private async void Button1_Click(object sender, EventArgs e)
{
    List<Task> lstTasks = new List<Task>();

    DataTable dt = (DataTable)gridview1.DataSource;

    foreach (DataRow dr in dt.Rows)
    {
        lstTasks.Add(FuncAsync(dr["colname"].ToString());                
    } …