Question about session usage / link allocation in RabbitMQ.AMQP.Client (.NET, AMQP 1.0)

[allenyo]

Hi, I’m using RabbitMQ.AMQP.Client in a banking system (ASP.NET Core, .NET 8) and I’m trying to understand how sessions are actually used by the client library.

From the docs and AMQP 1.0 model I expect something like “one connection → some sessions → many links”, but in practice I’m seeing different behavior in the RabbitMQ management UI depending on startup vs reconnect.

What I’m doing

I have a small wrapper around the client:

• I create N connections (configurable) via AmqpEnvironment.Create(settings) and CreateConnectionAsync().
• For publishers, I have a PublisherPool per connection that builds publishers with
  connection.PublisherBuilder().Exchange(...).Key(...).BuildAsync().
  (Pool is keyed by (exchange, routingKey) and reuses publishers.)
• For consumers, I have a RabbitMqConsumerManager that builds one consumer per queue via
  connection.ConsumerBuilder().Queue(...).MessageHandler(...).BuildAndStartAsync().

How I actually use the consumers

Every queue in the system is handled by a dedicated BackgroundService (derived from a base message-processing service).
Each background service calls manager.GetConsumerAsync(queue, handler) once on startup, obtains a single long-lived IConsumer, and keeps it alive for the lifetime of the process.
Message processing includes accept/discard/retry logic, and retry/dead-letter messages are published using the same publisher pool.
So effectively I have one consumer link per queue, created once, running continuously.

Relevant pieces

public sealed class RabbitMqConnectionManager : IAsyncDisposable
{
    private readonly IEnvironment _environment;
    private readonly RabbitLinkPoolOptions _poolOptions;
    private IConnection[]? _connections;
    private PublisherPool[]? _pubPools;
    private readonly RabbitMqConsumerManager _consumerManager;

    public RabbitMqConnectionManager(
        IOptions<RabbitHostOption> rabbitOptions,
        IOptions<RabbitLinkPoolOptions> poolOptions,
        ILogger<RabbitMqConnectionManager> logger)
    {
        _poolOptions = poolOptions.Value;
        var options = rabbitOptions.Value;

        var recovery = new RecoveryConfiguration()
            .Activated(true)
            .Topology(true)
            .BackOffDelayPolicy(new BackOffDelayPolicy());

        var uris = options.Hosts!
            .Select(h => new Uri($"{options.Scheme}://{options.Username}:{options.Password}@{h.Host}:{h.Port}/{options.VirtualHost}"))
            .ToList();

        var settings = ConnectionSettingsBuilder.Create()
            .Uris(uris)
            .RecoveryConfiguration(recovery)
            .ContainerId(options.ContainerId ?? "RMQ")
            .SaslMechanism(SaslMechanism.Plain)
            .Build();

        _environment = AmqpEnvironment.Create(settings);
        _consumerManager = new RabbitMqConsumerManager();
    }

    public async Task InitializeAsync()
    {
        var n = Math.Max(1, _poolOptions.ConnectionCount);
        _connections = new IConnection[n];

        for (int i = 0; i < n; i++)
            _connections[i] = await _environment.CreateConnectionAsync();

        _pubPools = new PublisherPool[n];
        for (int i = 0; i < n; i++)
            _pubPools[i] = new PublisherPool(
                _connections[i],
                _poolOptions.MaxOpenRoutesPerConnection,
                _poolOptions.RouteIdleEvictAfter);

        _consumerManager.Initialize(_connections[0]);
    }

    public Task<IConsumer> GetConsumerAsync(string queue, MessageHandler handler)
        => _consumerManager.GetOrStartConsumerAsync(queue, handler);

    public async Task<IPublisher> GetPublisherAsync(string exchange, string routingKey, CancellationToken ct = default)
    {
        var pools = _pubPools ?? throw new InvalidOperationException("Manager not initialized.");
        var index = PickPoolIndex(exchange, routingKey);
        return await pools[index].GetAsync(exchange, routingKey, ct);
    }

    private int PickPoolIndex(string exchange, string routingKey)
    {
        var h = unchecked(exchange.GetHashCode() * 397 ^ routingKey.GetHashCode());
        var pools = _pubPools ?? throw new InvalidOperationException("Manager not initialized.");
        return (int)((uint)h % (uint)pools.Length);
    }

