时间戳

首页 > 解决方案 > 通过基类发布时不执行消费者代码

.net-core - 通过基类发布时不执行消费者代码

问题描述

我有一个基类IntegrationEvent和从它继承的所有其他业务事件。

    public abstract class IntegrationEvent
    {
        public Guid Id { get; private set; }
        public DateTimeOffset OccuredOn { get; private set; }

        protected IntegrationEvent()
        {
            this.Id = Guid.NewGuid();
            this.OccuredOn = DateTimeOffset.Now;
        }
    }

    public sealed class StudentRegisteredIntegrationEvent : IntegrationEvent
    {
        public Guid StudentId { get; set; }
        public string FullName { get; set; }

        public StudentRegisteredIntegrationEvent(Guid studentId, string fullName)
        {
            StudentId = studentId;
            FullName = fullName;
        }
    }

然后我创建了一个消费者：


 public sealed class StudentRegisteredConsumer: IConsumer<StudentRegisteredIntegrationEvent>
    {
        private readonly ILogger<StudentRegisteredConsumer> _logger;

        public StudentRegisteredConsumer(ILogger<StudentRegisteredConsumer> logger)
        {
            _logger = logger;
        }


        public Task Consume(ConsumeContext<StudentRegisteredIntegrationEvent> context)
        {
            _logger.LogWarning("========== Message Received +==========================");
            _logger.LogInformation($"Sending notification to {context.Message.FullName}");
            _logger.LogWarning("========== Message Received +==========================");
            return Task.CompletedTask;
        }
    }

在生产者方面，我有一个List<IntegrationEvent>列表，我publish通过IPublishEndpoint但它没有路由到正确的队列，而是创建另一个交换，Sample.Abstraction.Domain:IntegrationEvent. 我如何告诉 MassTransit 不要使用基类而使用真实类型类？我也尝试过，ISendEndpointProvider但它们再次被路由到另一个队列，student-registered-integration-event_skipped队列，因为该类没有可用的消费者base。

这是消费者端的日志：


[00:33:21 DBG] Configuring endpoint student-registered-integration-event, Consumer: Sample.University.Notification.Consumers.StudentRegisteredConsumer
[00:33:21 DBG] Declare exchange: name: student-registered-integration-event, type: fanout, durable
[00:33:21 DBG] Declare exchange: name: Sample.IntegrationEvents:StudentRegisteredIntegrationEvent, type: fanout, durable
[00:33:21 DBG] Bind exchange: source: Sample.IntegrationEvents:StudentRegisteredIntegrationEvent, destination: student-registered-integration-event
[00:33:21 DBG] Declare queue: name: student-registered-integration-event, durable
[00:33:21 DBG] Bind queue: source: student-registered-integration-event, destination: student-registered-integration-event
[00:33:21 DBG] Prefetch Count: 16
[00:33:21 DBG] Consumer Ok: rabbitmq://localhost/wrapperizer/student-registered-integration-event - amq.ctag-nqSrJ0A5UQZCXg3tIr9Hfg

我不知道如何配置，我也使用过ConsumerDefinition<StudentRegisteredConsumer>但无济于事，代码如下：

    public sealed class StudentRegisteredConsumerDefinition : ConsumerDefinition<StudentRegisteredConsumer>
    {
        public StudentRegisteredConsumerDefinition()
        {
            const string eventName = nameof(StudentRegisteredIntegrationEvent);
            var sanitized = KebabCaseEndpointNameFormatter.Instance.SanitizeName(eventName);
            this.EndpointName = sanitized;
        }
    }

在生产者端获取发送端点的 uri：

var eventName = logEvt.IntegrationEvent.GetType().Name;
                    var sanitized = KebabCaseEndpointNameFormatter.Instance.SanitizeName(eventName);
var uri = new Uri($"exchange:{sanitized}");
var sender = await _sendEndpointProvider.GetSendEndpoint(uri);
await sender.Send(logEvt.IntegrationEvent);

我知道上面的代码有点像 MT 的默认行为，但没有它，我就没有正确的队列和交换。任何解决方案将不胜感激。

标签： .net-corerabbitmqmasstransitrabbitmq-exchange

解决方案

首先，您可以完全使用 来完成此操作Publish，无需配置您为尝试解决此问题所做的所有事情。您可以按照约定配置您的消费者，并让他们在自己的端点上进行配置。您错过的一个部分是消息的发布方式。

从你的情况来看List<IntegrationEvent>，我怀疑你是在调用Publish<T>(T message)哪里T被推断出来的IntegrationEvent。这就是为什么您只在该交易所收到消息的原因。您需要使用Publish(object message)重载，以便确定适当的类型。

您可以简单地将列表中的每条消息分配给对象，然后Publish使用该对象进行调用。或者，您可以强制使用重载：

await Task.WhenAll(events.Select(x => bus.Publish((object)x, x.GetType()));

这样，MassTransit 将使用对象类型来调用适当的泛型重载。

推荐阅读