Quiero crear un servicio WCF que use un enlace MSMQ ya que tengo un alto volumen de notificaciones que el servicio debe procesar. Es importante que los clientes no sean retenidos por el servicio y que las notificaciones se procesen en el orden en que se generan, de ahí la implementación de la cola.

Otra consideración es la resiliencia. Sé que podría agrupar MSMQ para hacer que la cola sea más robusta, pero quiero poder ejecutar una instancia de mi servicio en diferentes servidores, por lo que si un servidor falla las notificaciones no se acumulan en la cola, pero otro servidor continúa procesando .

Experimenté con el enlace MSMQ y descubrí que puedes tener varias instancias de un servicio escuchando en la misma cola, y dejaron para sí mismos haciendo una especie de round-robin con la carga distribuida entre los servicios disponibles. Esto es genial, pero termino perdiendo la secuencia de la cola a medida que diferentes instancias toman una cantidad diferente de tiempo para procesar la solicitud.

He estado usando una aplicación de consola simple para experimentar, que es el volcado de código épico a continuación. Cuando se ejecuta me sale una salida como esta:

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

Lo que quiero que suceda es:

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.

Pensé que como S2 no se ha completado, podría fallar y devolver el mensaje que estaba procesando a la cola. Por lo tanto, no se debe permitir que S1 saque otro mensaje de la cola. Mi cola es transaccional y he intentado configurarTransactionScopeRequired = true en el servicio, pero en vano.

¿Es esto posible? ¿Estoy yendo por el camino equivocado? ¿Hay alguna otra forma de crear un servicio de conmutación por error sin algún tipo de mecanismo de sincronización central?

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);
        }
    }
}