diff --git a/src/Dafda.Tests/Builders/ProducerFactoryBuilder.cs b/src/Dafda.Tests/Builders/ProducerFactoryBuilder.cs
deleted file mode 100644
index df22c0f9..00000000
--- a/src/Dafda.Tests/Builders/ProducerFactoryBuilder.cs
+++ /dev/null
@@ -1,12 +0,0 @@
-using Dafda.Producing;
-
-namespace Dafda.Tests.Builders
-{
-    internal class ProducerFactoryBuilder
-    {
-        public ProducerFactory Build()
-        {
-            return new ProducerFactory();
-        }
-    }
-}
\ No newline at end of file
diff --git a/src/Dafda.Tests/Configuration/TestServiceCollectionExtensions.cs b/src/Dafda.Tests/Configuration/TestServiceCollectionExtensions.cs
index ad3a7f21..93a399a8 100644
--- a/src/Dafda.Tests/Configuration/TestServiceCollectionExtensions.cs
+++ b/src/Dafda.Tests/Configuration/TestServiceCollectionExtensions.cs
@@ -1,3 +1,4 @@
+using System;
 using System.Threading.Tasks;
 using Dafda.Configuration;
 using Dafda.Producing;
@@ -9,22 +10,6 @@ namespace Dafda.Tests.Configuration
 {
     public class TestServiceCollectionExtensions
     {
-        [Fact]
-        public void registers_a_producer_factory()
-        {
-            var services = new ServiceCollection();
-
-            services.AddProducerFor<SimpleSender>(options =>
-            {
-                options.WithBootstrapServers("dummy");
-            });
-
-            var provider = services.BuildServiceProvider();
-            var producerFactory = provider.GetService<ProducerFactory>();
-
-            Assert.NotNull(producerFactory);
-        }
-
         [Fact]
         public async Task Can_produce_message()
         {
@@ -70,7 +55,7 @@ public void registers_a_typed_producer()
             Assert.NotNull(messageSender.ADependency);
             
             Assert.Equal("hello one", messageSender.ADependency.Message);
-            Assert.Equal(ProducerFactory.GetKeyNameOf<MessageSenderOne>(), messageSender.Producer.Name);
+            Assert.Equal(typeof(MessageSenderOne).FullName, messageSender.Producer.Name);
         }
 
         [Fact]
@@ -88,7 +73,7 @@ public void resolves_a_typed_producer_for_an_abstract_service()
 
             Assert.NotNull(messageSender);
             Assert.NotNull(messageSender.Producer);
-            Assert.Equal(ProducerFactory.GetKeyNameOf<MessageSenderOne>(), messageSender.Producer.Name);
+            Assert.Equal(typeof(IMessageSenderOne).FullName, messageSender.Producer.Name);
         }
 
         [Fact]
@@ -118,7 +103,7 @@ public void registers_multiple_typed_producer()
             Assert.NotNull(messageSenderOne.ADependency);
             
             Assert.Equal("hello one", messageSenderOne.ADependency.Message);
-            Assert.Equal(ProducerFactory.GetKeyNameOf<MessageSenderOne>(), messageSenderOne.Producer.Name);
+            Assert.Equal(typeof(MessageSenderOne).FullName, messageSenderOne.Producer.Name);
 
             var messageSenderTwo = provider.GetRequiredService<MessageSenderTwo>();
             
@@ -127,8 +112,7 @@ public void registers_multiple_typed_producer()
             Assert.NotNull(messageSenderTwo.ADependency);
             
             Assert.Equal("hello two", messageSenderTwo.ADependency.Message);
-            Assert.Equal(ProducerFactory.GetKeyNameOf<MessageSenderTwo>(), messageSenderTwo.Producer.Name);
-            
+            Assert.Equal(typeof(MessageSenderTwo).FullName, messageSenderTwo.Producer.Name);
             Assert.NotEqual(messageSenderOne.Producer.Name, messageSenderTwo.Producer.Name);
         }
 
@@ -148,6 +132,332 @@ public void does_not_register_a_producer_directly()
             Assert.Null(producer);
         }
 
