zoukankan      html  css  js  c++  java
  • Azure Messaging-ServiceBus Messaging消息队列技术系列5-重复消息:at-least-once at-most-once

    上篇博客中,我们用实际的业务场景和代码示例了Azure Messaging-ServiceBus Messaging对复杂对象消息的支持和消息的持久化:

    Azure Messaging-ServiceBus Messaging消息队列技术系列4-复杂对象消息是否需要支持序列化和消息持久化

    本文中我们主要研究并介绍Azure Messaging对重复消息的支持。

    MessageReceiver 对象创建时可以指定消息接收模式: ReceiveAndDeletePeekLock (默认),其中:

    1. 使用 ReceiveAndDelete 模式时,接收是单步操作,即当 Service Bus 收到请求时,它将消息标记为“正在使用”,然后将其返回给应用程序。ReceiveAndDelete 模式是最简

    单的模型,并且最适合在出现故障时应用程序能够容许不处理消息的场景。理解此模式时,可考虑这种情况:使用者发出了接收请求,但在处理消息之前发生崩溃。由于 Service B

    us已将消息标记为“正在使用”,因此当应用程序重新启动并重新开始使用消息时,它就会错过在崩溃前已使用的消息。

    2. 在 PeekLock 模式下,接收变成两阶段操作,因此可以支持不能容许错过消息的应用程序。当 Service Bus 收到请求时,它会找到下一条要使用的消息,将其锁定以防止其他使

    用者接收它,然后将其返回给应用程序。应用程序完成消息处理(或将消息可靠地存储以便将来处理)后,会对收到的消息调用 Complete 以完成接收过程的第二阶段。当Service

    Bus 看到 Complete 时,会将该消息标记为“正在使用”。另外两个结果也是可能的。第一个结果,如果由于某种原因应用程序无法处理该消息,它可以对收到的消息Abandon(而

    不是 Complete)。这将导致 Service Bus 解锁该消息,并使该消息可以重新被同一使用者或其他竞争的使用者接收。第二个结果,即存在与锁定关联的超时,如果应用程序在锁

    定超时到期前无法处理改消息(例如,应用程序崩溃)则 Service Bus 将解锁该消息并使其可以重新被接收。如果应用程序在处理该消息后崩溃,但此时尚未发出 Complete 请

    求,则在应用程序重新启动时,该消息将重新传递给应用程序。这通常称为“至少一次”处理。这意味着每条消息都将至少处理一次,但在某些情况下可能会重新传递同一消息。如果

    方案不能容许重复处理,则需要在应用程序中添加检测重复项的逻辑。这可以基于消息的 MessageId 属性来实现。此属性的值在传递尝试过程中保持不变。这称为“恰好一次”处、

    理。

    接下来,我们通过Code show一下消息的重复发送和重复接收。

    消息重复发送:同一个消息BrokeredMessage发送两次

    /// <summary>
            /// 发送消息
            /// </summary>
            private static void MessageMultiSendTest()
            {
                var sbUtils = new ServiceBusUtils();
    
                //创建队列
                sbUtils.CreateQueue(queueName, false);
    
                //多次发送消息到OrderQueue
                var queueSendClient = sbUtils.GetQueueClient(queueName);
    
                var order = CreateSalesOrder(1);
                var message = sbUtils.Create(order);
                queueSendClient.Send(message);
                queueSendClient.Send(message);
    
                Console.WriteLine("Send Completed!");
            }

    实际执行过程中是出错的:

    由此可以得出:

    Azure Messaging 不支持同一个消息发送多次,必须通过new多个BrokeredMessage实例实现。

    同时,消息的唯一性由消息的MessageID来标识!

    PeekAndLock模式下消息的重复接收:

    接收模式PeekAndLock,同一个队列,第一个Consumer接收消息,但是不Complete;然后第二个Consumer继续接收消息,此时第一个Consumer未Complete的消息有一个

    TTL,在TTL时间区间之内,第二个Consumer可以继续接收当前队列未锁定的消息,当TTL时间到达后,释放第一个Consumer锁定的消息,第二个Consumer读取到了第一个

    Consumer未Complete的消息。

    The duration of a peek lock; that is, the amount of time that the message is locked for other receivers. The maximum value for LockDuration is 5 minutes;

    the default value is 1 minute.

     /// <summary>
            /// 接收消息
            /// </summary>
            private static void MessageReceive()
            {
                int index = 0;            
                var sbUtils = new ServiceBusUtils();
                var queueReveiveClient1 = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.PeekLock);
                for (int i = 0; i < 10; i++)
                {
                    var msg = queueReveiveClient1.Peek();
                    Console.WriteLine(string.Format("Received {0} MessageID: {1}", i, msg.MessageId));
                }
    
                var queueReveiveClient2 = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.PeekLock);
                for (int i = 0; i < 10; i++)
                {
                    var msg = queueReveiveClient2.Receive();
                    Console.WriteLine(string.Format("Second received {0} MessageID: {1}", i, msg.MessageId));
                    msg.Complete();
                }
    
                ////删除队列
                //sbUtils.DeleteQueue(queueName);
    
                Console.WriteLine("Receive Completed!");
            }

    第一个队列Consumer使用Peek模式接收到消息,只是取出消息,不从消息队列中移出。

    第二个队列Consumer使用Receive模式同样可以接收消息。

    如果两个队列Consumer都使用Receive模式接收消息,只有第一个Consumer可以接收到,第二个Consumer则接收不到,一直在等待消息的入队!

     /// <summary>
            /// 接收消息
            /// </summary>
            private static void MessageReceive()
            {
                int index = 0;            
                var sbUtils = new ServiceBusUtils();
                var queueReveiveClient1 = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.PeekLock);
                for (int i = 0; i < 10; i++)
                {
                    var msg = queueReveiveClient1.Receive();
                    Console.WriteLine(string.Format("Received {0} MessageID: {1}", i, msg.MessageId));
                }
    
                var queueReveiveClient2 = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.PeekLock);
                for (int i = 0; i < 10; i++)
                {
                    var msg = queueReveiveClient2.Receive();
                    Console.WriteLine(string.Format("Second received {0} MessageID: {1}", i, msg.MessageId));
                    msg.Complete();
                }
    
                ////删除队列
                //sbUtils.DeleteQueue(queueName);
    
                Console.WriteLine("Receive Completed!");
            }

    因为消息已经被第一个Consumer消费。

    通过本篇我们了解了Azure Messaging-ServiceBus Messaging对重复消息的处理机制。

    周国庆

    2017/3/16

  • 相关阅读:
    线程的取消/撤销 (转)
    Linux 3.0发布有感(转)
    Linux下挂载与解除挂载U盘
    Ubuntu 中文编码设置
    Linux Kernel 3.0新特性概览(转)
    pthread_cond_wait()用法分析
    brk和sbrk及内存分配函数介绍
    [转]Vmware ESX 4上虚拟机 Redhat 5.2(CentOS 5.2)启动在Starting udev 停几个小时
    [贺]通过Oracle 10g OCP的三门考试
    [原]Oracle外部表结合游标完成统计一例
  • 原文地址:https://www.cnblogs.com/tianqing/p/6561635.html
Copyright © 2011-2022 走看看