如何使用多个WCF服务实例强制执行消息队列顺序

问题描述

我想创建一个使用MSMQ绑定的WCF服务，因为我要处理的通知数量很多.重要的是，不要让客户端阻塞服务，并按照发出通知的顺序处理通知，从而实现队列的实现.

I want to create a WCF service which uses an MSMQ binding as I have a high volume of notifications the service is to process. It is important that clients are not held up by the service and that the notifications are processed in the order they are raised, hence the queue implementation.

另一个考虑因素是弹性.我知道我可以对MSMQ本身进行群集以使队列更坚固，但是我希望能够在其他服务器上运行我的服务的实例，因此，如果服务器崩溃，则通知不会在队列中建立，而是由另一台服务器进行处理

Another consideration is resilience. I know I could cluster MSMQ itself to make the queue more robust, but I want to be able to run an instance of my service on different servers, so if a server crashes notifications do not build up in the queue but another server carries on processing.

我已经对MSMQ绑定进行了实验，发现您可以在同一队列上侦听服务的多个实例，而任由他们最终进行某种循环操作，负载分散在可用服务上.很好，但是由于不同的实例花费不同的时间来处理请求，我最终失去了队列的顺序.

I have experimented with the MSMQ binding and found that you can have multiple instances of a service listening on the same queue, and left to themselves they end up doing a sort of round-robin with the load spread across the available services. This is great, but I end up losing the sequencing of the queue as different instances take a different amount of time to process the request.

我一直在使用一个简单的控制台应用程序进行实验，这是下面的史诗代码转储.当它运行时，我得到这样的输出:

I've been using a simple console app to experiment, which is the epic code dump below. When it's run I get an output like this:

host1 open
host2 open
S1: 01
S1: 03
S1: 05
S2: 02
S1: 06
S1: 08
S1: 09
S2: 04
S1: 10
host1 closed
S2: 07
host2 closed

我想发生的是:

host1 open
host2 open
S1: 01
<pause while S2 completes>
S2: 02
S1: 03
<pause while S2 completes>
S2: 04
S1: 05
S1: 06
etc.

我会认为，由于S2尚未完成，它可能仍然会失败，并将正在处理的消息返回到队列中.因此，不应允许S1从队列中拉出另一条消息.我以事务性方式排队，我们尝试在服务上设置TransactionScopeRequired = true，但无济于事.

I would have thought that as S2 has not completed, it might still fail and return the message it was processing to the queue. Therefore S1 should not be allowed to pull another message off of the queue. My queue us transactional and I have tried setting TransactionScopeRequired = true on the service but to no avail.

这甚至有可能吗?我会以错误的方式去做吗?没有某种中央同步机制，还有其他方法可以构建故障转移服务吗?

Is this even possible? Am I going about it the wrong way? Is there some other way to build a failover service without some kind of central synchronisation mechanism?

class WcfMsmqProgram
{
    private const string QueueName = "testq1";

    static void Main()
    {
        // Create a transactional queue
        string qPath = ".\\private$\\" + QueueName;
        if (!MessageQueue.Exists(qPath))
            MessageQueue.Create(qPath, true);
        else
            new MessageQueue(qPath).Purge();

        // S1 processes as fast as it can
        IService s1 = new ServiceImpl("S1");
        // S2 is slow
        IService s2 = new ServiceImpl("S2", 2000);

        // MSMQ binding
        NetMsmqBinding binding = new NetMsmqBinding(NetMsmqSecurityMode.None);

        // Host S1
        ServiceHost host1 = new ServiceHost(s1, new Uri("net.msmq://localhost/private"));
        ConfigureService(host1, binding);
        host1.Open();
        Console.WriteLine("host1 open");

        // Host S2
        ServiceHost host2 = new ServiceHost(s2, new Uri("net.msmq://localhost/private"));
        ConfigureService(host2, binding);
        host2.Open();
        Console.WriteLine("host2 open");

        // Create a client
        ChannelFactory<IService> factory = new ChannelFactory<IService>(binding, new EndpointAddress("net.msmq://localhost/private/" + QueueName));
        IService client = factory.CreateChannel();

        // Periodically call the service with a new number
        int counter = 1;
        using (Timer t = new Timer(o => client.EchoNumber(counter++), null, 0, 500))
        {
            // Enter to stop
            Console.ReadLine();
        }

        host1.Close();
        Console.WriteLine("host1 closed");
        host2.Close();
        Console.WriteLine("host2 closed");

        // Wait for exit
        Console.ReadLine();
    }