+        [Fact]
+        public async Task Can_produce_message_with_service_provider_options_factory()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var services = new ServiceCollection();
+            services.AddLogging();
+            services.AddProducerFor<SimpleSender>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                options.WithKafkaProducerFactory(__ => spy);
+                options.Register<DummyMessage>("foo", "bar", x => "baz");
+                return options;
+            });
+
+            var provider = services.BuildServiceProvider();
+
+            var simpleSender = provider.GetRequiredService<SimpleSender>();
+            var producer = simpleSender.Producer;
+
+            await producer.Produce(new DummyMessage());
+
+            Assert.Equal("foo", spy.Topic);
+            Assert.Equal("baz", spy.Key);
+        }
+
+        [Fact]
+        public void registers_a_typed_producer_with_service_provider_options_factory()
+        {
+            var services = new ServiceCollection();
+            services.AddLogging();
+            services.AddTransient<MessageSenderOne.AnotherDependency>();
+
+            services.AddProducerFor<MessageSenderOne>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                return options;
+            });
+
+            var provider = services.BuildServiceProvider();
+            var messageSender = provider.GetRequiredService<MessageSenderOne>();
+
+            Assert.NotNull(messageSender);
+            Assert.NotNull(messageSender.Producer);
+            Assert.NotNull(messageSender.ADependency);
+
+            Assert.Equal("hello one", messageSender.ADependency.Message);
+            Assert.Equal(typeof(MessageSenderOne).FullName, messageSender.Producer.Name);
+        }
+
+        [Fact]
+        public void resolves_a_typed_producer_for_an_abstract_service_with_service_provider_options_factory()
+        {
+            var services = new ServiceCollection();
+            services.AddLogging();
+            services.AddTransient<MessageSenderOne.AnotherDependency>();
+            services.AddProducerFor<IMessageSenderOne, MessageSenderOne>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                return options;
+            });
+            var provider = services.BuildServiceProvider();
+            var messageSender = provider.GetRequiredService<IMessageSenderOne>();
+
+            Assert.NotNull(messageSender);
+            Assert.NotNull(messageSender.Producer);
+            Assert.Equal(typeof(IMessageSenderOne).FullName, messageSender.Producer.Name);
+        }
+
+        [Fact]
+        public void registers_multiple_typed_producers_with_service_provider_options_factory()
+        {
+            var services = new ServiceCollection();
+
+            services.AddLogging();
+            services.AddTransient<MessageSenderOne.AnotherDependency>();
+            services.AddProducerFor<MessageSenderOne>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                return options;
+            });
+
+            services.AddTransient<MessageSenderTwo.AnotherDependency>();
+            services.AddProducerFor<MessageSenderTwo>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                return options;
+            });
+
+            var provider = services.BuildServiceProvider();
+
+            var messageSenderOne = provider.GetRequiredService<MessageSenderOne>();
+
+            Assert.NotNull(messageSenderOne);
+            Assert.NotNull(messageSenderOne.Producer);
+            Assert.NotNull(messageSenderOne.ADependency);
+
+            Assert.Equal("hello one", messageSenderOne.ADependency.Message);
+            Assert.Equal(typeof(MessageSenderOne).FullName, messageSenderOne.Producer.Name);
+
+            var messageSenderTwo = provider.GetRequiredService<MessageSenderTwo>();
+
+            Assert.NotNull(messageSenderTwo);
+            Assert.NotNull(messageSenderTwo.Producer);
+            Assert.NotNull(messageSenderTwo.ADependency);
+
+            Assert.Equal("hello two", messageSenderTwo.ADependency.Message);
+            Assert.Equal(typeof(MessageSenderTwo).FullName, messageSenderTwo.Producer.Name);
+        }
+
+        [Fact]
+        public void does_not_register_a_producer_directly_with_service_provider_options_factory()
+        {
+            var services = new ServiceCollection();
+
+            services.AddProducerFor<SimpleSender>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                return options;
+            });
+
+            var provider = services.BuildServiceProvider();
+            var producer = provider.GetService<Producer>();
+
+            Assert.Null(producer);
+        }
+
+        [Fact]
+        public void producer_factory_is_singleton_reusing_same_kafka_producer_instance()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var services = new ServiceCollection();
+            services.AddLogging();
+            services.AddProducerFor<SimpleSender>(options =>
+            {
+                options.WithBootstrapServers("dummy");
+                options.WithKafkaProducerFactory(_ => spy);
+            });
+
+            var provider = services.BuildServiceProvider();
+
+            var senderOne = provider.GetRequiredService<SimpleSender>();
+            var senderTwo = provider.GetRequiredService<SimpleSender>();
+
+            Assert.Same(senderOne.Producer.KafkaProducer, senderTwo.Producer.KafkaProducer);
+        }
+
+        [Fact]
+        public void producer_factory_is_singleton_reusing_same_kafka_producer_instance_with_service_provider_options_factory()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var services = new ServiceCollection();
+            services.AddLogging();
+            services.AddProducerFor<SimpleSender>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                options.WithKafkaProducerFactory(__ => spy);
+                return options;
+            });
+
+            var provider = services.BuildServiceProvider();
+
+            var senderOne = provider.GetRequiredService<SimpleSender>();
+            var senderTwo = provider.GetRequiredService<SimpleSender>();
+
+            Assert.Same(senderOne.Producer.KafkaProducer, senderTwo.Producer.KafkaProducer);
+        }
+
+        [Fact]
+        public void disposing_service_provider_disposes_kafka_producer()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var services = new ServiceCollection();
+            services.AddLogging();
+            services.AddProducerFor<SimpleSender>(options =>
+            {
+                options.WithBootstrapServers("dummy");
+                options.WithKafkaProducerFactory(_ => spy);
+            });
+
+            var provider = services.BuildServiceProvider();
+
+            // Resolve the service so that the KafkaProducer is created inside ProducerFactory
+            _ = provider.GetRequiredService<SimpleSender>();
+
+            Assert.False(spy.WasDisposed);
+
+            provider.Dispose();
+
+            Assert.True(spy.WasDisposed);
+        }
+
+        [Fact]
+        public void disposing_service_provider_does_not_throw_when_kafka_producer_was_never_created()
+        {
+            var services = new ServiceCollection();
+            services.AddLogging();
+            services.AddProducerFor<SimpleSender>(options =>
+            {
+                options.WithBootstrapServers("dummy");
+            });
+
+            var provider = services.BuildServiceProvider();
+
+            // Do NOT resolve SimpleSender — KafkaProducer is never instantiated
+            var exception = Record.Exception(() => provider.Dispose());
+
+            Assert.Null(exception);
+        }
+
+        [Fact]
+        public void throws_when_same_service_type_is_registered_twice_with_action_options()
+        {
+            var services = new ServiceCollection();
+            services.AddProducerFor<SimpleSender>(options =>
+            {
+                options.WithBootstrapServers("dummy");
+            });
+
+            var exception = Assert.Throws<ProducerFactoryException>(() =>
+                services.AddProducerFor<SimpleSender>(options =>
+                {
+                    options.WithBootstrapServers("dummy");
+                }));
+
+            Assert.Contains(typeof(SimpleSender).FullName, exception.Message);
+        }
+
+        [Fact]
+        public void throws_when_same_service_type_is_registered_twice_with_factory_options()
+        {
+            var services = new ServiceCollection();
+            services.AddProducerFor<SimpleSender>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                return options;
+            });
+
+            var exception = Assert.Throws<ProducerFactoryException>(() =>
+                services.AddProducerFor<SimpleSender>(_ =>
+                {
+                    var options = new ProducerOptions();
+                    options.WithBootstrapServers("dummy");
+                    return options;
+                }));
+
+            Assert.Contains(typeof(SimpleSender).FullName, exception.Message);
+        }
+
+        [Fact]
+        public void throws_when_same_abstract_service_type_is_registered_twice_with_action_options()
+        {
+            var services = new ServiceCollection();
+            services.AddTransient<MessageSenderOne.AnotherDependency>();
+            services.AddProducerFor<IMessageSenderOne, MessageSenderOne>(options =>
+            {
+                options.WithBootstrapServers("dummy");
+            });
+
+            var exception = Assert.Throws<ProducerFactoryException>(() =>
+                services.AddProducerFor<IMessageSenderOne, MessageSenderOne>(options =>
+                {
+                    options.WithBootstrapServers("dummy");
+                }));
+
+            Assert.Contains(typeof(IMessageSenderOne).FullName, exception.Message);
+        }
+
+        [Fact]
+        public void throws_when_same_abstract_service_type_is_registered_twice_with_factory_options()
+        {
+            var services = new ServiceCollection();
+            services.AddTransient<MessageSenderOne.AnotherDependency>();
+            services.AddProducerFor<IMessageSenderOne, MessageSenderOne>(_ =>
+            {
+                var options = new ProducerOptions();
+                options.WithBootstrapServers("dummy");
+                return options;
+            });
+
+            var exception = Assert.Throws<ProducerFactoryException>(() =>
+                services.AddProducerFor<IMessageSenderOne, MessageSenderOne>(_ =>
+                {
+                    var options = new ProducerOptions();
+                    options.WithBootstrapServers("dummy");
+                    return options;
+                }));
+
+            Assert.Contains(typeof(IMessageSenderOne).FullName, exception.Message);
+        }
+
+        [Fact]
+        public void options_factory_receives_service_provider()
+        {
+            var services = new ServiceCollection();
+            services.AddLogging();
+            services.AddSingleton(new BootstrapServerSettings("dummy"));
+
+            IServiceProvider capturedProvider = null;
+
+            services.AddProducerFor<SimpleSender>(sp =>
+            {
+                capturedProvider = sp;
+                var settings = sp.GetRequiredService<BootstrapServerSettings>();
+                var options = new ProducerOptions();
+                options.WithBootstrapServers(settings.BootstrapServers);
+                return options;
+            });
+
+            var provider = services.BuildServiceProvider();
+            _ = provider.GetRequiredService<SimpleSender>();
+
+            Assert.NotNull(capturedProvider);
+        }
+
         #region helper classes
 
         private class SimpleSender
