Threads Потоки синхронизация, один producer два consumer`a и очередь

@OlegLysovych · Регистрация: 19.03.2017

Author24 — интернет-сервис помощи студентам

Привет. Только начал в шарпы, не говоря уже про потоки
и мне тут такую задачку дали
Итак дано: один produser два consumer`a.
Продюсер добавляет задачи в очередь, а эти два их обрабатывают.
Если задача поступает в момент когда 1 консюмер свободен, тот её обрабатывает, иначе в очередь. Второй вступает в работу только когда в очереди наберется допустим более 10 задач.
Я уже накидал таймер в продюсере, что добавляет задачи каждые 2 секунды в очередь, два потока потребителей у меня используют один и тот же метод, где я пытался изобразить поочередность вступления в работу с EventWaitHandle.
Наделал костылей, что работает только на выключения второго потока вовсе, потом он не включается.
Кажется, что делать две EventWaitHandle бред, ибо их видят сразу все потоки.
Потом тот же результат получил с Семафор(1,2).
Собственно вопрос, как сделать такую поочередность доступа, чтобы второй потребитель включался в зависимости от количества елементов в очереди. Что использовать? хелп ми плиз
Метод с которым работают потребители:

C#

EventWaitHandle SignalForConsumerOne = new ManualResetEvent(true);
EventWaitHandle SignalForConsumerTwo = new ManualResetEvent(false);
 
public Semaphore semaphore = new Semaphore(1,2); 
 
 
public void Consume()
        {
 
            //Вот тут я пробовал понять как работаетEventWaitHandle, так в работу вступает один потребитель
            SignalForConsumerOne.Set();
            SignalForConsumerTwo.WaitOne();
            if (Thread.CurrentThread.Name == "Thread_1" && _queue.Lenght <= 10)
                SignalForConsumerTwo.WaitOne();
            else
            {
                if (Thread.CurrentThread.Name == "Thread_0")
                {
                    SignalForConsumerOne.Set();
                    if (_queue.Lenght > 10)
                        SignalForConsumerTwo.Set();
                }
 
            }
            //ИЛИ
            semaphore.WaitOne(); //Вот так приступает в роботу сразу два потребителя
 
            Console.WriteLine($" Consumer {Thread.CurrentThread.Name} started ");
            Thread.Sleep(1000);
            while (true)
            {
                                  
                string task = null;
                lock (_queue.locker)
                {
                    if (_queue.Lenght > 0)
                    {
                        task = _queue.Dequeue();
                        if (task == null)
                        {
                            Console.WriteLine("Queue is empty");
                            return;
                        }
                    }
                }
                if (task != null)
                {
                    Console.WriteLine($"Number of elements in queue: {_queue.Lenght}");
                    Console.WriteLine("Produser prodused: {0} ---- and Consumer consume in Thread {1}", task, Thread.CurrentThread.Name);
                    Thread.Sleep(2000);
                }
                else
                {
                   // dispose
                }
                if ( _queue.Lenght == 10)
                {
                    //Отсюда делать Semaphore.Release() или SignalForConsumerTwo.Set() и снова вызывать Consume, чтобы уже с двумя потребителями началась работа, но получается, что далее работает только один.
                    //Consume;
                    //break; 
                }
            }
        }

@ShaRpN00B · 22.03.2020, 07:20

OlegLysovych, Здравствуйте, я новичок в C# и потоках, но стала интересна ваша задача, посидел поковырялся, один из вариантов который пришел в голову:

