EasyNetQ订阅者不使用TopShelf

Question

我对RabbitMQ、EasyNetQ和TopShelf的这种组合还是个新手。目前，我没有使用任何DI。

我正在尝试使用EasyNetQ订阅队列。订阅使用此控制台应用程序代码

class Program
{
    static void Main(string[] args)
    {
        using (var bus = RabbitHutch.CreateBus("host=localhost"))
        {
            bus.Subscribe<Entity>("entity", Handler);
            Console.ReadLine();

        }
    }

    private static void Handler(Entity obj)
    {
        Console.WriteLine($"{obj.ID}, {obj.Name}");
    }
}

对于TopShelf，它不会命中处理程序方法。我没有看到TopShelf或EasyNetQ报告的任何错误

class Program
{
    static void Main(string[] args)
    {
        HostFactory.Run(config =>
        {
            config.Service<TestEasyNet>(service =>
            {
                service.ConstructUsing(s => new TestEasyNet());
                service.WhenStarted(s => s.Start());
                service.WhenStopped(s => s.Stop());
            });

            config.SetServiceName("TestSubscribe");
            config.SetDisplayName("Test Subscribe");
            config.SetDescription("Test Subscribe");
        });
    }
}

class TestEasyNet
{
    public void Start()
    {
        using (var bus = EasyNetQ.RabbitHutch.CreateBus("host=localhost"))
        {
            bus.Subscribe<Entity>("entity", Handler);
        }
    }

    private void Handler(Entity obj)
    {
        Console.WriteLine("Subscribing");
        Console.WriteLine($"{obj.ID}, {obj.Name}");
    }

    public void Stop()
    { }
}

消息发布代码是

class Program
{
    static void Main(string[] args)
    {
        HostFactory.Run(c =>
        {
            c.Service<Hosting>(service =>
            {
                service.ConstructUsing(s => new Hosting());
                service.WhenStarted(s => s.Start());
                service.WhenStopped(s => s.Stop());
            });
            c.SetServiceName("TempService");
            c.SetDisplayName("Temp Service");
            c.SetDescription("Temp Service");
        });
    }
}

public class Hosting
{
    public void Start()
    {
        var entity = new Entity()
        {
            ID = 1,
            Name = "Entity 1"
        };

        var entity2 = new Entity()
        {
            ID = 2,
            Name = "Entity 2"
        };

        var entity3 = new Entity()
        {
            ID = 3,
            Name = "Entity 3"
        };

        using (var bus = RabbitHutch.CreateBus("host=localhost"))
        {
            bus.Publish<Entity>(entity);
            bus.Publish<Entity>(entity2);
            bus.Publish<Entity>(entity3);
        }
    }

    public void Stop()
    {             
    }
}

我找不到哪里错了！

Answer 1

这里：

using (var bus = EasyNetQ.RabbitHutch.CreateBus("host=localhost"))
{
    bus.Subscribe<Entity>("entity", Handler);
}

代码在订阅后立即处理到EasyNetQ的连接-这将断开连接并再次终止订阅。根据EasyNetQ documentation

标准做法是在应用程序的整个生命周期内创建单个IBus实例。在您的应用程序关闭时将其释放。

在这种情况下，您可能希望将EasyNetQ总线的生命周期与通过TopShelf启动或停止的服务绑定在一起。所以：

private IBus bus;

public void Start()
{
    bus = EasyNetQ.RabbitHutch.CreateBus("host=localhost"));
    bus.Subscribe<Entity>("entity", Handler);
}

public void Stop()
{
    bus?.Dispose();
    bus = null;
}