zoukankan      html  css  js  c++  java
  • 【EasyNetQ】- 自动订阅者

    从v0.7.1.30开始,EasyNetQ简单易用AutoSubscriber你可以用它来轻松地扫描实现任何接口的类的特定组件IConsume<T>IConsumeAsync<T>,然后让汽车用户订阅这些消费者到你的总线。一个实现IConsume<T>将使用总线订阅方法,而实现IConsumeAsync<T>将使用总线SubscribeAsync方法,请参阅订阅详细信息。您当然可以让您的消费者处理多条消息。我们来看看一些样品。

    注意:从版本0.13.0开始,所有AutoSubscriber类都在EasyNetQ.AutoSubscribe命名空间中,因此请添加以下using语句:

    using EasyNetQ.AutoSubscribe;

    让我们定义一个简单的消费,处理三个消息:MessageAMessageBMessageC

    public class MyConsumer : IConsume<MessageA>, IConsume<MessageB>, IConsumeAsync<MessageC>
    {
        public void Consume(MessageA message) {...}
    
        public void Consume(MessageB message) {...}
    
        public Task Consume(MessageC message) {...}
    }
     

    首先创建一个AutoSubscriber的新实例,将您的IBus实例和subscriptionId前缀传递给构造函数。subscriptionId前缀以所有自动生成的subscriptionId为前缀,但不以自定义subscriptionIds为前缀(见下文)。

    在同一个程序集中注册它和所有其他使用者,我们只需要将包含您的使用者的程序集传递给:AutoSubscriber.Subscribe(assembly)注意!这是你唯一应该做的ONCE,最好在应用程序启动。

    var subscriber = new AutoSubscriber(bus, "my_applications_subscriptionId_prefix");
    subscriber.Subscribe(Assembly.GetExecutingAssembly());
     

    按主题订阅

    默认情况下,AutoSubscriber将绑定没有主题。在下面的示例中,MessageA注册了两个主题。

    注意!如果您运行没有ForTopic属性的代码,它将具有“#”的路由键,它将获取对消息类型的任何订阅。假设安装了默认端口和管理插件,只需访问http//localhost:15672 /#/ queues并解除绑定路由(如果需要)。

    public class MyConsumer : IConsume<MessageA>, IConsume<MessageB>, IConsumeAsync<MessageC>
    {
        [ForTopic("Topic.Foo")]
        [ForTopic("Topic.Bar")]
        public void Consume(MessageA message) {...}
    
        public void Consume(MessageB message) {...}
    
        public Task Consume(MessageC message) {...}
    }
    
    //To publish by topic
    var bus = RabbitHutch.CreateBus("host=localhost");
    
    var msg1 = new MessageA(msg1, "Topic.Foo");   //picked up
    var msg2 = new MessageA(msg2, "Topic.Bar");   //picked up
    var msg3 = new MessageA(msg3);                //not picked up

    指定特定的SubscriptionId

    默认情况下,AutoSubscriber将为SubscriptionId每个消息/消费者组合生成唯一这意味着您将启动同一个使用者的多个实例,并且它们将以循环方式(工作模式)从相同的队列中读取。

    如果你想修改订阅ID,你可以ConsumeAutoSubscriberConsumerAttribute修饰方法为什么要修复它,你可以在这里阅读

    可以说,上面的消费者应该有一个固定SubscriptionId的消费方法MessageB只是装饰它并定义一个值SubscriptionId

    [AutoSubscriberConsumer(SubscriptionId = "MyExplicitId")]
    public void Consume(MessageB message) { }

    控制SubscriptionId生成

    你当然也可以控制实际的SubscriptionId一代。只需更换AutoSubscriber.GenerateSubscriptionId : Func<ConsumerInfo, string>

    var subscriber = new AutoSubscriber(bus)
    {
        CreateConsumer = t => objectResolver.Resolve(t),
        GenerateSubscriptionId = c => AppDomain.CurrentDomain.FriendlyName + c.ConcreteType.Name
    };
    subscriber.Subscribe(Assembly.GetExecutingAssembly());

    注意!只是一个示例实现。确保您已阅读并理解SubscriptionId价值的重要性

    控制消费者配置设置

    使用autosubscriber订阅队列时,您可以设置ISubscriptionConfiguration值,例如AutoDelete,Priority等。

    通过在创建AutoSubscriber时设置操作。

    var subscriber = new AutoSubscriber(bus)
    {    
        ConfigureSubscriptionConfiguration = c => c.WithAutoDelete()
                                                   .WithPriority(10)
    };
    subscriber.Subscribe(Assembly.GetExecutingAssembly());

    或者,您可以将一个属性应用于consume方法,该方法优先于ConfigureSubscriptionConfiguration操作设置的任何配置值。

    public class MyConsumer : IConsume<MessageA>
    {
        [SubscriptionConfiguration(CancelOnHaFailover = true, PrefetchCount = 10)]
        public void Consume(MessageA message) {...}
    }

    在AutoSubscriber中使用IoC容器

    AutoSubscriber有一个属性MessageDispatcher,它允许您插入自己的消息调度代码。这允许您从IoC容器中解析您的使用者或执行其他自定义调度时间任务。

    让我们编写一个自定义IAutoSubscriberMessageDispatcher来解析Windsor IoC容器中的使用者

    public class WindsorMessageDispatcher : IAutoSubscriberMessageDispatcher
    {
        private readonly IWindsorContainer container;
    
        public WindsorMessageDispatcher(IWindsorContainer container)
        {
            this.container = container;
        }
    
        public void Dispatch<TMessage, TConsumer>(TMessage message) where TMessage : class where TConsumer : IConsume<TMessage>
        {
            var consumer = container.Resolve<TConsumer>();
            try
            {
                consumer.Consume(message);
            }
            finally
            {
                container.Release(consumer);
            }
        }
    
        public Task DispatchAsync<TMessage, TConsumer>(TMessage message) where TMessage: class where TConsumer: IConsumeAsync<TMessage>
        {
            var consumer = _container.Resolve<TConsumer>();           
            return consumer.Consume(message).ContinueWith(t=>_container.Release(consumer));            
        }
    }

    现在我们需要使用IoC容器注册我们的消费者:

    var container = new WindsorContainer();
    container.Register(
        Component.For<MyConsumer>().ImplementedBy<MyConsumer>()
        );

    接下来使用我们的自定义IMessageDispatcher设置AutoSubscriber:

    var bus = RabbitHutch.CreateBus("host=localhost");
    
    var autoSubscriber = new AutoSubscriber(bus, "My_subscription_id_prefix")
    {
        MessageDispatcher = new WindsorMessageDispatcher(container)
    };
    autoSubscriber.Subscribe(GetType().Assembly);
    autoSubscriber.SubscribeAsync(GetType().Assembly);

    现在,每当消息到达时,我们的消费者的新实例将从我们的容器中解析。

  • 相关阅读:
    2017 年终总结 —— 在路上
    尝试造了个工具类库,名为 Diana
    走近 Python (类比 JS)
    Node.js 异步异闻录
    使用 Node.js 搭建一个 API 网关
    不就是语法和长难句吗—笔记总结Day4
    不就是语法和长难句吗—笔记总结Day3
    不就是语法和长难句吗—笔记总结Day2
    不就是语法和长难句吗—笔记总结Day1
    Kali Day1
  • 原文地址:https://www.cnblogs.com/wangwust/p/9437514.html
Copyright © 2011-2022 走看看