@@ -199,6 +509,16 @@ public class AnotherDependency
             }
         }
 
+        private class BootstrapServerSettings
+        {
+            public BootstrapServerSettings(string bootstrapServers)
+            {
+                BootstrapServers = bootstrapServers;
+            }
+
+            public string BootstrapServers { get; }
+        }
+
         #endregion
 
         public class DummyMessage
diff --git a/src/Dafda.Tests/Producing/TestProducerFactory.cs b/src/Dafda.Tests/Producing/TestProducerFactory.cs
deleted file mode 100644
index 6bd0b4a1..00000000
--- a/src/Dafda.Tests/Producing/TestProducerFactory.cs
+++ /dev/null
@@ -1,90 +0,0 @@
-using Dafda.Producing;
-using Dafda.Tests.Builders;
-using Dafda.Tests.TestDoubles;
-using Microsoft.Extensions.Logging.Abstractions;
-using Xunit;
-
-namespace Dafda.Tests.Producing
-{
-    public class TestProducerFactory
-    {
-        [Fact]
-        public void returns_expected_when_nothing_has_been_registered()
-        {
-            var sut = new ProducerFactoryBuilder().Build();
-            var result = sut.Get("foo", NullLoggerFactory.Instance);
-
-            Assert.Null(result);
-        }
-
-        [Fact]
-        public void returns_expected_when_getting_by_a_known_name()
-        {
-            var producerConfigurationStub = A.ValidProducerConfiguration.Build();
-            var outgoingMessageRegistryStub = new OutgoingMessageRegistry();
-
-            var sut = new ProducerFactoryBuilder().Build();
-            sut.ConfigureProducer("foo", producerConfigurationStub, outgoingMessageRegistryStub);
-
-            var result = sut.Get("foo", NullLoggerFactory.Instance);
-
-            Assert.IsType<Producer>(result);
-            Assert.NotNull(result);
-        }
-
-        [Fact]
-        public void returns_expected_when_getting_by_an_unknown_name()
-        {
-            var producerConfigurationStub = A.ValidProducerConfiguration.Build();
-            var outgoingMessageRegistryStub = new OutgoingMessageRegistry();
-
-            var sut = new ProducerFactoryBuilder().Build();
-            sut.ConfigureProducer("foo", producerConfigurationStub, outgoingMessageRegistryStub);
-
-            var result = sut.Get("bar", NullLoggerFactory.Instance);
-
-            Assert.Null(result);
-        }
-
-        [Fact]
-        public void throws_expected_when_adding_multiple_producers_with_same_name()
-        {
-            var sut = new ProducerFactoryBuilder().Build();
-
-            sut.ConfigureProducer(
-                producerName: "foo",
-                configuration: A.ValidProducerConfiguration.Build(),
-                outgoingMessageRegistry: new OutgoingMessageRegistry()
-            );
-
-            Assert.Throws<ProducerFactoryException>(() => sut.ConfigureProducer(
-                producerName: "foo",
-                configuration: A.ValidProducerConfiguration.Build(),
-                outgoingMessageRegistry: new OutgoingMessageRegistry()
-            ));
-        }
-
-        [Fact]
-        public void when_disposing_the_factory_all_kafka_producers_are_also_disposed()
-        {
-            var spy = new KafkaProducerSpy();
-
-            using (var sut = new ProducerFactoryBuilder().Build())
-            {
-                var producerConfiguration = A.ValidProducerConfiguration
-                    .WithKafkaProducerFactory(_ => spy)
-                    .Build();
-
-                sut.ConfigureProducer(
-                    producerName: "foo",
-                    configuration: producerConfiguration,
-                    outgoingMessageRegistry: new OutgoingMessageRegistry()
-                );
-
-                sut.Get("foo", NullLoggerFactory.Instance);
-            }
-
-            Assert.True(spy.WasDisposed);
-        }
-    }
-}
\ No newline at end of file
diff --git a/src/Dafda/Configuration/OutboxProducerOptions.cs b/src/Dafda/Configuration/OutboxProducerOptions.cs
index 2c788f33..e2930c6f 100644
--- a/src/Dafda/Configuration/OutboxProducerOptions.cs
+++ b/src/Dafda/Configuration/OutboxProducerOptions.cs
@@ -2,7 +2,6 @@
 using Dafda.Outbox;
 using Dafda.Producing;
 using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging;
 
 namespace Dafda.Configuration
 {
@@ -88,7 +87,7 @@ public void WithBootstrapServers(string bootstrapServers)
             _builder.WithBootstrapServers(bootstrapServers);
         }
 