    public IConnection GetConnection()
    {
        if (_connections is null || _connections.Length == 0)
            throw new InvalidOperationException("RabbitMQ not initialized. Call InitializeAsync() first.");
        return _connections[0];
    }

    // DisposeAsync omitted for brevity
}

Publisher pool

public sealed class PublisherPool : IAsyncDisposable
{
    private readonly IConnection _connection;
    private readonly int _maxRoutes;
    private readonly ConcurrentDictionary<(string ex, string key), Entry> _senders = new();
    private readonly ConcurrentQueue<(string ex, string key)> _lru = new();
    private volatile bool _disposed;

    private sealed class Entry
    {
        public Lazy<Task<IPublisher>> Lazy;
        public long LastTicks;

        public Entry(Lazy<Task<IPublisher>> lazy)
        {
            Lazy = lazy;
            Touch();
        }

        public void Touch() => LastTicks = Stopwatch.GetTimestamp();
    }

    public PublisherPool(IConnection connection, int maxRoutes, TimeSpan idleEvictAfter)
    {
        _connection = connection;
        _maxRoutes = Math.Max(1, maxRoutes);
        // idle eviction currently disabled (I may enable later)
    }

    public async Task<IPublisher> GetAsync(string exchange, string routingKey, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        var key = (exchange, routingKey);

        if (!_senders.ContainsKey(key) && _senders.Count >= _maxRoutes)
            EvictOneRoute();

        var entry = _senders.GetOrAdd(
            key,
            _ => new Entry(new Lazy<Task<IPublisher>>(
                () => CreateAsync(exchange, routingKey, ct)))
        );

        var publisher = await entry.Lazy.Value.ConfigureAwait(false);
        entry.Touch();
        _lru.Enqueue(key);

        return publisher;
    }

    private Task<IPublisher> CreateAsync(string exchange, string routingKey, CancellationToken ct)
        => _connection.PublisherBuilder()
            .Exchange(exchange)
            .Key(routingKey)
            .BuildAsync(ct);

    private void EvictOneRoute()
    {
        while (_lru.TryDequeue(out var old))
        {
            if (_senders.TryRemove(old, out var entry))
            {
                if (entry.Lazy.IsValueCreated)
                    _ = SafeCloseAsync(entry.Lazy.Value);
                return;
            }
        }
    }

    private async Task SafeCloseAsync(Task<IPublisher> t)
    {
        try
        {
            var p = await t.ConfigureAwait(false);
            try { await p.CloseAsync().ConfigureAwait(false); } catch { }
            p.Dispose();
        }
        catch { }
    }

    private void ThrowIfDisposed()
    {
        if (_disposed) throw new ObjectDisposedException(nameof(PublisherPool));
    }

    public async ValueTask DisposeAsync()
    {
        if (_disposed) return;
        _disposed = true;

        foreach (var entry in _senders.Values)
        {
            if (!entry.Lazy.IsValueCreated) continue;
            await SafeCloseAsync(entry.Lazy.Value).ConfigureAwait(false);
        }

        _senders.Clear();
        GC.SuppressFinalize(this);
    }
}

ConsumerManager

public sealed class RabbitMqConsumerManager : IAsyncDisposable
{
    private readonly ConcurrentDictionary<string, Lazy<Task<IConsumer>>> _consumers = new();
    private IConnection? _connection;
    private bool _disposed;

    public void Initialize(IConnection connection) =>
        _connection = connection ?? throw new ArgumentNullException(nameof(connection));

    public async Task<IConsumer> GetOrStartConsumerAsync(string queue, MessageHandler handler)
    {
        if (_disposed) throw new ObjectDisposedException(nameof(RabbitMqConsumerManager));

        var lazy = _consumers.GetOrAdd(queue,
            _ => new Lazy<Task<IConsumer>>(() => CreateAsync(queue, handler)));

        try
        {
            return await lazy.Value.ConfigureAwait(false);
        }
        catch
        {
            _consumers.TryRemove(queue, out _);
            throw;
        }
    }

    private Task<IConsumer> CreateAsync(string queue, MessageHandler handler) =>
        _connection!.ConsumerBuilder()
            .Queue(queue)
            .MessageHandler(async (context, message) =>
            {
                await handler(context, message).ConfigureAwait(false);
            })
            .BuildAndStartAsync();

