Чтение из нескольких очередей, RabbitMQ

Question

Я новичок в RabbitMQ.Я хочу иметь возможность обрабатывать чтение сообщений без блокировки, когда есть несколько очередей (для чтения из).Есть какие-то отзывы о том, как я могу это сделать?

// Редактировать 1

public class Rabbit : IMessageBus
{   

    private List<string> publishQ = new List<string>();
    private List<string> subscribeQ = new List<string>();

    ConnectionFactory factory = null;
    IConnection connection = null;
    IModel channel = null;  
    Subscription sub = null;

    public void writeMessage( Measurement m1 ) {
        byte[] body = Measurement.AltSerialize( m1 );
        int msgCount = 1;
        Console.WriteLine("Sending message to queue {1} via the amq.direct exchange.", m1.id);

        string finalQueue = publishToQueue( m1.id );

        while (msgCount --> 0) {
            channel.BasicPublish("amq.direct", finalQueue, null, body);
        }

        Console.WriteLine("Done. Wrote the message to queue {0}.\n", m1.id);
    }

     public string publishToQueue(string firstQueueName) {
        Console.WriteLine("Creating a queue and binding it to amq.direct");
        string queueName = channel.QueueDeclare(firstQueueName, true, false, false, null);
        channel.QueueBind(queueName, "amq.direct", queueName, null);
        Console.WriteLine("Done.  Created queue {0} and bound it to amq.direct.\n", queueName);
        return queueName;
    }


    public Measurement readMessage() {
        Console.WriteLine("Receiving message...");
        Measurement m = new Measurement();

        int i = 0;
        foreach (BasicDeliverEventArgs ev in sub) {
            m = Measurement.AltDeSerialize(ev.Body);
            //m.id = //get the id here, from sub
            if (++i == 1)
                break;
            sub.Ack();
        }

        Console.WriteLine("Done.\n");
        return m;
    }


    public void subscribeToQueue(string queueName ) 
    {
        sub = new Subscription(channel, queueName);
    }

    public static string MsgSysName;
    public string MsgSys
    {
        get 
        { 
            return MsgSysName;
        }
        set
        {
            MsgSysName = value;
        }
    }

    public Rabbit(string _msgSys) //Constructor
    {   
        factory = new ConnectionFactory();
        factory.HostName = "localhost"; 
        connection = factory.CreateConnection();
        channel = connection.CreateModel();
        //consumer = new QueueingBasicConsumer(channel);

        System.Console.WriteLine("\nMsgSys: RabbitMQ");
        MsgSys = _msgSys;
    }

    ~Rabbit()
    {
        //observer??
        connection.Dispose();
        //channel.Dispose();
        System.Console.WriteLine("\nDestroying RABBIT");
    }   
}

// Редактировать 2

private List<Subscription> subscriptions = new List<Subscription>();
    Subscription sub = null;

public Measurement readMessage()
    {
        Measurement m = new Measurement();
        foreach(Subscription element in subscriptions)
        {
            foreach (BasicDeliverEventArgs ev in element) {
                //ev = element.Next();
                if( ev != null) {
                    m = Measurement.AltDeSerialize( ev.Body );
                    return m;
                }
                m =  null;  
            }           
        }   
        System.Console.WriteLine("No message in the queue(s) at this time.");
        return m;
    }

    public void subscribeToQueue(string queueName) 
    {   
        sub = new Subscription(channel, queueName);
        subscriptions.Add(sub);     
    }

// Редактировать 3

//MessageHandler.cs

public class MessageHandler
{   
    // Implementation of methods for Rabbit class go here
    private List<string> publishQ = new List<string>();
    private List<string> subscribeQ = new List<string>();

    ConnectionFactory factory = null;
    IConnection connection = null;
    IModel channel = null;  
    QueueingBasicConsumer consumer = null;  

    private List<Subscription> subscriptions = new List<Subscription>();
    Subscription sub = null;

    public void writeMessage ( Measurement m1 )
    {
        byte[] body = Measurement.AltSerialize( m1 );
        //declare a queue if it doesn't exist
        publishToQueue(m1.id);

        channel.BasicPublish("amq.direct", m1.id, null, body);
        Console.WriteLine("\n  [x] Sent to queue {0}.", m1.id);
    }

