Message monitoring in RabbitMQ

Consider the classic RabbitMQ message pipeline construction scheme consisting of Producer, Exchange, Queue and Consumer elements.



The task is to organize monitoring of what is happening in the queue and not to affect the main software (software), add a flexible opportunity to generate reports and at the same time avoid additional costs. The final design should allow you to quickly build reports to analyze the data flow on the pipeline without using the main server capacity (which will avoid additional load) and basic software. The approach should be easily portable to more complex architectures.

First of all, we will organize a demonstration stand, for this we will make the following changes to the work of the conveyor:
')


Initially, the following configuration was set for Exchange (faust), which does not change in this example when modified:



It is important to configure the type fanaut - which allows you to create two equal queues and duplicate the entire message flow to the new Statistics queue:





without interfering with the main process in the Logs queue. We proceed to the processing of the message flow. First of all, we create a table in MS SQL server for storing statistical information. You can use any other approach, for example, save messages to a file in xml format or any other method, in this example SQL server is selected in order to avoid additional programming.

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

As can be seen from the SQL query, this is a table with the record number, some text, and the date the record was inserted into the table.

Create a client who will access RabbitMQ in the Statistics queue, take the data and transfer it to the RabbitMsg table

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

Let's see how this works in real time.


Meanwhile on MS SQL Server



Let's build a report according to the Statistics queue


Here's what happened:

Conclusion

The above example shows how to quickly collect statistics and even build a report that can be saved in PDF and sent by mail according to the RabbitMQ pipeline and an additional queue. It is easy to come up with examples of tasks when information is collected about any processes and reports are built without developing a server part. Given that FastReports offers an open-source version, it is possible to significantly reduce development costs without additional costs. The conveyor itself is also easily reconfigured and can be adapted for more complex tasks.