    public async ValueTask DisposeAsync()
    {
        if (_disposed) return;
        _disposed = true;

        foreach (var lazy in _consumers.Values)
        {
            if (!lazy.IsValueCreated) continue;

            try
            {
                var c = await lazy.Value.ConfigureAwait(false);
                try { await c.CloseAsync().ConfigureAwait(false); } catch { }
                c.Dispose();
            }
            catch { }
        }

        _consumers.Clear();
        GC.SuppressFinalize(this);
    }
}

So effectively:

• A small fixed number of connections (ConnectionCount).
• Many publishers.
• Many long-lived consumers.
• All built on top of the same few connections.

What I see in RabbitMQ UI

On first startup of the API:

• Each consumer (incoming link) appears to be using its own session.
• Each publisher (outgoing link) also seems to be attached in a way that suggests multiple sessions.

Then, if I:

• stop the app, or
• close/reopen the connection on the broker side,
  and let recovery / reconnect kick in, I often end up with:
• all links under a single session per connection.

So from the outside it looks like:

initial startup: one connection → many sessions → links
after reconnect: one connection → one session → all links

This might be me misreading the UI, but it’s consistent.

Questions

1. How does the library decide when to create a new session?

• Does ConsumerBuilder().BuildAndStartAsync() create (or share) a session internally?
• Does PublisherBuilder().BuildAsync() create a new session per publisher, or do all publishers share a single session per connection?
• Is there any documented rule like “one session per connection” vs “one session per builder”?

2. Why do I sometimes see “one session per link” and later “one session for all links”?

Is this expected behavior with recovery + negotiation?

• Does the client optimize sessions on reconnect?
• Or is this just how RabbitMQ reports them after recovery?

3. Max sessions per connection

• If the broker’s max sessions per connection limit is reached:
  • Will link creation fail?
  • Does the client reuse sessions automatically?
• In my setup (few connections, many consumers/publishers), is the client already trying to keep session count low?

4. Recommended usage pattern

For long-lived services with many consumers and publishers on a few connections, what is the intended pattern?

• “few connections, few sessions, many links”?
• or “few connections, multiple sessions, each with a subset of links”?

If explicit control of session count is recommended, how should it be done with the current API?

I want to make sure I’m not unintentionally creating too many sessions and that I’m interpreting the management UI correctly.
If needed I can provide a minimal repro, but the snippets above are already close to real code.

Thanks in advance.

[Gsantomaggio]

Thank you, @allenyo, will check it soon.

[Gsantomaggio]

@allenyo I tried our HAClient example with these changes, just to add more producers and consumers:

diff --git a/docs/Examples/HAClient/Program.cs b/docs/Examples/HAClient/Program.cs
index bc0c3e9..ce4ce75 100644
--- a/docs/Examples/HAClient/Program.cs
+++ b/docs/Examples/HAClient/Program.cs
@@ -56,6 +56,9 @@
 await queueSpec.DeclareAsync();
 
 IPublisher publisher = await connection.PublisherBuilder().Queue(queueName).BuildAsync();
+IPublisher publisher1 = await connection.PublisherBuilder().Queue(queueName).BuildAsync();
+IPublisher publisher2 = await connection.PublisherBuilder().Queue(queueName).BuildAsync();
+IPublisher publisher3 = await connection.PublisherBuilder().Queue(queueName).BuildAsync();
 
 AsyncManualResetEvent pausePublishing = new(true);
 publisher.ChangeState += (sender, fromState, toState, e) =>
@@ -80,6 +83,31 @@
     }
 ).BuildAndStartAsync();
 
