Simple RabbitMQ Consumer Implementation in C#

The article introduces a straightforward RabbitMQ consumer implementation, where the consumer continuously listens to the message broker via a long‑lived TCP connection and processes incoming messages through a user‑defined callback.

/// /// 消息消费者 /// public class RabbitMqSubscriber : Lind.DDD.Commons.DisposableBase { private readonly string exchangeName; private readonly string queueName; private readonly IConnection connection; private readonly IModel channel; private bool disposed; /// /// 从消息服务器拉到消息后触发 /// public event EventHandler<MessageReceivedEventArgs> MessageReceived; /// /// Initializes a new instance of RabbitMqMessageSubscriber class. /// /// /// /// public RabbitMqSubscriber(string uri, string queueName, string userName = "", string password = "") { this.exchangeName = exchangeName; this.queueName = queueName; var factory = new ConnectionFactory() { Uri = uri }; if (!string.IsNullOrWhiteSpace(userName)) factory.UserName = userName; if (!string.IsNullOrWhiteSpace(password)) factory.Password = password; this.connection = factory.CreateConnection(); this.channel = connection.CreateModel(); } public void Subscribe() { channel.QueueDeclare( queue: this.queueName, durable: false, exclusive: false, autoDelete: false, arguments: null); var consumer = new EventingBasicConsumer(channel); consumer.Received += (sender, e) => { var body = e.Body; var json = Encoding.UTF8.GetString(body); var message = JsonConvert.DeserializeObject(json, new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All }); this.OnMessageReceived(new MessageReceivedEventArgs(message)); channel.BasicAck(e.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: queueName, noAck: false, consumer: consumer); } private void OnMessageReceived(MessageReceivedEventArgs e) { this.MessageReceived?.Invoke(this, e); } protected override void Finalize(bool disposing) { if (disposing) { if (!disposed) { this.channel.Dispose(); this.connection.Dispose(); disposed = true; } } } }

The usage example shows how to instantiate the subscriber, attach a handler, start listening, and keep the console alive:

class Program { static void Main(string[] args) { var subscriber = new Lind.DDD.RabbitMq.RabbitMqSubscriber("amqp://localhost:5672", "zzl"); subscriber.MessageReceived += Subscriber_MessageReceived; subscriber.Subscribe(); Console.ReadKey(); } private static void Subscriber_MessageReceived(object sender, RabbitMq.MessageReceivedEventArgs e) { Console.WriteLine("消费者2->消费了一个消息{0}", e.Message); Lind.DDD.Logger.LoggerFactory.Instance.Logger_Debug("消费者2->消费了一个消息{0}" + e.Message); Thread.Sleep(2000); } }

After the code, the article explains RabbitMQ’s exchange models. It describes the four exchange types—direct, fanout, topic, and header—focusing on the first three which are most commonly used. For each type it outlines routing behavior, the need for a routing key, and binding requirements.

Finally, the article includes diagrams illustrating the message flow and exchange mechanisms, helping readers visualize how messages are routed from producers to queues based on the selected exchange type.