    public void publishToQueue(string queueName)
    {   
        string finalQueueName = channel.QueueDeclare(queueName, true, false, false, null);
        channel.QueueBind(finalQueueName, "amq.direct", "", null);
    }

    public Measurement readMessage()
    {
        Measurement m = new Measurement();
        foreach(Subscription element in subscriptions)
        {
            if( element.QueueName == null)
            {
                m = null;
            }
            else 
            {
                BasicDeliverEventArgs ev = element.Next();
                if( ev != null) {
                    m = Measurement.AltDeSerialize( ev.Body );
                    m.id = element.QueueName;
                    element.Ack();
                    return m;
                }
                m =  null;                      
            }
            element.Ack();
        }   
        System.Console.WriteLine("No message in the queue(s) at this time.");
        return m;
    }

    public void subscribeToQueue(string queueName) 
    {   
        sub = new Subscription(channel, queueName);
        subscriptions.Add(sub); 
    }

    public static string MsgSysName;
    public string MsgSys
    {
        get 
        { 
            return MsgSysName;
        }
        set
        {
            MsgSysName = value;
        }
    }

    public MessageHandler(string _msgSys) //Constructor
    {   
        factory = new ConnectionFactory();
        factory.HostName = "localhost"; 
        connection = factory.CreateConnection();
        channel = connection.CreateModel();
        consumer = new QueueingBasicConsumer(channel);

        System.Console.WriteLine("\nMsgSys: RabbitMQ");
        MsgSys = _msgSys;
    }

    public void disposeAll()
    {
        connection.Dispose();
        channel.Dispose();
        foreach(Subscription element in subscriptions)
        {
            element.Close();
        }
        System.Console.WriteLine("\nDestroying RABBIT");
    }   
}

// App1.cs

using System;
using System.IO;

using UtilityMeasurement;
using UtilityMessageBus;


public class MainClass
{
    public static void Main()
    {

    MessageHandler obj1 = MessageHandler("Rabbit");

    System.Console.WriteLine("\nA {0} object is now created.", MsgSysName);

    //Create new Measurement messages
    Measurement m1 = new Measurement("q1", 2345, 23.456); 
    Measurement m2 = new Measurement("q2", 222, 33.33);

    System.Console.WriteLine("Test message 1:\n    ID: {0}", m1.id);
    System.Console.WriteLine("    Time: {0}", m1.time);
    System.Console.WriteLine("    Value: {0}", m1.value);

    System.Console.WriteLine("Test message 2:\n    ID: {0}", m2.id);
    System.Console.WriteLine("    Time: {0}", m2.time);
    System.Console.WriteLine("    Value: {0}", m2.value);   

    // Ask queue name and store it
    System.Console.WriteLine("\nName of queue to publish to: ");
    string queueName = (System.Console.ReadLine()).ToString();
    obj1.publishToQueue( queueName );

    // Write message to the queue
    obj1.writeMessage( m1 );    

    System.Console.WriteLine("\nName of queue to publish to: ");
    string queueName2 = (System.Console.ReadLine()).ToString();
    obj1.publishToQueue( queueName2 );

    obj1.writeMessage( m2 );

    obj1.disposeAll();
}
}

// App2.cs

using System;
using System.IO;

using UtilityMeasurement;
using UtilityMessageBus;

public class MainClass
{
    public static void Main()
    {
    //Asks for the message system
    System.Console.WriteLine("\nEnter name of messageing system: ");
    System.Console.WriteLine("Usage: [Rabbit] [Zmq]");
    string MsgSysName = (System.Console.ReadLine()).ToString();

    //Declare an IMessageBus instance:
    //Here, an object of the corresponding Message System
        // (ex. Rabbit, Zmq, etc) is instantiated
    IMessageBus obj1 = MessageBusFactory.GetMessageBus(MsgSysName);

    System.Console.WriteLine("\nA {0} object is now created.", MsgSysName);

    //Create a new Measurement object m
    Measurement m = new Measurement();  

    System.Console.WriteLine("Queue name to subscribe to: ");
    string QueueName1 = (System.Console.ReadLine()).ToString();
    obj1.subscribeToQueue( QueueName1 );

    //Read message into m
    m = obj1.readMessage();

    if (m != null ) {
        System.Console.WriteLine("\nMessage received from queue {0}:\n    ID: {1}", m.id, m.id);
        System.Console.WriteLine("    Time: {0}", m.time);
        System.Console.WriteLine("    Value: {0}", m.value);
    }

    System.Console.WriteLine("Another queue name to subscribe to: ");
    string QueueName2 = (System.Console.ReadLine()).ToString();
    obj1.subscribeToQueue( QueueName2 );

    m = obj1.readMessage();

    if (m != null ) {
        System.Console.WriteLine("\nMessage received from queue {0}:\n    ID: {1}", m.id, m.id);
        System.Console.WriteLine("    Time: {0}", m.time);
        System.Console.WriteLine("    Value: {0}", m.value);
    }

    obj1.disposeAll();
}
}

