c# - 使用 Microsoft.ServiceBus.Messaging.MessageSession 许多消息 Complete()、Abort() 等

Question

当要接收多条消息时，我在关闭 MessageSession 时遇到问题...我等待所有序列进入，然后尝试“Complete()”中的所有消息 session 。我尝试了很多不同的方法，从迭代收到的消息到“Complete()”-ing它们，“Close()”-ing session ，“Completing()”-ing session ，我找不到获胜的公式我的生活。微软的文档也不是很有帮助。它当然没有提供一个像样的例子。以下是我在最近迭代中的代码

公共(public)QueueMessageBody接收响应(字符串sessionId) {

        try
        {
            MessagingFactory _messagingFactory;
            _messagingFactory = MessagingFactory.CreateFromConnectionString(connectionString);

            var receiver = _messagingFactory.CreateQueueClient(ResponseQueue, ReceiveMode.PeekLock);
            var messageBodyStream = new MemoryStream();

            while (true)
            {
                Console.WriteLine("Waiting for new message");
                using (var scope = new TransactionScope())
                {
                    var receiverSession = receiver.AcceptMessageSession(sessionId);
                    messageBodyStream = new MemoryStream();

                    var keepPolling = true;
                    var isFirstMessage = true;
                    var expectedNoMessages = 0;
                    BrokeredMessage[] messages = null;
                    var messagesReceived = 0;
                    while (keepPolling)
                    {
                        var message = receiverSession.Receive(TimeSpan.FromSeconds(10));
                        if (message == null)
                            continue;

                        if (isFirstMessage)
                        {
                            Console.WriteLine("Receiving first message");
                            expectedNoMessages = (int)message.Properties["TotalMessages"];
                            messages = new BrokeredMessage[expectedNoMessages];
                            isFirstMessage = false;
                        }

                        var messageNo = (int)message.Properties["MessageNo"];
                        var messageIndex = messageNo - 1;
                        Console.WriteLine(string.Format("Receiving message {0}", messageNo));
                        messages[messageIndex] = message;
                        messagesReceived++;

                        if (messagesReceived == expectedNoMessages)
                            keepPolling = false;
                    }





                    //Rebuild Object
                    var messageBodyBuilder = new ChunkedMessageBuilder().ReconstructMessageBody(messages);
                    var o = SerializationHelper.Deserialize<QueueMessageBody>(messageBodyBuilder);

                    var completeTasks = new List<Task>();
                    if (messages.Count()< 3)
                    {
                        foreach (var message in messages)
                            completeTasks.Add(receiverSession.CompleteAsync(message.LockToken));
                        Task.WaitAll(completeTasks.ToArray());
                    }else
                    {
                        receiverSession.Close();
                    }


                    return o;


                }
            }



        }
        catch(Exception ex)
        {
           System.Diagnostics.Debug.Print(ex.Message);
            return null;
        }

    }


  public   QueueMessageBody ReceiveResponse(string sessionId)
    {

        try
        {
            MessagingFactory _messagingFactory;
            _messagingFactory = MessagingFactory.CreateFromConnectionString(connectionString);

            var receiver = _messagingFactory.CreateQueueClient(ResponseQueue, ReceiveMode.PeekLock);
            var messageBodyStream = new MemoryStream();

            while (true)
            {
                Console.WriteLine("Waiting for new message");
                using (var scope = new TransactionScope())
                {
                    var receiverSession = receiver.AcceptMessageSession(sessionId);
                    messageBodyStream = new MemoryStream();

                    var keepPolling = true;
                    var isFirstMessage = true;
                    var expectedNoMessages = 0;
                    BrokeredMessage[] messages = null;
                    var messagesReceived = 0;
                    while (keepPolling)
                    {
                        var message = receiverSession.Receive(TimeSpan.FromSeconds(10));
                        if (message == null)
                            continue;

                        if (isFirstMessage)
                        {
                            Console.WriteLine("Receiving first message");
                            expectedNoMessages = (int)message.Properties["TotalMessages"];
                            messages = new BrokeredMessage[expectedNoMessages];
                            isFirstMessage = false;
                        }

                        var messageNo = (int)message.Properties["MessageNo"];
                        var messageIndex = messageNo - 1;
                        Console.WriteLine(string.Format("Receiving message {0}", messageNo));
                        messages[messageIndex] = message;
                        messagesReceived++;

                        if (messagesReceived == expectedNoMessages)
                            keepPolling = false;
                    }





                    //Rebuild Object
                    var messageBodyBuilder = new ChunkedMessageBuilder().ReconstructMessageBody(messages);
                    var o = SerializationHelper.Deserialize<QueueMessageBody>(messageBodyBuilder);

                    var completeTasks = new List<Task>();
                    if (messages.Count()< 3)
                    {
                        foreach (var message in messages)
                            completeTasks.Add(receiverSession.CompleteAsync(message.LockToken));
                        Task.WaitAll(completeTasks.ToArray());
                    }else
                    {
                        receiverSession.Close();
                    }


                    return o;


                }
            }



        }
        catch(Exception ex)
        {
           System.Diagnostics.Debug.Print(ex.Message);
            return null;
        }

    }

队列配置如下。

我能够接收消息并完成它们的工作流程，问题是它们只是停留在队列中。我无法在

完成它们

Answer 1

在这种情况下，问题可能是交易从未完成。删除 TransactionScope 就解决了这个问题。