C#

        private static Queue<string> works;
        static Thread producerThread,worker1, worker2;
        static Random rndSleep;
        static void Main(string[] args)
        {
            works = new Queue<string>();
            rndSleep = new Random();
            producerThread = new Thread(new ThreadStart(producer));
            producerThread.Start();
 
            worker1 = new Thread(new ThreadStart(worker));
            worker1.Name = "woker1";
            worker1.Start();
 
            worker2 = new Thread(new ThreadStart(worker));
            worker2.Name = "worker2";
            worker2.Start();
 
            Console.ReadKey();
        }
 
        static void producer()
        {
            int workId = 0;
            while (true)
            {
                try
                {
                    works.Enqueue("work" + workId);
                    workId++;
                    Thread.Sleep(rndSleep.Next(200,1000));
                }
                catch(Exception e)
                {
                    Console.WriteLine(e.Message);
                    Console.WriteLine(e.StackTrace);
                    Console.WriteLine(works.Count);
                }
            }
        }
 
        static void worker()
        {
            while (true)
            {
                if (Thread.CurrentThread == worker2)
                {
                    if (works.Count > 10)
                    {
                        Console.Write("Помогаю ");
                        doWork();
                    }
                }
                else
                {
                    doWork();
                }
            }
        }
 
        static void doWork()
        {
            if (works.Count != 0)
            {
                Console.WriteLine($"WORK:{works.Dequeue()} THREAD:{Thread.CurrentThread.Name} WORKS:{works.Count}");
                Thread.Sleep(rndSleep.Next(600, 1500));
            }
        }

Я не уверен что так правильно и что так можно делать, но вроде что-то работает)) Надеюсь кто-то из опытных поправит и подскажет как правильно)

Добавлено через 13 минут
OlegLysovych, Хотя прошу прощения, не совсем верно, не все условия задачи выполняются) Например проверка на занятость потока)

@freeba · 22.03.2020, 07:48

Как то так.

C#

using System;
using System.Threading.Channels;
using System.Threading.Tasks;
 
namespace ConsoleApp46
{
    class WorkData
    {
        public string Message { get; set; }
        public int Delay { get; set; }
    }
 
    class Program
    {
        static readonly Random rnd = new Random();
 
        static readonly Channel<WorkData> channel1 = Channel.CreateBounded<WorkData>(new BoundedChannelOptions(10) { SingleWriter = true });
        static readonly Channel<WorkData> channel2 = Channel.CreateUnbounded<WorkData>(new UnboundedChannelOptions() { SingleWriter = true });
 
        static void Main(string[] args)
        {
            Task.Run(async () => await Worker1());
            Task.Run(async () => await Worker2());
 
            Task task = Go();
            Console.ReadKey();
        }
 
        static async Task Go()
        {
            var count = 100;
 
            while (count-- > 0)
            {
                var data = new WorkData() { Delay = rnd.Next(100, 300), Message = $"Operation: {100 - count}" };
 
                var isWrite = channel1.Writer.TryWrite(data);
                if (!isWrite) await channel2.Writer.WriteAsync(data);
 
                await Task.Delay(100);
            }
        }
 
        static async Task Worker1()
        {
            while (await channel1.Reader.WaitToReadAsync())
            {
                if (channel1.Reader.TryRead(out WorkData data))
                {
                    await Task.Delay(data.Delay);
                    Console.WriteLine($"[Worker1]: {data.Message}");
                }
            }
        }
 
        static async Task Worker2()
        {
            while (await channel2.Reader.WaitToReadAsync())
            {
                if (channel2.Reader.TryRead(out WorkData data))
                {
                    await Task.Delay(data.Delay);
                    Console.WriteLine($"[Worker2]: {data.Message}");
                }
            }
        }
    }
}

@nicolas2008 · 22.03.2020, 22:49

C#

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
 
namespace ConsoleApp1
{
    class Program
    {
        static void Main(string[] args)
        {
            ConcurrentQueue<int> workItems = new ConcurrentQueue<int>();
            AutoResetEvent mainWorkerSignal = new AutoResetEvent(false);
            AutoResetEvent reserveWorkerSignal = new AutoResetEvent(false);
            CancellationTokenSource cancellation = new CancellationTokenSource();
 
            var mainWorker = StartThread(() =>
            {
                while (!cancellation.Token.IsCancellationRequested)
                {
                    mainWorkerSignal.WaitOne();
 
                    while (workItems.TryDequeue(out int item))
                    {
                        Console.WriteLine($"Processing work item {item} in Main Worker. Remaining items in queue: {workItems.Count}");
                        Thread.Sleep(1000);
                    }
                }
            });
            
            var reserveWorker = StartThread(() =>
            {
                while (!cancellation.Token.IsCancellationRequested)
                {
                    reserveWorkerSignal.WaitOne();
                    
                    while (workItems.Count >= 10 && workItems.TryDequeue(out int item))
                    {
                        Console.WriteLine($"Processing work item {item} in Reserve Worker. Remaining items in queue: {workItems.Count}");
                        Thread.Sleep(500);
                    }
                }
            });
 
            var producer = StartThread(() =>
            {
                foreach (var workItem in Enumerable.Range(1, 30))
                {
                    workItems.Enqueue(workItem);
                    mainWorkerSignal.Set();
 
                    if (workItems.Count >= 10)
                    {
                        reserveWorkerSignal.Set();
                    }
                    
                    Thread.Sleep(200);
                }
            });
 
            producer.Join();
            mainWorker.Join();
            reserveWorker.Join();
 
            Console.WriteLine("Finished");
            Console.ReadKey();
        }
 
