Чтение сообщения отправлено до начала?

Question

Сервер Erlang RMQ работает без остановок .

Допустим, у меня есть получатель RMQ и отправитель RMQ, который отправляет сообщение с самого начала.

Когда я их запускаю: сначала получатель, потом отправитель, сообщение отправляется, а сервер его забирает.

Но когда я сначала запускаю отправителя и сразу после того, как он отправляет свое сообщение, я запускаю получателя, получатель не видит это сообщение.

Мой вопрос заключается в том, способен ли RMQ обрабатывать (читать) сообщение во втором случае, и если да, каковы требуемые параметры? Обмен уже длительный, но это не помогло.

Answer 1

ДА , RabbitMQ способен хранить эти сообщения, пока ваш потребитель не станет доступным. Вопрос только в том, когда вы объявили очередь и конфигурацию очереди.

Короче говоря; Если вы объявляете и связываете свою очередь до того, как сообщения будут опубликованы, сообщения будут храниться в этой очереди независимо от того, у кого еще нет получателя.

Как объяснено в документации , некоторые параметры влияют на это поведение при объявлении очереди:

• exclusive, если true, удалит очередь (и сообщения), когда потребитель отключится.
• durable, если true, будет хранить сообщения даже после перезапуска RabbitMQ.

Когда потребители автоматически создают и настраивают свои собственные очереди при запуске, ситуация в точности соответствует описанной вами; Сообщения теряются, если потребитель никогда не запускался.

Это довольно распространенная практика, которую вы можете проверить с помощью учебных пособий, особенно в сценарии pub-sub (см. Руководство № 3) , где издатели не знают о потребителях или каких очередях они будут необходимо.

Если вы действительно хотите убедиться, что все сообщения получены, необходимо какое-то средство заранее объявить очереди.

Answer 2

Простой производитель

    // producer
    static void Main( string[] args )
    {
        var factory = new ConnectionFactory() 
        { 
            HostName = "localhost"                
        };
        using ( var connection = factory.CreateConnection() )
        {
            using ( var channel = connection.CreateModel() )
            {
                channel.QueueDeclare( "hello", true, false, false, null );

                string message = "Hello World!";
                var body = Encoding.UTF8.GetBytes( message );

                while ( true )
                {
                    Console.WriteLine( "Press key to send message" );
                    Console.ReadLine();

                    channel.BasicPublish( "", "hello", null, body );

                    Console.WriteLine( " [x] Sent {0}", message );
                }
            }
        }

Я запускаю это, публикую несколько сообщений и вижу их всех в очереди на RMQ.

Затем я запускаю этого простого потребителя

    // consumer
    static void Main( string[] args )
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare("hello", true, false, false, null);

                var consumer = new QueueingBasicConsumer(channel);
                channel.BasicConsume("hello", true, consumer);

                Console.WriteLine(" [*] Waiting for messages." +
                                         "To exit press CTRL+C");
                while (true)
                {
                    var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                    var body    = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine(" [x] Received {0}", message);
                }
            }
        }
    }

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

Пожалуйста, проверьте, как это отличается от вашей настройки.

Редактировать: Это также работает, если существует обмен, поддерживающий очередь в производителе:

    // producer
    static void Main( string[] args )
    {
        var factory = new ConnectionFactory() 
        { 
            HostName = "localhost"                
        };
        using ( var connection = factory.CreateConnection() )
        {
            using ( var channel = connection.CreateModel() )
            {
                channel.QueueDeclare( "hello", true, false, false, null );
                channel.ExchangeDeclare( "helloe", "fanout", true );
                channel.QueueBind( "hello", "helloe", "" );

                string message = "Hello World!";
                var body = Encoding.UTF8.GetBytes( message );

                while ( true )
                {
                    Console.WriteLine( "Press key to send message" );
                    Console.ReadLine();

                    channel.BasicPublish( "helloe", "", null, body );

                    Console.WriteLine( " [x] Sent {0}", message );
                }
            }
        }

        Console.WriteLine( "finished" );
        Console.ReadLine();
    }

(изменения для потребителя не требуются)

Редактировать 2 : это также работает при использовании потребителя событий:

    static void Main( string[] args )
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare("hello", true, false, false, null);

                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (s, e) =>
                {
                    var message = Encoding.UTF8.GetString(e.Body);
                    Console.WriteLine(" [x] Received {0}", message);
                };
                channel.BasicConsume("hello", true, consumer);

                Console.WriteLine(" [*] Waiting for messages." +
                                         "To exit press CTRL+C");

                Console.ReadLine();
            }
        }
    }