Почему очередь не передаёт данные через мютекс из процесса в поток?
Почему очередь не передаёт данные через мютекс из процесса в поток ??
Вот код :
// Применить мьютекс.
using System;
using System.Threading;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
// В этом классе содежится общий ресурс (переменная Queue),
// а также мьютекс(Mtx), управляющий доступом к ней.
class SharedRes
{
//static readonly Queue<string> data = new Queue<string>();
public static Queue<string> data = new Queue<string>();
public static Mutex Mtx = new Mutex();
}
// В этом потоке переменная SharedRes.Queue инкрементируется.
class IncThread
{
int num;
public Thread Thrd;
public IncThread(string name)
{
Thrd = new Thread(this.Run);
Thrd.Name = name;
Thrd.Start();
}
// Точка входа в поток.
void Run()
{
ProcessStartInfo startInfo = new ProcessStartInfo()
{
FileName = "ping",
Arguments = "www.ya.ru",
RedirectStandardOutput = true,
UseShellExecute = false,
CreateNoWindow = true
};
using (var pingProcess = new Process())
{
pingProcess.StartInfo = startInfo;
pingProcess.OutputDataReceived += OutputDataReceived;
pingProcess.Start();
pingProcess.BeginOutputReadLine();
// На этом методе будет ожидание завершения запущенного процесса.
// При получении каждой новой порции данных
// будет вызываться событие OutputDataReceived.
pingProcess.WaitForExit();
}
Console.WriteLine(Thrd.Name + " ожидает мютекс.");
}
private static void OutputDataReceived(object sender, DataReceivedEventArgs e)
{
// Получить мьютекс.
SharedRes.Mtx.WaitOne();
Console.WriteLine("Мы в потоке один");
SharedRes.data.Enqueue(e.Data);
//Console.WriteLine(e.Data);
using (var writer = File.AppendText("log.txt"))
writer.WriteLine(e.Data);
// Освободить мьютекс.
SharedRes.Mtx.ReleaseMutex();
}
}
// В этом потоке переменная SharedRes.Count декрементируеться.
class DecThread
{
int num;
public Thread Thrd;
public DecThread(string name)
{
Thrd = new Thread(new ThreadStart(this.Run));
Thrd.Name = name;
Thrd.Start();
}
// ТОчка входа в поток.
void Run()
{
// Получить мьютекс.
SharedRes.Mtx.WaitOne();
Console.WriteLine("Мы в потоке два");
while (SharedRes.data.Count > 0)
Console.WriteLine(SharedRes.data.Dequeue());
// Освободить мьютекс.
SharedRes.Mtx.ReleaseMutex();
}
}
class MutexDemo
{
static void Main()
{
// Construct three threads.
IncThread mt1 = new IncThread("Инкрементирующий поток");
Thread.Sleep(1); // let the Increment thread start
DecThread mt2 = new DecThread("Декрементирующий поток");
mt1.Thrd.Join();
mt2.Thrd.Join();
}
}
Вот вывод :
dima@komp:~/mita/TCP_IP$ mcs lovla_terminal7.cs
dima@komp:~/mita/TCP_IP$ mono lovla_terminal7.exe
Мы в потоке два
Мы в потоке один
Мы в потоке один
Мы в потоке один
Мы в потоке один
Мы в потоке один
Мы в потоке один
Мы в потоке один
Мы в потоке один
^C
Ответы (1 шт):
Попробую разобрать все проблемы, которые я увидел в коде.
Я бы не использовал
Mutexдля такой задачи, как иAutoResetEvent, так как оба тяжелые, и если вход в некую область выполнения кода должен производиться одновременно только одним потоком, я бы использовалManualResetEventSlimили простоlock()какого-нибудь общего объекта.А
Mutexя бы вообще оставил для кросс-процессной синхронизации, например, с помощью него можно сделать, чтобы нельзя было запустить 2 копии программы. Его именно для этого чаще всего и используют.В вашем же случае второй поток стартует и пытается освободить еще пустую очередь, и сразу же завершается. Проблема наверное была не в мьютексе. Ну или я неправильно понял, как вы хотели, чтобы оно работало.
Если задача состоит в том, чтобы сначала один поток заполнил очередь, а второй потом ее освободил, то вы неправильно вызываете потоки, у вас должно получиться что-то такое.
static void Main() { IncThread mt1 = new IncThread("Инкрементирующий поток"); mt1.Thrd.Join(); DecThread mt2 = new DecThread("Декрементирующий поток"); mt2.Thrd.Join(); }И да, здесь вообще теряется весь смысл многопоточности, ну кроме того, что она есть и работает. Эффекта от нее нет.
Лучше использовать
TaskвместоThread.Если же нужна реализация фокуса, состоящего в том, чтобы получать данные от внешнего процесса и одновременно по факту получения забирать их в другом потоке, то в бой вступают конкурентные коллекции, специально предназначенные для многопоточности. Моя любимая коллекция из таких -
BlockingCollection.
Итак, попробую воплотить в коде то, что понаписал выше. На базе обычного консольного приложения.
public class Program
{
static void Main(string[] args)
{
// Запускаем задачи
Task task1 = new Producer("Передающая задача").Task;
Task task2 = new Consumer("Принимающая задача").Task;
// И ждем их завершения
Task.WaitAll(task1, task2);
Console.WriteLine("Все задачи завершены.");
// Ну и чтобы окошко сразу не закрылось
Console.ReadKey();
}
}
static class SharedRes
{
// та самая блокирующая коллекция
public static BlockingCollection<string> Data { get; } = new BlockingCollection<string>();
// содержит отменяющий токен, необходимый для прерывания ожидания блокирующей коллекции
public static CancellationTokenSource Cts { get; } = new CancellationTokenSource();
}
class Producer
{
public Task Task { get; }
public string Name { get; }
public Producer(string name)
{
Name = name;
Task = Task.Run(Run);
}
void Run()
{
ProcessStartInfo startInfo = new ProcessStartInfo()
{
FileName = "ping",
Arguments = "www.ya.ru",
RedirectStandardOutput = true,
UseShellExecute = false,
CreateNoWindow = true
};
using (var pingProcess = new Process())
{
pingProcess.StartInfo = startInfo;
pingProcess.OutputDataReceived += OutputDataReceived;
pingProcess.Start();
pingProcess.BeginOutputReadLine();
pingProcess.WaitForExit();
}
SharedRes.Cts.Cancel(); // отправить сигнал в BlockingCollection, что пора заканчивать
Console.WriteLine(Name + " завершает работу.");
}
private static void OutputDataReceived(object sender, DataReceivedEventArgs e)
{
Console.WriteLine("Мы в задаче один");
SharedRes.Data.Add(e.Data);
// немного упростил запись в лог
File.AppendAllText("log.txt", e.Data + Environment.NewLine);
}
}
class Consumer
{
public Task Task { get; }
public string Name { get; }
public Consumer(string name)
{
Name = name;
Task = Task.Run(Run);
}
void Run()
{
try
{
// Здесь коллекция ждет, пока не появится новый элемент или пока токен не получит отмену
foreach (string s in SharedRes.Data.GetConsumingEnumerable(SharedRes.Cts.Token))
{
Console.WriteLine("Мы в задаче два");
Console.WriteLine(s);
}
}
catch (OperationCanceledException ex)
{
// Здесь можно вообще ничего не делать
Console.WriteLine("Ожидание новых данных прервано: " + ex.Message);
}
Console.WriteLine(Name + " завершает работу.");
}
}
Можно еще упростить код и сделать его приятнее с точки зрения логичности и читабельности, а также не блокировать основной поток при ожидании дочерних, с помощью async/await, но это уже другая история.
Вывод:
Мы в задаче один
Мы в задаче два
Мы в задаче один
Мы в задаче два
Pinging ya.ru [87.250.250.242] with 32 bytes of data:
Мы в задаче один
Мы в задаче два
Reply from 87.250.250.242: bytes=32 time=17ms TTL=249
Мы в задаче один
Мы в задаче два
Reply from 87.250.250.242: bytes=32 time=16ms TTL=249
Мы в задаче один
Мы в задаче два
Reply from 87.250.250.242: bytes=32 time=17ms TTL=249
Мы в задаче один
Мы в задаче два
Reply from 87.250.250.242: bytes=32 time=16ms TTL=249
Мы в задаче один
Мы в задаче два
Мы в задаче один
Мы в задаче два
Ping statistics for 87.250.250.242:
Мы в задаче один
Мы в задаче два
Packets: Sent = 4, Received = 4, Lost = 0 (0% loss),
Мы в задаче один
Мы в задаче два
Approximate round trip times in milli-seconds:
Мы в задаче один
Мы в задаче два
Minimum = 16ms, Maximum = 17ms, Average = 16ms
Мы в задаче один
Мы в задаче два
Предающая задача завершает работу.
Ожидание новых данных прервано: The operation was canceled.
Принимающая задача завершает работу.
Все задачи завершены.