-        internal void WithKafkaProducerFactory(Func<ILoggerFactory, KafkaProducer> inlineFactory)
+        internal void WithKafkaProducerFactory(Func<IServiceProvider, KafkaProducer> inlineFactory)
         {
             _builder.WithKafkaProducerFactory(inlineFactory);
         }
diff --git a/src/Dafda/Configuration/OutboxServiceCollectionExtensions.cs b/src/Dafda/Configuration/OutboxServiceCollectionExtensions.cs
index fb4f4d77..382cc47c 100644
--- a/src/Dafda/Configuration/OutboxServiceCollectionExtensions.cs
+++ b/src/Dafda/Configuration/OutboxServiceCollectionExtensions.cs
@@ -54,7 +54,7 @@ public static void AddOutboxProducer(this IServiceCollection services, Action<Ou
             {
                 var outboxUnitOfWorkFactory = provider.GetRequiredService<IOutboxUnitOfWorkFactory>();
                 var loggerFactory = provider.GetRequiredService<ILoggerFactory>();
-                var kafkaProducer = configuration.KafkaProducerFactory(loggerFactory);
+                var kafkaProducer = configuration.KafkaProducerFactory(provider);
                 var producer = new OutboxProducer(kafkaProducer);
                 var outboxDispatcher = new OutboxDispatcher(loggerFactory, outboxUnitOfWorkFactory, producer);
 
diff --git a/src/Dafda/Configuration/ProducerConfiguration.cs b/src/Dafda/Configuration/ProducerConfiguration.cs
index 15e4cea0..ddde970e 100644
--- a/src/Dafda/Configuration/ProducerConfiguration.cs
+++ b/src/Dafda/Configuration/ProducerConfiguration.cs
@@ -7,7 +7,7 @@ namespace Dafda.Configuration
 {
     internal class ProducerConfiguration
     {
-        public ProducerConfiguration(IDictionary<string, string> configuration, MessageIdGenerator messageIdGenerator, Func<ILoggerFactory, KafkaProducer> kafkaProducerFactory)
+        public ProducerConfiguration(IDictionary<string, string> configuration, MessageIdGenerator messageIdGenerator, Func<IServiceProvider, KafkaProducer> kafkaProducerFactory)
         {
             KafkaConfiguration = configuration;
             MessageIdGenerator = messageIdGenerator;
@@ -16,6 +16,6 @@ public ProducerConfiguration(IDictionary<string, string> configuration, MessageI
 
         public IDictionary<string, string> KafkaConfiguration { get; }
         public MessageIdGenerator MessageIdGenerator { get; }
-        public Func<ILoggerFactory, KafkaProducer> KafkaProducerFactory { get; }
+        public Func<IServiceProvider, KafkaProducer> KafkaProducerFactory { get; }
     }
 }
\ No newline at end of file
diff --git a/src/Dafda/Configuration/ProducerConfigurationBuilder.cs b/src/Dafda/Configuration/ProducerConfigurationBuilder.cs
index 28936dc6..0fc72b0b 100644
--- a/src/Dafda/Configuration/ProducerConfigurationBuilder.cs
+++ b/src/Dafda/Configuration/ProducerConfigurationBuilder.cs
@@ -3,6 +3,7 @@
 using System.Linq;
 using Dafda.Producing;
 using Dafda.Serializing;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 
 namespace Dafda.Configuration
@@ -31,7 +32,7 @@ internal sealed class ProducerConfigurationBuilder
 
         private ConfigurationSource _configurationSource = ConfigurationSource.Null;
         private MessageIdGenerator _messageIdGenerator = MessageIdGenerator.Default;
-        private Func<ILoggerFactory, KafkaProducer> _kafkaProducerFactory;
+        private Func<IServiceProvider, KafkaProducer> _kafkaProducerFactory;
         private static readonly IPayloadSerializer _defaultPayloadSerializer = new DefaultPayloadSerializer();
         private readonly TopicPayloadSerializerRegistry _topicPayloadSerializerRegistry = new TopicPayloadSerializerRegistry(() => _defaultPayloadSerializer);
 
@@ -82,7 +83,7 @@ public ProducerConfigurationBuilder WithMessageIdGenerator(MessageIdGenerator me
             return this;
         }
 
-        internal ProducerConfigurationBuilder WithKafkaProducerFactory(Func<ILoggerFactory,KafkaProducer> inlineFactory)
+        internal ProducerConfigurationBuilder WithKafkaProducerFactory(Func<IServiceProvider, KafkaProducer> inlineFactory)
         {
             _kafkaProducerFactory = inlineFactory;
             return this;
@@ -112,7 +113,11 @@ internal ProducerConfiguration Build()
 
             if (_kafkaProducerFactory == null)
             {
-                _kafkaProducerFactory = loggerFactory => new KafkaProducer(loggerFactory, configurations, _topicPayloadSerializerRegistry);
+                _kafkaProducerFactory = provider =>
+                {
+                    var loggerFactory = provider.GetRequiredService<ILoggerFactory>();
+                    return new KafkaProducer(loggerFactory, configurations, _topicPayloadSerializerRegistry);
+                };
             }
 
             return new ProducerConfiguration(
diff --git a/src/Dafda/Configuration/ProducerOptions.cs b/src/Dafda/Configuration/ProducerOptions.cs
index 05c8b980..1393a71c 100644
--- a/src/Dafda/Configuration/ProducerOptions.cs
+++ b/src/Dafda/Configuration/ProducerOptions.cs
@@ -1,7 +1,6 @@
 using System;
 using Dafda.Producing;
 using Dafda.Serializing;
-using Microsoft.Extensions.Logging;
 
 namespace Dafda.Configuration
 {
@@ -10,14 +9,8 @@ namespace Dafda.Configuration
     /// </summary>
     public sealed class ProducerOptions
     {
-        private readonly ProducerConfigurationBuilder _builder;
-        private readonly OutgoingMessageRegistry _outgoingMessageRegistry;
-
-        internal ProducerOptions(ProducerConfigurationBuilder builder, OutgoingMessageRegistry outgoingMessageRegistry)
-        {
-            _builder = builder;
-            _outgoingMessageRegistry = outgoingMessageRegistry;
-        }
+        internal readonly ProducerConfigurationBuilder Builder = new();
+        internal readonly OutgoingMessageRegistry OutgoingMessageRegistry = new();
 
         /// <summary>
         /// Specify a custom implementation of the <see cref="ConfigurationSource"/> to use. 
@@ -25,7 +18,7 @@ internal ProducerOptions(ProducerConfigurationBuilder builder, OutgoingMessageRe
         /// <param name="configurationSource">The <see cref="ConfigurationSource"/> to use.</param>
         public void WithConfigurationSource(ConfigurationSource configurationSource)
         {
-            _builder.WithConfigurationSource(configurationSource);
+            Builder.WithConfigurationSource(configurationSource);
         }
 
         /// <summary>
@@ -34,7 +27,7 @@ public void WithConfigurationSource(ConfigurationSource configurationSource)
         /// <param name="configuration">The configuration instance.</param>
         public void WithConfigurationSource(Microsoft.Extensions.Configuration.IConfiguration configuration)
         {
-            _builder.WithConfigurationSource(new DefaultConfigurationSource(configuration));
+            Builder.WithConfigurationSource(new DefaultConfigurationSource(configuration));
         }
 
         /// <summary>
@@ -44,7 +37,7 @@ public void WithConfigurationSource(Microsoft.Extensions.Configuration.IConfigur
         /// <param name="converter">Use this to transform keys.</param>
         public void WithNamingConvention(Func<string, string> converter)
         {
-            _builder.WithNamingConvention(converter);
+            Builder.WithNamingConvention(converter);
         }
 
         /// <summary>
@@ -63,7 +56,7 @@ public void WithNamingConvention(Func<string, string> converter)
         /// <param name="additionalPrefixes">Additional prefixes to use before keys.</param>
         public void WithEnvironmentStyle(string prefix = null, params string[] additionalPrefixes)
         {
-            _builder.WithEnvironmentStyle(prefix, additionalPrefixes);
+            Builder.WithEnvironmentStyle(prefix, additionalPrefixes);
         }
 
         /// <summary>
@@ -73,7 +66,7 @@ public void WithEnvironmentStyle(string prefix = null, params string[] additiona
         /// <param name="value">The configuration value.</param>
         public void WithConfiguration(string key, string value)
         {
-            _builder.WithConfiguration(key, value);
+            Builder.WithConfiguration(key, value);
         }
 
         /// <summary>
@@ -82,12 +75,12 @@ public void WithConfiguration(string key, string value)
         /// <param name="bootstrapServers">A list of bootstrap servers.</param>
         public void WithBootstrapServers(string bootstrapServers)
         {
-            _builder.WithBootstrapServers(bootstrapServers);
+            Builder.WithBootstrapServers(bootstrapServers);
         }
 
-        internal void WithKafkaProducerFactory(Func<ILoggerFactory, KafkaProducer> inlineFactory)
+        internal void WithKafkaProducerFactory(Func<IServiceProvider, KafkaProducer> inlineFactory)
         {
-            _builder.WithKafkaProducerFactory(inlineFactory);
+            Builder.WithKafkaProducerFactory(inlineFactory);
         }
 
         /// <summary>
@@ -96,7 +89,7 @@ internal void WithKafkaProducerFactory(Func<ILoggerFactory, KafkaProducer> inlin
         /// <param name="messageIdGenerator">A custom implementation of <see cref="MessageIdGenerator"/>.</param>
         public void WithMessageIdGenerator(MessageIdGenerator messageIdGenerator)
         {
-            _builder.WithMessageIdGenerator(messageIdGenerator);
+            Builder.WithMessageIdGenerator(messageIdGenerator);
         }
 
         /// <summary>
@@ -109,7 +102,7 @@ public void WithMessageIdGenerator(MessageIdGenerator messageIdGenerator)
         /// and returns a string of the Kafka partition key.</param>
         public void Register<T>(string topic, string type, Func<T, string> keySelector) where T : class
         {
-            _outgoingMessageRegistry.Register(topic, type, keySelector);
+            OutgoingMessageRegistry.Register(topic, type, keySelector);
         }
 
         /// <summary>
@@ -129,7 +122,7 @@ public void WithDefaultPayloadSerializer(IPayloadSerializer payloadSerializer)
         /// </param>
         public void WithDefaultPayloadSerializer(Func<IPayloadSerializer> payloadSerializerFactory)
         {
-            _builder.WithDefaultPayloadSerializer(payloadSerializerFactory);
+            Builder.WithDefaultPayloadSerializer(payloadSerializerFactory);
         }
 
         /// <summary>
@@ -153,21 +146,15 @@ public void WithPayloadSerializer(string topic, IPayloadSerializer payloadSerial
         /// </param>
         public void WithPayloadSerializer(string topic, Func<IPayloadSerializer> payloadSerializerFactory)
         {
-            _builder.WithPayloadSerializer(topic, payloadSerializerFactory);
+            Builder.WithPayloadSerializer(topic, payloadSerializerFactory);
         }
 
-        private class DefaultConfigurationSource : ConfigurationSource
+        private class DefaultConfigurationSource(Microsoft.Extensions.Configuration.IConfiguration configuration)
+            : ConfigurationSource
         {
-            private readonly Microsoft.Extensions.Configuration.IConfiguration _configuration;
-
-            public DefaultConfigurationSource(Microsoft.Extensions.Configuration.IConfiguration configuration)
-            {
-                _configuration = configuration;
-            }
-
             public override string GetByKey(string key)
             {
-                return _configuration[key];
+                return configuration[key];
             }
         }
     }
diff --git a/src/Dafda/Configuration/ProducerServiceCollectionExtensions.cs b/src/Dafda/Configuration/ProducerServiceCollectionExtensions.cs
index 18b034c5..bb7e2016 100644
--- a/src/Dafda/Configuration/ProducerServiceCollectionExtensions.cs
+++ b/src/Dafda/Configuration/ProducerServiceCollectionExtensions.cs
@@ -1,19 +1,20 @@
 using System;
 using System.Linq;
+using System.Threading;
 using Dafda.Producing;
 using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging;
 
 namespace Dafda.Configuration
 {
-    /// <summary></summary>
+    /// <summary>
+    /// Extension methods for registering Kafka producers with the Microsoft dependency injection container.
+    /// </summary>
     public static class ProducerServiceCollectionExtensions
     {
         /// <summary>
         /// Add a Kafka producer available through the Microsoft dependency injection's <see cref="IServiceProvider"/>
-        /// as <see cref="Producer"/>. 
-        ///
-        /// NOTE: currently only a single producer can be configured per <typeparamref name="TImplementation"/>.
+        /// as <typeparamref name="TService"/>. Each <typeparamref name="TService"/> must be unique —
+        /// duplicate registrations will throw a <see cref="ProducerFactoryException"/>.
         /// </summary>
         /// <param name="services">The <see cref="IServiceCollection"/> used in <c>Startup</c>.</param>
         /// <param name="options">Use this action to override Dafda and underlying Kafka configuration.</param>
@@ -21,62 +22,119 @@ public static void AddProducerFor<TService, TImplementation>(this IServiceCollec
             where TImplementation : class, TService 
             where TService : class
         {
-            var factory = CreateProducerFactory<TImplementation>(services, options);
-            services.AddTransient<TService, TImplementation>(provider =>
-                CreateInstance<TImplementation>(provider, factory));
+            ThrowIfProducerServiceAlreadyRegisteredFor<TService>(services);
+
+            services.AddSingleton(provider =>
+            {
+                var producerOptions = new ProducerOptions();
+                options?.Invoke(producerOptions);
+                return new ProducerFactory<TService>(provider, producerOptions);
+            });
+            
+            services.AddTransient<TService, TImplementation>(CreateProducerService<TService, TImplementation>);
         }
 
         /// <summary>
         /// Add a Kafka producer available through the Microsoft dependency injection's <see cref="IServiceProvider"/>
-        /// as <see cref="Producer"/>. 
-        ///
-        /// NOTE: currently only a single producer can be configured per <typeparamref name="TClient"/>.
+        /// as <typeparamref name="TClient"/>. Each <typeparamref name="TClient"/> must be unique —
+        /// duplicate registrations will throw a <see cref="ProducerFactoryException"/>.
         /// </summary>
         /// <param name="services">The <see cref="IServiceCollection"/> used in <c>Startup</c>.</param>
         /// <param name="options">Use this action to override Dafda and underlying Kafka configuration.</param>
         public static void AddProducerFor<TClient>(this IServiceCollection services, Action<ProducerOptions> options) where TClient : class
         {
-            var factory = CreateProducerFactory<TClient>(services, options);
-            services.AddTransient<TClient>(provider =>
-                 CreateInstance<TClient>(provider, factory));
+            AddProducerFor<TClient, TClient>(services, options);
         }
 
-        private static TImplementation CreateInstance<TImplementation>(IServiceProvider provider, ProducerFactory factory)
+        /// <summary>
+        /// Add a Kafka producer available through the Microsoft dependency injection's <see cref="IServiceProvider"/>
+        /// as <typeparamref name="TService"/>. Each <typeparamref name="TService"/> must be unique —
+        /// duplicate registrations will throw a <see cref="ProducerFactoryException"/>.
+        ///
+        /// Use this overload when configuration depends on other services (for example, <c>IConfiguration</c>).
+        /// </summary>
+        /// <param name="services">The <see cref="IServiceCollection"/> used in <c>Startup</c>.</param>
+        /// <param name="optionsFactory">Factory that creates and configures <see cref="ProducerOptions"/> using the built <see cref="IServiceProvider"/>.</param>
+        public static void AddProducerFor<TService, TImplementation>(this IServiceCollection services, Func<IServiceProvider, ProducerOptions> optionsFactory)
+            where TImplementation : class, TService
+            where TService : class
         {
-            var loggerFactory = provider.GetRequiredService<ILoggerFactory>();
-            var producer = factory.GetFor<TImplementation>(loggerFactory);
+            ThrowIfProducerServiceAlreadyRegisteredFor<TService>(services);
+
+            services.AddSingleton(provider =>
+            {
+                var options = optionsFactory(provider);
+                return new ProducerFactory<TService>(provider, options);
+            });
+            
+            services.AddTransient<TService, TImplementation>(CreateProducerService<TService, TImplementation>);
+        }
+
+        /// <summary>
+        /// Add a Kafka producer available through the Microsoft dependency injection's <see cref="IServiceProvider"/>
+        /// as <typeparamref name="TClient"/>. Each <typeparamref name="TClient"/> must be unique —
+        /// duplicate registrations will throw a <see cref="ProducerFactoryException"/>.
+        ///
+        /// Use this overload when configuration depends on other services (for example, <c>IConfiguration</c>).
+        /// </summary>
+        /// <param name="services">The <see cref="IServiceCollection"/> used in <c>Startup</c>.</param>
+        /// <param name="optionsFactory">Factory that creates and configures <see cref="ProducerOptions"/> using the built <see cref="IServiceProvider"/>.</param>
+        public static void AddProducerFor<TClient>(this IServiceCollection services, Func<IServiceProvider, ProducerOptions> optionsFactory) where TClient : class
+        {
+            AddProducerFor<TClient, TClient>(services, optionsFactory);
+        }
+    
+        private static void ThrowIfProducerServiceAlreadyRegisteredFor<TService>(IServiceCollection services) where TService : class
+        {
+            if (services.Any(d => d.ServiceType == typeof(TService)))
+            {
+                throw new ProducerFactoryException(
+                    $"Service type \"{typeof(TService).FullName}\" is already registered in the dependency injection container. Each Dafda producer must use a unique service type.");
+            }
+        }
+
+        private static TImplementation CreateProducerService<TService, TImplementation>(IServiceProvider provider)
+            where TImplementation : class, TService
+            where TService : class
+        {
+            var producerFactory = provider.GetRequiredService<ProducerFactory<TService>>();
+            var producer = producerFactory.CreateProducerInstance();
             return ActivatorUtilities.CreateInstance<TImplementation>(provider, producer);
         }
+    }
+    
+    internal class ProducerFactory<TService>(IServiceProvider provider, ProducerOptions options) : IDisposable
+    {
+        private readonly ProducerConfiguration _configuration = options.Builder.Build();
+        private readonly OutgoingMessageRegistry _messageRegistry = options.OutgoingMessageRegistry;
+        private Lazy<KafkaProducer> _kafkaProducer;
 
-        private static ProducerFactory CreateProducerFactory<TImplementation>(IServiceCollection services, Action<ProducerOptions> options)
+        public Producer CreateProducerInstance()
         {
-            var outgoingMessageRegistry = new OutgoingMessageRegistry();
-            var configurationBuilder = new ProducerConfigurationBuilder();
-            var consumerOptions = new ProducerOptions(configurationBuilder, outgoingMessageRegistry);
-            options?.Invoke(consumerOptions);
+            // Use LazyInitializer to ensure thread-safe, once-only construction while
+            // capturing `provider` in the factory delegate.
+            LazyInitializer.EnsureInitialized(
+                ref _kafkaProducer,
+                () => new Lazy<KafkaProducer>(() => _configuration.KafkaProducerFactory(provider), isThreadSafe: true));
 
-            var producerConfiguration = configurationBuilder.Build();
+            var producer = new Producer(
+                kafkaProducer: _kafkaProducer.Value,
+                outgoingMessageRegistry: _messageRegistry,
+                messageIdGenerator: _configuration.MessageIdGenerator
+            )
+            {
+                Name = typeof(TService).FullName
+            };
 
-            var factory = AddOrGetRegisteredProducerFactory(services);
-            factory.ConfigureProducerFor<TImplementation>(producerConfiguration, outgoingMessageRegistry);
-            return factory;
+            return producer;
         }
 
-        private static ProducerFactory AddOrGetRegisteredProducerFactory(IServiceCollection services)
+        public void Dispose()
         {
-            var factory = services
-                .Where(x => x.ServiceType == typeof(ProducerFactory))
-                .Select(x => x.ImplementationInstance)
-                .Cast<ProducerFactory>()
-                .SingleOrDefault();
-
-            if (factory == null)
+            if (_kafkaProducer is { IsValueCreated: true })
             {
-                factory = new ProducerFactory();
-                services.AddSingleton(factory);
+                _kafkaProducer.Value.Dispose();
             }
-
-            return factory;
         }
     }
-}
\ No newline at end of file
+}
diff --git a/src/Dafda/Producing/Producer.cs b/src/Dafda/Producing/Producer.cs
index 44059544..571eb745 100644
--- a/src/Dafda/Producing/Producer.cs
+++ b/src/Dafda/Producing/Producer.cs
@@ -11,16 +11,16 @@ namespace Dafda.Producing
     /// </summary>
     public sealed class Producer
     {
-        private readonly KafkaProducer _kafkaProducer;
         private readonly PayloadDescriptorFactory _payloadDescriptorFactory;
 
         internal Producer(KafkaProducer kafkaProducer, OutgoingMessageRegistry outgoingMessageRegistry, MessageIdGenerator messageIdGenerator)
         {
-            _kafkaProducer = kafkaProducer;
+            KafkaProducer = kafkaProducer;
             _payloadDescriptorFactory = new PayloadDescriptorFactory(outgoingMessageRegistry, messageIdGenerator);
         }
 
         internal string Name { get; set; } = "__Default Producer__";
+        internal KafkaProducer KafkaProducer { get; }
 
         /// <summary>
         /// Produce a <paramref name="message"/> on Kafka
@@ -39,10 +39,10 @@ public async Task Produce(object message)
         public async Task Produce(object message, Metadata headers)
         {
             var payloadDescriptor = _payloadDescriptorFactory.Create(message, headers);
-            payloadDescriptor.ClientId = _kafkaProducer.ClientId;
+            payloadDescriptor.ClientId = KafkaProducer.ClientId;
             using var activity = DafdaActivitySource.StartPublishingActivity(payloadDescriptor);
 
-            await _kafkaProducer.Produce(payloadDescriptor);
+            await KafkaProducer.Produce(payloadDescriptor);
         }
 
         /// <summary>
@@ -75,10 +75,10 @@ public async Task Produce(object message, MessageHandlerContext context)
         public async Task Produce(object message, MessageHandlerContext context, Dictionary<string, string> headers)
         {
             var payloadDescriptor = _payloadDescriptorFactory.Create(message, context, headers);
-            payloadDescriptor.ClientId = _kafkaProducer.ClientId;
+            payloadDescriptor.ClientId = KafkaProducer.ClientId;
             using var activity = DafdaActivitySource.StartPublishingActivity(payloadDescriptor);
 
-            await _kafkaProducer.Produce(payloadDescriptor);
+            await KafkaProducer.Produce(payloadDescriptor);
         }
     }
 }
\ No newline at end of file
diff --git a/src/Dafda/Producing/ProducerFactory.cs b/src/Dafda/Producing/ProducerFactory.cs
deleted file mode 100644
index bdf7f811..00000000
--- a/src/Dafda/Producing/ProducerFactory.cs
+++ /dev/null
@@ -1,101 +0,0 @@
-using System;
-using System.Collections.Generic;
-using Dafda.Configuration;
-using Microsoft.Extensions.Logging;
-
-namespace Dafda.Producing
-{
-    internal sealed class ProducerFactory : IDisposable
-    {
-        private readonly Dictionary<string, ProducerBuilder> _producerBuilders = new Dictionary<string, ProducerBuilder>();
-
-        internal void ConfigureProducer(string producerName, ProducerConfiguration configuration, OutgoingMessageRegistry outgoingMessageRegistry)
-        {
-            if (_producerBuilders.ContainsKey(producerName))
-            {
-                throw new ProducerFactoryException($"A producer with the name \"{producerName}\" has already been configured. Producer names should be unique.");
-            }
-
-            _producerBuilders.Add(producerName, new ProducerBuilder(
-                    producerName: producerName,
-                    configuration: configuration, 
-                    messageRegistry: outgoingMessageRegistry
-                ));
-        }
-
-        internal static string GetKeyNameOf<TClient>() => $"__INTERNAL__FOR_CLIENT__{typeof(TClient).FullName}";
-
-        internal void ConfigureProducerFor<TClient>(ProducerConfiguration configuration, OutgoingMessageRegistry outgoingMessageRegistry)
-        {
-            var producerName = GetKeyNameOf<TClient>();
-            ConfigureProducer(producerName, configuration, outgoingMessageRegistry);
-        }
-
-        public Producer Get(string producerName, ILoggerFactory loggerFactory) 
-        {
-            if (_producerBuilders.TryGetValue(producerName, out var builder))
-            {
-                return builder.Build(loggerFactory);
-            }
-
-            return null;
-        }
-
-        public Producer GetFor<TClient>(ILoggerFactory loggerFactory)
-        {
-            var producerName = GetKeyNameOf<TClient>();
-            return Get(producerName, loggerFactory);
-        }
-
-        public void Dispose()
-        {
-            if (_producerBuilders != null)
-            {
-                foreach (var registration in _producerBuilders.Values)
-                {
-                    registration.Dispose();
-                }
-            }
-        }
-
-        private class ProducerBuilder : IDisposable
-        {
-            private readonly OutgoingMessageRegistry _messageRegistry;
-            private readonly Func<ILoggerFactory, KafkaProducer> _kafkaProducerFactory;
-            private readonly MessageIdGenerator _messageIdGenerator;
-            private readonly string _producerName;
-            private KafkaProducer _kafkaProducer;
-
-            public ProducerBuilder(string producerName, ProducerConfiguration configuration, OutgoingMessageRegistry messageRegistry)
-            {
-                _messageRegistry = messageRegistry;
-                _producerName = producerName;
-                _kafkaProducerFactory = configuration.KafkaProducerFactory;
-                _messageIdGenerator = configuration.MessageIdGenerator;
-            }
-
-            public Producer Build(ILoggerFactory loggerFactory)
-            {
-                if (_kafkaProducer is null)
-                {
-                    _kafkaProducer = _kafkaProducerFactory(loggerFactory);
-                }
-
-                var producer = new Producer(
-                    kafkaProducer: _kafkaProducer,
-                    outgoingMessageRegistry: _messageRegistry,
-                    messageIdGenerator: _messageIdGenerator
-                );
-
-                producer.Name = _producerName;
-
-                return producer;
-            }
-
-            public void Dispose()
-            {
-                _kafkaProducer?.Dispose();
-            }
-        }
-    }
-}
\ No newline at end of file
diff --git a/src/Dafda/Producing/ProducerFactoryException.cs b/src/Dafda/Producing/ProducerFactoryException.cs
index 328ad444..0f353c52 100644
--- a/src/Dafda/Producing/ProducerFactoryException.cs
+++ b/src/Dafda/Producing/ProducerFactoryException.cs
@@ -4,8 +4,14 @@
 namespace Dafda.Producing
 {
     /// <summary>
-    /// Exception thrown when multiple identical producers are added via
-    /// <see cref="ProducerServiceCollectionExtensions.AddProducerFor{TClient}"/>
+    /// Exception thrown when the service type used to identify a Dafda producer is already present in the
+    /// <see cref="Microsoft.Extensions.DependencyInjection.IServiceCollection"/> at the time
+    /// <see cref="ProducerServiceCollectionExtensions.AddProducerFor{TClient}(Microsoft.Extensions.DependencyInjection.IServiceCollection,System.Action{Dafda.Configuration.ProducerOptions})"/>,
+    /// <see cref="ProducerServiceCollectionExtensions.AddProducerFor{TService,TImplementation}(Microsoft.Extensions.DependencyInjection.IServiceCollection,System.Action{Dafda.Configuration.ProducerOptions})"/>,
+    /// <see cref="ProducerServiceCollectionExtensions.AddProducerFor{TClient}(Microsoft.Extensions.DependencyInjection.IServiceCollection,System.Func{System.IServiceProvider,Dafda.Configuration.ProducerOptions})"/>, or
+    /// <see cref="ProducerServiceCollectionExtensions.AddProducerFor{TService,TImplementation}(Microsoft.Extensions.DependencyInjection.IServiceCollection,System.Func{System.IServiceProvider,Dafda.Configuration.ProducerOptions})"/>
+    /// is called. Each Dafda producer requires a service type that is not yet registered — the conflicting
+    /// registration may come from another <c>AddProducerFor</c> call or from any other source.
     /// </summary>
     public sealed class ProducerFactoryException : Exception
     {