MSMQ получает с транзакцией - откат не делает сообщение снова доступным - PullRequest
Новая
       41

MSMQ получает с транзакцией - откат не делает сообщение снова доступным

9 голосов
/ 16 ноября 2011

У меня есть это в классе под названием "MessageQueueReceive". 

public MessageQueueTransaction BlockingReceive(out Message message)
{
    MessageQueueTransaction tran = null;
    message = null;

    tran = new MessageQueueTransaction();

    tran.Begin();
    try
    {
        message = Queue.Receive(new TimeSpan(0, 0, 5), tran);
    }
    catch (MessageQueueException ex)
    {
        // If the exception was a timeout, then just continue
        // otherwise re-raise it.
        if (ex.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
            throw ex;
    }

    return tran;

}

Тогда мой цикл обработки имеет следующее: - 

while (!Abort)
{
    try
    {
        tran = this.Queue.BlockingReceive(out msg);

        if (msg != null)
        {
            // Process message here

            if (tran != null)
                tran.Commit();
        }
    }
    catch (Exception ex)
    {
        if (tran != null)
            tran.Abort();

    }
}

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

Этот код создает очередь: - 

private static MessageQueue CreateMessageQueue(string queueName, bool transactional = false)
{
    MessageQueue messageQueue = MessageQueue.Create(queueName, transactional);
    messageQueue.SetPermissions("Administrators", MessageQueueAccessRights.FullControl,
            AccessControlEntryType.Allow);
    return messageQueue;
}

Транзакционный параметр устанавливается как "true", когда он вызывается.

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

Я что-то упускаю из виду?Кто-нибудь может увидеть, что я делаю не так?

Ответы [ 2 ]

5 голосов
/ 26 января 2012

Спасибо за все комментарии. Я реорганизовал свой код, как предложил Рассел МакКлюр, и попытался создать простые тестовые случаи, но не смог воспроизвести проблему.

В конце концов, проблема была вовсе не в том, куда я смотрел (как часто это происходит?).

В моем конвейере у меня была программа проверки дубликатов сообщений. «Сообщения», с которыми работает моя система, поступают с удаленных устройств в глобальной сети, и иногда сообщения в сети дублируются.

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

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

1 голос
/ 16 ноября 2011

Ваша очередь должна быть создана как транзакционная очередь, чтобы получить то, что вы хотите.

enter image description here

EDIT:

Что ж, если ваша очередь является транзакционной, это указывает на тот факт, что вы неправильно обрабатываете свою транзакцию, хотя я не могу точно понять, как это происходит. Я бы изменил ваш BlockingReceive метод для возврата сообщения. Я бы переместил создание MessageQueueTransaction во внешний метод. Ваш код будет гораздо более удобен в обслуживании, если у вас есть вызовы методов Begin, Commit и Abort в одном и том же методе.

...