Мониторинг сообщений в RabbitMQ

Рассмотрим классическую схему построения конвейера сообщений в RabbitMQ состоящую из элементов Producer, Exchange, Queue и Consumer.

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

Прежде всего организуем демонстрационный стенд, для этого внесем следующие изменения в работу конвейера:

Изначально для Exchange (faust) была задана следующая конфигурация, которая не меняется в рассматриваем примере при модификации:

Важным является настройка типа fanaut — которая позволяет создать две равноправные очереди и дублировать весь поток сообщений в новую очередь Statistics:

без какого бы то ни-было вмешательства в основной процесс в очереди Logs. Приступим к обработке потока сообщений. Прежде всего создаем таблицу на MS SQL сервере для хранения статистической информации. Вы можете использовать любой другой подход, например, сохранить сообщения в файл в xml формате или любой другой способ, в рассматриваемом примере выбран SQL сервер для того чтобы избежать дополнительного программирования

create table RabbitMsg(
id int PRIMARY KEY IDENTITY(1000,1),
[Message] nvarchar(1000) DEFAULT '',
RegDate datetime default GETDATE())

Как видно из SQL запроса это таблица с номером записи, некоторым текстом и датой вставки записи в таблицу.

Создадим клиента, который будет обращаться к RabbitMQ в очередь Statistics, забирать полученные данные и перекладывать их в таблицу RabbitMsg

using System;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Data.SqlClient;
namespace Getter
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "192.168.1.241", Port = 30672, UserName = "robotics01", Password = "" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: "faust", type: "fanout", durable: true);
                var queueName  = "Statistics";
                channel.QueueBind(queue: queueName, exchange: "faust", routingKey: "");
                Console.WriteLine(" [*] Waiting for logs.");
                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine(" [x] {0}", message);
                    SqlConnection sqlconnection = new SqlConnection("Server=tcp:fastreportsql,1433;Initial Catalog=FastReportSQL;Persist Security Info=False;User ID=ufocombat;Password=;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=False;Connection Timeout=30;");
                    sqlconnection.Open();
                    SqlCommand cmd = new SqlCommand($"INSERT INTO RabbitMsg(Message) VALUES (@msg)", sqlconnection);
                    cmd.Parameters.AddWithValue("msg", message);
                    cmd.ExecuteNonQuery();
                    sqlconnection.Close();
                };
                channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
                Console.WriteLine(" Press [enter] to exit.");
                Console.ReadLine();
            }
            Console.WriteLine("Hello World!");
        }
    }
} 

Посмотрим, как это работает в режиме реального времени

https://youtu.be/xf95Qvx5EUQ
Тем временем на MS SQL Server

Построим отчет по данным очереди Statistics

Вот что получилось:

Заключение

В рассмотренном примере показано как быстро собрать статистику и даже построить отчет, который можно сохранить в PDF и отправить по почте по данным конвейера RabbitMQ и дополнительной очереди. Легко придумать примеры задач, когда информация собирается о каких-либо процессах и строятся отчеты без разработки серверной части. Учитывая, что FastReports предлагает open-source версию, то можно значительно сократить стоимость разработки без дополнительных затрат. Сам конвейер, также легко перенастраивается и может быть приспособлен для более сложных задач.