+IConsumer consumer1 = await connection.ConsumerBuilder().Queue(queueName).InitialCredits(100).MessageHandler((context, message) =>
+    {
+        Interlocked.Increment(ref messagesReceived);
+        context.Accept();
+        return Task.CompletedTask;
+    }
+).BuildAndStartAsync();
+
+
+IConsumer consumer2 = await connection.ConsumerBuilder().Queue(queueName).InitialCredits(100).MessageHandler((context, message) =>
+    {
+        Interlocked.Increment(ref messagesReceived);
+        context.Accept();
+        return Task.CompletedTask;
+    }
+).BuildAndStartAsync();
+
+IConsumer consumer4 = await connection.ConsumerBuilder().Queue(queueName).InitialCredits(100).MessageHandler((context, message) =>
+    {
+        Interlocked.Increment(ref messagesReceived);
+        context.Accept();
+        return Task.CompletedTask;
+    }
+).BuildAndStartAsync();
+
 consumer.ChangeState += (sender, fromState, toState, e) =>
 {
     Trace.WriteLine(TraceLevel.Information, $"Consumer State Changed, from {fromState} to {toState}");

That's when the client starts:

That's when I kill the connection:

That's when I restart the node:

I am unable to reproduce your situation. If you can reproduce it with the HA client, it would be great.

The idea, at the moment, is to have --> one connection --> one session, --> multiple links! but we can ghange it

[allenyo]

Thanks for checking this. I’ll test this using the HA client and compare it with my app’s behavior. If I can reproduce the multi-session case, I’ll update the thread.

[allenyo]

Hey, I reproduced it and now I see what’s causing it.

With the Microsoft.Extensions.Hosting version 9.x BackgroundService behavior you didn’t see this, but in the latest versio hosting model (using Task.Run) when several consumers start concurrently, they can create multiple sessions on one connection instead of a single shared session.

What changed on my side
I upgraded Microsoft.Extensions.Hosting from 9..x to 10.0.0.
The key difference is in BackgroundService.StartAsync:
(9.x) – runs ExecuteAsync inline:

public virtual Task StartAsync(CancellationToken cancellationToken)
{
    _stoppingCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
    _executeTask = ExecuteAsync(_stoppingCts.Token);
    if (_executeTask.IsCompleted)
    {
        return _executeTask;
    }

    return Task.CompletedTask;
}

Latest (10.0.0) – always uses Task.Run:

public virtual Task StartAsync(CancellationToken cancellationToken)
{
    _stoppingCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
    _executeTask = Task.Run(() => ExecuteAsync(_stoppingCts.Token), _stoppingCts.Token);
    return Task.CompletedTask;
}

So with Hosting 10.x, each BackgroundService runs its ExecuteAsync on the thread pool via Task.Run, which means the consumers begin starting concurrently.

What happens with RabbitMQ.AMQP.Client

In my real app I have:

few connections (1–N),

many long-lived consumers,

many publishers pooled on top.

On Hosting 9..x:

I see one AMQP session per connection and many links under it (what I expected).

On Hosting 10.x (same code, same config):

I see that when consumers start concurrently, multiple sessions may be created on the same connection. It’s not always a 1:1 ratio — for example, in one service with 7 consumers I consistently observed 7 sessions, while in another setup with 70 consumers I saw only 24 sessions — but the key point is that multiple sessions appear instead of a single shared one.

After reconnect/recovery, it often collapses back to one session per connection again.

To be sure, I added a small SemaphoreSlim around consumer creation in my RabbitMqConsumerManager:

private static readonly SemaphoreSlim _consumerCreateLock = new(1, 1);

public async Task<IConsumer> GetOrStartConsumerAsync(string queue, MessageHandler handler)
{
    if (_disposed) throw new ObjectDisposedException(nameof(RabbitMqConsumerManager));

    var lazy = _consumers.GetOrAdd(queue,
        _ => new Lazy<Task<IConsumer>>(async () =>
        {
            await _consumerCreateLock.WaitAsync().ConfigureAwait(false);
            try
            {
                return await CreateAsync(queue, handler).ConfigureAwait(false);
            }
            finally
            {
                _consumerCreateLock.Release();
            }
        }));

    try
    {
        return await lazy.Value.ConfigureAwait(false);
    }
    catch
    {
        _consumers.TryRemove(queue, out _);
        throw;
    }
}

With this in place:

I always get exactly one session per connection, regardless of Hosting version.

Without the lock, Hosting 9.x “looks fine”, Hosting 10.x creates multiple sessions.

So it really looks like session creation is not thread-safe: if multiple links are built concurrently on the same IConnection, multiple sessions can be created.

Minimal repro based on HAClient

I took your HAClient sample and reproduced it like this:

Using background service as consumers

const int ConsumerCount = 8;
const string QueueName = "ha-amqp10-client-test-many-sessions";
const string ContainerId = "HA-Client-Connection-HOSTING-10-MULTI";

Trace.TraceLevel = TraceLevel.Information;
ConsoleTraceListener consoleListener = new();
Trace.TraceListener = (l, f, a) =>
{
    consoleListener.WriteLine($"[{DateTime.Now}] [{l}] - {f}");
};

var builder = Host.CreateDefaultBuilder(args);

builder.ConfigureLogging(logging =>
{
    logging.ClearProviders();
    logging.AddSimpleConsole(o =>
    {
        o.SingleLine = true;
        o.TimestampFormat = "HH:mm:ss ";
    });
});

builder.ConfigureServices(services =>
{
    services.AddSingleton<IEnvironment>(sp =>
    {
        var settings = ConnectionSettingsBuilder.Create()
            .Uris(new[]
            {
                new Uri("amqp://guest:guest@localhost:5672/vh")
            })
            .ContainerId(ContainerId)
            .SaslMechanism(SaslMechanism.Plain)
            .Build();

        return AmqpEnvironment.Create(settings);
    });

    services.AddSingleton<IConnection>(sp =>
    {
        var env = sp.GetRequiredService<IEnvironment>();
        return env.CreateConnectionAsync().Result;
    });

    for (int i = 0; i < ConsumerCount; i++)
    {
        int index = i;

        services.AddSingleton<IHostedService>(sp =>
        {
            return new ConsumerHostedService(
                index,
                QueueName,
                sp.GetRequiredService<IConnection>(),
                sp.GetRequiredService<ILogger<ConsumerHostedService>>());
        });
    }
});

using IHost host = builder.Build();

await host.RunAsync();

public sealed class ConsumerHostedService : BackgroundService
{
    private readonly int _index;
    private readonly string _queueName;
    private readonly IConnection _connection;
    private readonly ILogger<ConsumerHostedService> _logger;
    private IConsumer? _consumer;

    public ConsumerHostedService(
        int index,
        string queueName,
        IConnection connection,
        ILogger<ConsumerHostedService> logger)
    {
        _index = index;
        _queueName = queueName;
        _connection = connection;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {

        _consumer = await _connection.ConsumerBuilder()
            .Queue(_queueName)
            .InitialCredits(100)
            .MessageHandler((context, message) =>
            {
                context.Accept();
                return Task.CompletedTask;
            })
            .BuildAndStartAsync();

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await Task.Delay(500, stoppingToken);
            }
            catch (TaskCanceledException)
            {
                break;
            }
        }
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        await base.StopAsync(cancellationToken);

        if (_consumer != null)
        {
            try
            {
                await _consumer.CloseAsync();
            }
            catch
            {
            }
            _consumer.Dispose();
        }
    }
}