Answer 1

два источника информации:

1. http://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss

2. Вы действительно должны сначала попытаться понять примеры.

   • % программных файлов% \ RabbitMQ \ DotNetClient \ examples \ src (основные примеры)

   • получить полные рабочие примеры из своего репозитория Mercurial (проекты c #).

Полезные операции для понимания:

• Объявление / утверждение / прослушивание / подписка / публикация

Re: ваш вопрос -- нет никаких причин, почему вы не можете иметь несколько слушателей.Или вы можете подписаться на n путей маршрутизации с одним получателем на обмене.

** re: неблокирующее **

Типичный получатель принимает сообщения по одному за раз.Вы можете вытащить их из очереди, или они будут автоматически размещены рядом с потребителем в «оконном» режиме (определяемом параметрами качества обслуживания Qos).Прелесть этого подхода в том, что для вас проделана большая тяжелая работа (надежность, гарантированная доставка и т. Д.).

Ключевой особенностью RabbitMQ является то, что если при обработке возникает ошибка, то сообщение повторно добавляется обратно в очередь (функция отказоустойчивости).

Необходимо знать большео вашей ситуации.

Часто, если вы публикуете список, о котором я упоминал выше, вы можете найти кого-нибудь из сотрудников RabbitMQ.Они очень полезны.

Надеюсь, это немного поможет.Поначалу много чего стоит задуматься, но с этим стоит продолжать.

---

Q & A

см .: http://www.rabbitmq.com/faq.html

Q.Можно ли подписаться на несколько очередей, используя новую подписку (channel, queueName)?

Да.Вы используете либо ключ привязки, например, abc. *. Hij, либо abc. #. Hij, либо прикрепляете несколько привязок.Первый предполагает, что вы разработали свои ключи маршрутизации на основе какого-то принципа, который имеет смысл для вас (см. Ключи маршрутизации в FAQ).Для последнего вам необходимо привязать к нескольким очередям.

Реализация n-привязок вручную.см .: http://hg.rabbitmq.com/rabbitmq-dotnet-client/file/default/projects/client/RabbitMQ.Client/src/client/messagepatterns/Subscription.cs

за этим шаблоном не так много кода, поэтому вы можете свернуть свой собственный шаблон подписки, если подстановочных знаков недостаточно.вы можете наследовать от этого класса и добавить другой метод для дополнительных привязок ... возможно, это будет работать или что-то похожее (не проверено).

Спецификация AQMP говорит, что возможна множественная ручная привязка: http://www.rabbitmq.com/amqp-0-9-1-reference.html#queue.bind

Q.И если так, как я могу пройти через все подписанные очереди и вернуть сообщение (ноль, когда нет сообщений)?

С подписчиком вы будете уведомлены, когда сообщение будет доступно.В противном случае вы описываете интерфейс извлечения, в котором вы отправляете сообщение по запросу.Если нет доступных сообщений, вы получите нулевое значение.Кстати: метод Notify, вероятно, более удобен.

Q.Ох, и учтите, что у меня все эти операции выполняются разными способами.Я отредактирую свой пост, чтобы отразить код

Живой код:

эта версия должна использовать подстановочные знаки для подписки на более чем один ключ маршрутизации

n Ключи ручной маршрутизации с использованием подписки оставлены в качестве упражнения для читателя.;-) Я думаю, вы все равно склонялись к интерфейсу pull.Интерфейсы btw: pull менее эффективны, чем уведомляющие.

        using (Subscription sub = new Subscription(ch, QueueNme))
        {
            foreach (BasicDeliverEventArgs ev in sub)
            {
                Process(ev.Body);

        ...

Примечание: foreach использует IEnumerable, а IEnumerable переносит событие, что новое сообщение поступило через оператор yield,Фактически это бесконечный цикл.

--- ОБНОВЛЕНИЕ

AMQP был разработан с идеей сохранения количества соединений TCP на уровне приложений, что означает, что вы можете иметьмного каналов на соединение.

код в этом вопросе (правка 3) пытается использовать двух подписчиков с одним каналом, тогда как он должен (я считаю) быть одним подписчиком на канал на поток, чтобы избежать проблем с блокировкой.Sugestion: используйте ключ маршрутизации "подстановочный знак".С помощью java-клиента можно подписаться на несколько разных имен очереди, но, насколько мне известно, клиент .net не реализовал это в вспомогательном классе подписчика.

Если вам действительно нужны два разных имени очереди в одном и том же потоке подписки, то для .net предлагается следующая последовательность извлечения:

        using (IModel ch = conn.CreateModel()) {    // btw: no reason to close the channel afterwards IMO
            conn.AutoClose = true;                  // no reason to closs the connection either.  Here for completeness.

            ch.QueueDeclare(queueName);
            BasicGetResult result = ch.BasicGet(queueName, false);
            if (result == null) {
                Console.WriteLine("No message available.");
            } else {
                ch.BasicAck(result.DeliveryTag, false);
                Console.WriteLine("Message:");
            }

            return 0;
        }

- ОБНОВЛЕНИЕ 2:

из списка RabbitMQ:

"Предположим, что element.Next () блокирует одну из подписок. Вы можете получать доставки из каждой подписки с тайм-аутом для чтения после нее. В качестве альтернативы вы можете настроить однуочередь для получения всех измерений и извлечения из нее сообщений по одной подписке. "(Эмиль)

Это означает, что, когда первая очередь пуста, .Next () блокирует ожидание появления следующего сообщения.т. е. у подписчика есть встроенное сообщение ожидания следующего сообщения.

- ОБНОВЛЕНИЕ 3:

в .net, используйте QueueingBasicConsumer для потребления из нескольких очередей.

На самом деле, есть ветка об этом, чтобы почувствовать использование:

Ожидание одного сообщения RabbitMQ с таймаутом

- UPDATE4:

дополнительная информация о .QueueingBasicConsumer

Здесь приведен пример кода.

http://www.rabbitmq.com/releases/rabbitmq-dotnet-client/v1.4.0/rabbitmq-dotnet-client-1.4.0-net-2.0-htmldoc/type-RabbitMQ.Client.QueueingBasicConsumer.html

пример скопирован в ответ с несколькими изменениями (см. // <----). </p>

                IModel channel = ...;
            QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
            channel.BasicConsume(queueName, false, null, consumer);  //<-----
            channel.BasicConsume(queueName2, false, null, consumer); //<-----
            // etc. channel.BasicConsume(queueNameN, false, null, consumer);  //<-----

            // At this point, messages will be being asynchronously delivered,
            // and will be queueing up in consumer.Queue.

            while (true) {
                try {
                    BasicDeliverEventArgs e = (BasicDeliverEventArgs) consumer.Queue.Dequeue();
                    // ... handle the delivery ...
                    channel.BasicAck(e.DeliveryTag, false);
                } catch (EndOfStreamException ex) {
                    // The consumer was cancelled, the model closed, or the
                    // connection went away.
                    break;
                }
            }

- ОБНОВЛЕНИЕ 5: простое получение, которое будет действовать в любой очереди (более медленный, но иногда более удобный метод).

            ch.QueueDeclare(queueName);
            BasicGetResult result = ch.BasicGet(queueName, false);
            if (result == null) {
                Console.WriteLine("No message available.");
            } else {
                ch.BasicAck(result.DeliveryTag, false);
                Console.WriteLine("Message:"); 
                // deserialize body and display extra info here.
            }

Answer 2

Самый простой способ - использовать EventingBasicConsumer. У меня есть пример на моем сайте о том, как его использовать. RabbitMQ EventingBasicConsumer

Этот класс Consumer предоставляет полученное событие, которое вы можете использовать, и поэтому НЕ блокирует. Остальная часть кода в основном остается прежней.