    static void ConfigureService(ServiceHost host, NetMsmqBinding binding)
    {
        var endpoint = host.AddServiceEndpoint(typeof(IService), binding, QueueName);
    }

    [ServiceContract]
    interface IService
    {
        [OperationContract(IsOneWay = true)]
        void EchoNumber(int number);
    }

    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)]
    class ServiceImpl : IService
    {
        public ServiceImpl(string name, int sleep = 0)
        {
            this.name = name;
            this.sleep = sleep;
        }

        private string name;
        private int sleep;

        public void EchoNumber(int number)
        {
            Thread.Sleep(this.sleep);
            Console.WriteLine("{0}: {1:00}", this.name, number);
        }
    }
}

推荐答案

batwad

您正在尝试手动创建服务总线.您为什么不尝试使用现有的呢?

You are trying to manually create a service bus. Why don't you try to use an existing one?

NServiceBus，MassTransit，ServiceStack

NServiceBus, MassTransit, ServiceStack

其中至少有2个可以与MSMQ一起使用.

At least 2 of those work with MSMQ.

此外，如果您绝对需要订购，则可能实际上是由于另一个原因-您希望能够发送消息，并且不希望在第一个消息之前处理从属消息.您正在寻找Saga模式. NServiceBus和MassTransit都将使您能够轻松管理Sagas，它们都将使您可以简单地触发初始消息，然后根据条件触发其余消息.它将使您能够快速实现分布式应用程序的充实.

Furthermore, if you absolutely need order it may actually be for another reason - you want to be able to send a message and you don't want dependent messages to be processed before the first message. You are looking for the Saga Pattern. NServiceBus and MassTransit both will allow you to manage Sagas easily, they will both allow you to simply trigger the initial message and then trigger the remaining messages based on conditions. It will allow you to implement the plumping of your distributed application a snap.

您甚至可以扩展到数千个客户端，队列服务器和消息处理器，而无需编写任何代码，也不会出现任何问题.

You can then even scale up to thousands of clients, queue servers and message processors without having to write a single line of code nor have any issues.

我们试图在此处通过msmq实现我们自己的服务总线，我们放弃了，因为另一个问题一直在蔓延.我们选择了NServiceBus，但MassTransit还是一款出色的产品(它是100％开放源代码，不是NServiceBus). ServiceStack在制作API和使用消息队列方面非常出色-我相信您可以在几分钟内使用它来制作充当队列前端的服务.

We tried to implement our own service bus over msmq here, we gave up because another issue kept creeping up. We went with NServiceBus but MassTransit is also an excellent product (it's 100% open source, NServiceBus isn't). ServiceStack is awesome at making APIs and using Message Queues - I'm sure you could use it to make Services that act as Queue front-ends in minutes.

哦，我是否提到过，对于NSB和MT来说，它们都只需要10行以下的代码即可完全实现队列，发送者和处理者?

Oh, did I mention that in the case of NSB and MT both only require under 10 lines of code to fully implement queues, senders and handlers?

-----添加-----

----- ADDED -----

Udi Dahan(NServiceBus的主要贡献者之一)在以下方面谈到了这一点:乌迪·达汉(Udi Dahan)的有序消息传递神话" 消息订购:是否要付费有效的?"与乌迪·达汉(Udi Dahan)

Udi Dahan (one of the main contributers of NServiceBus) talks about this in:"In-Order Messaging a Myth" by Udi Dahan"Message Ordering: Is it Cost Effective?" with Udi Dahan

克里斯·帕特森(Chris Patterson)(地铁的主要贡献者之一)使用Sagas来确保正确的顺序消息顺序"问题

Chris Patterson (one of the main contributers of Mass Transit)"Using Sagas to ensure proper sequential message order" question

StackOverflow问题/解答:>在WCF应用程序"

StackOverflow questions/answers:"Preserve message order when consuming MSMQ messages in a WCF application"

-----问题-----

----- QUESTION -----

我必须对为什么需要保证消息顺序感到困惑-如果您使用的是HTTP/SOAP协议，您会处于同一位置吗?我的猜测是没有，那为什么在MSMQ中有问题?

I must say that I'm baffled as to why you need to guarantee message order - would you be in the same position if you were using an HTTP/SOAP protocol? My guess is no, then why is it a problem in MSMQ?

祝你好运，希望这对你有帮助，

Good luck, hope this helps,

这篇关于如何使用多个WCF服务实例强制执行消息队列顺序的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！