AMQP .NET Client
The recommended .NET client is the RabbitMQ .NET client. Version 7 has a fully async API, and it recovers connections, topology and consumers on its own. Install it with:
dotnet add package RabbitMQ.Client
Publish and consume
The example reads the broker URL from BROKER_URL and falls back to a local LavinMQ at amqp://guest:guest@localhost.
The project file is in the example repo.
Save the connection helper as Amqp.cs:
using RabbitMQ.Client;
namespace AmqpExample;
public static class Amqp
{
// Automatic recovery is on by default: the connection, channels and consumers are restored after a network failure
public static Task<IConnection> ConnectAsync()
{
var url = Environment.GetEnvironmentVariable("BROKER_URL") ?? "amqp://guest:guest@localhost";
var factory = new ConnectionFactory { Uri = new Uri(url) };
return factory.CreateConnectionAsync();
}
}
Save the publisher as Publisher.cs:
using System.Text;
using RabbitMQ.Client;
namespace AmqpExample;
public static class Publisher
{
// With confirmation tracking, BasicPublishAsync returns once the broker has confirmed the message.
// This opens a channel per call; when publishing many messages, create the channel once and reuse it.
public static async Task PublishAsync(IConnection connection, string queue, string message)
{
var options = new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true);
await using var channel = await connection.CreateChannelAsync(options);
await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false);
var properties = new BasicProperties { Persistent = true };
await channel.BasicPublishAsync("", queue, mandatory: true, properties, Encoding.UTF8.GetBytes(message));
Console.WriteLine($" [x] Sent '{message}'");
}
}
Save the consumer as Worker.cs:
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace AmqpExample;
public static class Worker
{
// The message is acked after the handler has processed it
public static async Task ConsumeAsync(IChannel channel, string queue, Func<string, Task> handler)
{
await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false);
await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 10, global: false);
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (_, delivery) =>
{
await handler(Encoding.UTF8.GetString(delivery.Body.Span));
await channel.BasicAckAsync(delivery.DeliveryTag, multiple: false);
};
await channel.BasicConsumeAsync(queue, autoAck: false, consumer);
}
}
Save the entry point as Program.cs in the same directory:
using AmqpExample;
await using var connection = await Amqp.ConnectAsync();
switch (args.FirstOrDefault())
{
case "publish":
await Publisher.PublishAsync(connection, "hello", "Hello world!");
break;
case "worker":
await using (var channel = await connection.CreateChannelAsync())
{
await Worker.ConsumeAsync(channel, "hello", message =>
{
Console.WriteLine($" [x] Received '{message}'");
return Task.CompletedTask;
});
Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");
var exit = new TaskCompletionSource();
Console.CancelKeyPress += (_, e) => { e.Cancel = true; exit.TrySetResult(); };
await exit.Task;
}
break;
default:
Console.Error.WriteLine("Usage: AmqpExample publish|worker");
return 1;
}
return 0;
Run dotnet run --project AmqpExample -- worker in one terminal and dotnet run --project AmqpExample -- publish in another.
The full, runnable example lives in dotnet/rabbitmq-client in cloudamqp/examples, and CI tests it against LavinMQ.
Good to know
- The worker acks after the handler returns. If the handler throws, the client reports the exception on the channel’s
CallbackExceptionAsyncevent and the message stays unacked. LavinMQ redelivers it when the channel closes, so keep handlers idempotent. - Set a prefetch limit. The client sends no prefetch on its own, so LavinMQ pushes messages without limit. The worker calls
BasicQosAsyncwith a prefetch of 10, and recovery reapplies it. - Recovery is on by default. After a network failure the client retries every 5 seconds, then redeclares exchanges, queues and bindings and restarts consumers.
- Use
amqps://for TLS. The client verifies the server certificate, but anamqps://URI makes it accept a hostname mismatch. SetSsl.AcceptablePolicyErrors = SslPolicyErrors.Noneafter settingUrito check the hostname too.
The RabbitMQ .NET client API guide covers publisher confirms, consumer concurrency and the full API.
Other .NET clients
- EasyNetQ, a higher-level API built on the RabbitMQ .NET client
- MassTransit, a distributed application framework with a RabbitMQ transport
Ready to take the next steps?
Managed LavinMQ instance via CloudAMQP
LavinMQ has been built with performance and ease of use in mind - we've benchmarked a throughput of about 1,000,000 messages/sec. You can try LavinMQ without any installation hassle by creating a free instance on CloudAMQP. Signing up is a breeze.
Get started with CloudAMQP ->
Help and feedback
We welcome your feedback and are eager to address any questions you may have about this piece or using LavinMQ. Join our Slack channel to connect with us directly. You can also find LavinMQ on GitHub.