Открытая тема в цикле foreach

Question

Я получаю канал XML и анализирую его на своем сервере MQ, затем у меня есть служба, которая прослушивает сервер MQ и читает все его сообщения.

У меня есть цикл foreach, который открывает новый поток на каждой итерации, чтобы ускорить анализ, потому что в MQ около 500 сообщений (значит, 500 XML)

foreach (System.Messaging.Message m in msgs)
{
    byte[] bytes = new byte[m.BodyStream.Length];
    m.BodyStream.Read(bytes, 0, (int)m.BodyStream.Length);
    System.Text.ASCIIEncoding ascii = new System.Text.ASCIIEncoding();

    ParserClass tst = new ParserClass(ascii.GetString(bytes, 0, (int)m.BodyStream.Length));
    new Thread( new ThreadStart(tst.ProcessXML)).Start();
}

В ParserClass у меня есть этот код:

private static object thLockMe = new object();
public string xmlString { get; set; }

public ParserClass(string xmlStringObj)
{
    this.xmlString = xmlStringObj;
}

public void ProcessXML()
{
    lock (thLockMe)
    {
        XDocument reader = XDocument.Parse(xmlString);
        //Some more code...
    }
}

Проблема в том, что когда я запускаю этот цикл foreach только с 1 потоком, он работает идеально, но медленно.

Когда я запускаю его с более чем одним потоком, я получаю ошибку «Ссылка на объект не установлена ​​на экземпляр объекта».

Полагаю, что с блокировкой что-то не так, поскольку я не очень разбираюсь в многопоточности.

Я немного безнадежен, надеюсь, вы поможете!

Ура!

Answer 1

Замечу, что вы запускаете кучу потоков с полным кодом, заключенным в оператор lock.Вы могли бы также запустить методы в последовательности таким образом, потому что вы не получаете никакого параллелизма.

Answer 2

Поскольку вы создаете новый экземпляр ParserClass на каждой итерации цикла, а также создаете и запускаете новый поток на каждой итерации, вам не требуется блокировка в вашем методе ParseXML.

Ваш объект, для которого вы блокируете, в настоящее время static, поэтому он не привязан к экземпляру, что означает, что как только один поток находится внутри вашего ParseXML метода, никто другой не сможет ничего сделать, пока не завершится первый.

Вы не делитесь никакими данными (из кода, который я вижу) в вашем классе Parser между потоками, поэтому вам не нужна блокировка внутри вашей функции ParseXML.

Если выиспользуя данные, которые совместно используются потоками, у вас должна быть блокировка.

Если вы собираетесь использовать много потоков, тогда вам лучше использовать ThreadPool и брать конечное число (4, возможно,) из вашего пула, назначив им некоторую работу и переработав их для следующих 4 задач.

Создание потоков - это дорогостоящая операция, которая требует обращения к ядру ОС, поэтому вы не хотите делать это 500 раз.Это слишком дорого.Кроме того, минимальная зарезервированная память для стека потоков в Windows составляет 1 МБ, то есть 500 МБ в одном только стековом пространстве для ваших потоков.

Оптимальное количество потоков должно быть равно количеству ядер на вашей машине, однако, поскольку это не реально для большинства целей, вы можете сделать это в два или три раза, но тогда вам лучше использовать пул потоковгде вы перерабатываете потоков, вместо , постоянно создающих новые.

Answer 3

Несмотря на то, что это, вероятно, не решит вашу проблему, вместо создания 500 одновременных потоков вы должны просто использовать ThreadPool, который управляет потоками гораздо более эффективным способом:

foreach (System.Messaging.Message m in msgs)
{
    byte[] bytes = new byte[m.BodyStream.Length];
    m.BodyStream.Read(bytes, 0, (int)m.BodyStream.Length);
    System.Text.ASCIIEncoding ascii = new System.Text.ASCIIEncoding();

    ParserClass tst = new ParserClass(ascii.GetString(bytes, 0, (int)m.BodyStream.Length));
    ThreadPool.QueueUserWorkItem(x => tst.ProcessXML());
}

И чтобы они выполнялись как можно более одновременно, измените ваш код в ParserClass следующим образом (при условии, что у вас действительно есть ресурсы, которые вы разделяете между потоками - если у вас их нет, вам не нужно блокировать все):

private static object thLockMe = new object();
public string XmlString { get; set; }

public ParserClass(string xmlString)
{
    XmlString = xmlString;
}

public void ProcessXML()
{
    XDocument reader = XDocument.Parse(xmlString);
    // some more code which doesn't need to access the shared resource

    lock (thLockMe)
    {
        // the necessary code to access the shared resource (and only that)
    }

    // more code
}

Относительно вашего актуального вопроса:

Вместо того, чтобы вызывать OddService.InsertEvent(...) несколько раз с одними и теми же параметрами (этот метод пахнет удаленными вызовами и побочными эффектами ...), вы должны вызвать его один раз, сохранить результат в переменной и выполнить все последующие операции с этой переменной. Таким образом, вы также можете удобно проверить, не является ли этот метод точным, который иногда возвращает ноль (при одновременном доступе?).

Изменить:

Работает ли это, если вы помещаете все вызовы на OddService.* в lock блоках?