With Microsoft.Extensions.Hosting 9.x this code produces only one AMQP session, but with version 10.0.0 the very same code ends up creating multiple sessions instead of just one.

multiple sessions (Microsoft.Extensions.Hosting 10.x)

single session (Microsoft.Extensions.Hosting 9.x)

The intended behavior is “few connections, few sessions, many links” (and the code internally suggests there’s a maxSessionsPerConnection = 1-style logic).

So it looks like the session manager may need internal locking to guarantee a single shared session whenever multiple links are created on the same connection, even if they start in parallel.

**Based on these tests, it looks like the difference comes from when the first link is created.

If a connection has no session yet and several links are created in parallel, each of them may end up getting its own session.
If at least one session already exists, or links are created sequentially, they all attach to the same session.

So this doesn’t necessarily mean the behavior is wrong — it may simply be a consequence of how session allocation currently works internally, since session creation does not appear to be synchronized. But it does explain why:

with sequential startup (or older hosting) there is one session

with concurrent startup there can be multiple sessions

and why this is easy to reproduce without BackgroundService by parallel link creation on a fresh connection.**

In example just.

var publisherTasks = new[]
{
    Task.Run(async () =>
    {
        publisher1 = await connection.PublisherBuilder()
            .Queue(queueName)
            .BuildAsync();
    }),
    Task.Run(async () =>
    {
        publisher2 = await connection.PublisherBuilder()
            .Queue(queueName)
            .BuildAsync();
    }),
    Task.Run(async () =>
    {
        publisher3 = await connection.PublisherBuilder()
            .Queue(queueName)
            .BuildAsync();
    })
};

await Task.WhenAll(publisherTasks);

In my app I start several consumers first, and with the newer hosting model each BackgroundService runs ExecuteAsync via Task.Run, so those consumers are created in parallel. If there is no existing AMQP session yet, these concurrent consumer creations can each trigger their own session instead of all of them sharing a single one.

[Gsantomaggio]

@allenyo can you please try #139 ?

[allenyo]

@Gsantomaggio Tested #139 with my real application. Even when several consumers start in parallel, I now consistently see only one session per connection. Works great — thanks for the quick fix!