1

Есть очередь, состаящая из конечноко числа потоков

Queue<Thread> queue = new Queue<Thread>();
Thread[] thread = new Thread[4];       

for (int i = 0; i < thread.Length; ++i)
{
thread[i] = new Thread( new ParameterizedThreadStart(fun));
queue.Enqueue(thread[i]);
}

есть лист данных

List<String> info = new List<String>();

for (int i = 1; i <= 10; ++i)
 info.Add($"info{i}");

Требуется время по листу, проверять, если есть потоки в очереди, если есть, то достать и запустить на обработку порции данных, если очередь пуста, то ждать, пока в ней появятся доступные потоки (которые окончили обработку)

как это организовать? как сделать там, что бы потоки обратно добавляли в очередь после обработки порции данных и, когда очередь пуста, ожидать появления в очереди потоков?

4
  • Вам нужно смотреть в сторону ThreadPool. Вот один из примеров использования менеджера потоков.
    – Rootware
    20 фев 2018 в 20:43
  • Вы изобретаете велосипед, есть TPL DataFlow, там уже всё что вам надо реализованно, берёте ActionBlock, настраиваете, и готово 21 фев 2018 в 0:39
  • Что значит "Требуется время по листу"? Моя твоя не понимать... 21 фев 2018 в 5:42
  • @PrimusSingularis: Напишете ответ с ActionBlock?
    – VladD
    22 фев 2018 в 8:39

2 ответа 2

3

Современной практикой является не ручное управление потоками, а использование TPL и async/await для подобных вещей.

Например, вы можете использовать BufferBlock, как описано здесь. При этом ваш код получится примерно таким (я добавил вывод на консоль, чтобы было видно, как программа работает):

class Program
{
    static async Task Main(string[] args) => await new Program().Run();

    async Task Run()
    {
        // инициализация данных
        List<String> info = new List<String>();
        for (int i = 1; i <= 10; ++i)
            info.Add($"info{i}");

        // очередь заданий
        BufferBlock<string> queue = new BufferBlock<string>();
        var p = Produce(queue, info);
        var c = ConsumeAll(queue, 4);
        await Task.WhenAll(p, c, queue.Completion);
    }

    // поставщик
    async Task Produce(ITargetBlock<string> queue, List<String> info)
    {
        foreach (var data in info)
            await queue.SendAsync(data);
        queue.Complete();
    }

    // потребитель
    async Task ConsumeCooperative(IReceivableSourceBlock<string> queue, int number)
    {
        while (await queue.OutputAvailableAsync())
        {
            if (queue.TryReceive(out var v))
                await Task.Run(() => Process(v, number));
        }
    }

    Task ConsumeAll(IReceivableSourceBlock<string> queue, int quantity)
    {
        var tasks = Enumerable.Range(0, quantity)
                              .Select(n => ConsumeCooperative(queue, n));
        return Task.WhenAll(tasks);
    }

    // тут ваша обработка данных
    Random r = new Random();
    Stopwatch sw = Stopwatch.StartNew();
    void Process(string v, int consumerNumber)
    {
        Console.WriteLine($"Processing started, comsumer #{consumerNumber}," +
            $" data {v} at {sw.Elapsed:m\\:ss\\.ff}");
        Thread.Sleep(1000 * r.Next(5, 10));
        Console.WriteLine($"Processing finished, comsumer #{consumerNumber}," +
            $" data {v} at {sw.Elapsed:m\\:ss\\.ff}");
    }
}

Результат:

Processing started, comsumer #0, data info1 at 0:00.38
Processing started, comsumer #1, data info2 at 0:00.38
Processing started, comsumer #2, data info3 at 0:00.38
Processing started, comsumer #3, data info4 at 0:00.39
Processing finished, comsumer #1, data info2 at 0:05.41
Processing finished, comsumer #0, data info1 at 0:05.41
Processing started, comsumer #1, data info5 at 0:05.41
Processing started, comsumer #0, data info6 at 0:05.41
Processing finished, comsumer #3, data info4 at 0:05.43
Processing started, comsumer #3, data info7 at 0:05.44
Processing finished, comsumer #2, data info3 at 0:07.42
Processing started, comsumer #2, data info8 at 0:07.42
Processing finished, comsumer #0, data info6 at 0:10.42
Processing started, comsumer #0, data info9 at 0:10.42
Processing finished, comsumer #3, data info7 at 0:11.44
Processing started, comsumer #3, data info10 at 0:11.45
Processing finished, comsumer #1, data info5 at 0:12.44
Processing finished, comsumer #2, data info8 at 0:14.45
Processing finished, comsumer #3, data info10 at 0:17.48
Processing finished, comsumer #0, data info9 at 0:18.46
1

Поток - это активный объект. Он сам решает в каком момент взять себе задание и начать его выполнять. Поэтому очередь потоков сделать довольно трудно.

Проще всего развернуть задачу и сделать очередь задач.

BlockingCollection<string> queue = new BlockingCollection<string>();

// в каждом потоке:
foreach (var item in queue.GetConsumingEnumerable())
{
    // Обрабатываем item
}

// добавление задачи в очередь:
queue.Add(...);

// остановка потоков:
queue.CompleteAdding();
foreach (var thread in threads) thread.Join();
queue.Dispose(); // Важно: нельзя разрушать очередь пока потоки с ней работают

Если же вам по какой-то причине нужна именно очередь потоков - придется исхитриться. Идея в том, чтобы каждый поток складывал в очередь TaskCompletionSource, через который будет получать задание.

BlockingCollection<TaskCompletionSource<string>> queue = new BlockingCollection<TaskCompletionSource<string>>();

// в каждом потоке:
while (...)
{
    var tcs = new TaskCompletionSource<string>();
    queue.Add(tcs);
    var item = tcs.Task.Result;

    // Обработка item
}

// добавление задачи в очередь:
queue.Take().SetResult(...);

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.