        static Thread StartThread(Action threadTask)
        {
            var thread = new Thread(() => threadTask());
            thread.Start();
            return thread;
        }
    }
}

Блоги программистов
Как перейти с Options API на Composition API в Vue.js BasicMan 06.01.2025 Почему переход на Composition API актуален В мире современной веб-разработки фреймворк Vue. js продолжает эволюционировать, предлагая разработчикам все более совершенные инструменты для создания. . .	Архитектура современных процессоров inter-admin 06.01.2025 Процессор (центральный процессор, ЦП) является основным вычислительным устройством компьютера, которое выполняет обработку данных и управляет работой всех остальных компонентов системы. Архитектура. . .	История создания реляционной модели баз данных, правила Кодда Programming 06.01.2025 Предпосылки создания реляционной модели В конце 1960-х годов компьютерная индустрия столкнулась с серьезными проблемами в области управления данными. Существовавшие на тот момент модели данных -. . .	Полезные поделки на Arduino, которые можно сделать самому raxper 06.01.2025 Arduino как платформа для творчества Arduino представляет собой удивительную платформу для технического творчества, которая открывает безграничные возможности для создания уникальных проектов. Эта. . .	Подборка решений задач на Python IT_Exp 06.01.2025 Целью данной подборки является предоставление возможности ознакомиться с различными задачами и их решениями на Python, что может быть полезно как для начинающих, так и для опытных программистов. . . .	С чего начать программировать микроконтроллеры raxper 06.01.2025 Введение в мир микроконтроллеров Микроконтроллеры стали неотъемлемой частью современного мира, окружая нас повсюду: от простых бытовых приборов до сложных промышленных систем. Эти маленькие. . .
Из чего собрать игровой компьютер inter-admin 06.01.2025 Сборка игрового компьютера требует особого внимания к выбору комплектующих и их совместимости. Правильно собранный игровой ПК не только обеспечивает комфортный геймплей в современных играх, но и. . .	Обновление сайта www.historian.by Reglage 05.01.2025 Обещал подвести итоги 2024 года для сайта. Однако начну с того, что изменилось за неделю. Добавил краткий урок по последовательности действий при анализе вредоносных файлов и значительно улучшил урок. . .	Как использовать GraphQL в C# с HotChocolate Programming 05.01.2025 GraphQL — это современный подход к разработке API, который позволяет клиентам запрашивать только те данные, которые им необходимы. Это делает взаимодействие с API более гибким и эффективным по. . .	Модель полного двоичного сумматора с помощью логических операций (python) AlexSky-coder 04.01.2025 def binSum(x:list, y:list): s=^y] p=x and y for i in range(1,len(x)): s. append((x^y)^p) p=(x and y)or(p and (x or y)) return s x=list() y=list()	Это мы не проходили, это нам не задавали...(асихронный счётчик с управляющим сигналом задержки). Hrethgir 04.01.2025 Асинхронный счётчик на сумматорах (шестиразрядный по числу диодов на плате, но наверное разрядов будет больше - восемь или шестнадцать, а диоды на старшие), так как триггеры прошли тестирование и. . .	Руководство по созданию бота для Телеграм на Python IT_Exp 04.01.2025 Боты для Телеграм представляют собой автоматизированные программы, которые выполняют различные задачи, взаимодействуя с пользователями через интерфейс мессенджера. В данной статье мы рассмотрим,. . .

	22.03.2020, 22:49

Threads Потоки синхронизация, один producer два consumer`a и очередь

Решение