如何为 Azure 服务编写 MassTransit Json 反序列化程序答案

【问题标题】：How to write a MassTransit Json Deserializer for Azure Services如何为 Azure 服务编写 MassTransit Json 反序列化程序
【发布时间】：2020-03-25 04:45:56
【问题描述】：

这就是我将对象发布到事件网格的方式。我希望能够使用 azure 服务总线来收听它。

        public void Publicar<T>(T model, string operation, string entity)
    {
        _nomeEvento = entity + operation;

        Boolean.TryParse(Configuration["EventGridConfig:Enabled"], out var eventGridIsActive);
        if (!eventGridIsActive)
            return;

        var primaryTopicKey = Configuration["EventGridConfig:AcessKey"];
        var primaryTopic = Configuration["EventGridConfig:Endpoint"];

        var primaryTopicHostname = new Uri(primaryTopic).Host;

        var topicCredentials = new TopicCredentials(primaryTopicKey);
        var client = new EventGridClient(topicCredentials);

        client.PublishEventsAsync(primaryTopicHostname, GetEventsList(model)).GetAwaiter().GetResult();
    }

    private List<EventGridEvent> GetEventsList<T>(T model)
    {
        return new List<EventGridEvent>
        {
            new EventGridEvent()
            {
                Id = Guid.NewGuid().ToString(),
                EventType = _nomeEvento,
                Data = model,
                EventTime = DateTime.Now,
                Subject = "MS_Clientes",
                DataVersion = "1.0",
            }
        };
    }

这就是我连接到服务总线的方式

    static class CustomExtensionsMethods
{
    public static IServiceCollection AddBus(this IServiceCollection services, IConfiguration configuration,
        IHostingEnvironment env)
    {
        services.AddMassTransit(x => { x.AddConsumer<NomeEmailChangeConsumer>(); });
        services.AddSingleton(provider => Bus.Factory.CreateUsingAzureServiceBus(cfg =>
        {
            var keyName = "RootManageSharedAccessKey";
            var busName = configuration["ServiceBus:Name"];
            var secret = configuration["ServiceBus:Secret"];
            var host = cfg.Host(
                "Endpoint=sb://" + busName + ".servicebus.windows.net/;" +
                "SharedAccessKeyName=" + keyName + ";" +
                "SharedAccessKey=" + secret,
                z =>
                {
                    TokenProvider
                        .CreateSharedAccessSignatureTokenProvider(keyName, secret);
                });
            cfg.ConfigureJsonSerializer(settings =>
            {
                settings.Converters.Add(new InterfaceConverter());

                return settings;
            });
            cfg.UseExtensionsLogging(provider.GetService<ILoggerFactory>());
            cfg.ReceiveEndpoint(host, configuration["ServiceBus:Topic"],
                e => { e.Consumer<NomeEmailChangeConsumer>(provider); });
        }));
        services.AddSingleton<IPublishEndpoint>(provider => provider.GetRequiredService<IBusControl>());
        services.AddSingleton<ISendEndpointProvider>(provider => provider.GetRequiredService<IBusControl>());
        services.AddSingleton<IBus>(provider => provider.GetRequiredService<IBusControl>());
        services.AddScoped(provider => provider.GetRequiredService<IBus>().CreateRequestClient<NomeEmailChange>());
        services.AddSingleton<IHostedService, BusService>();
        return services;
    }
}

然后我得到同样的错误

    fail: MassTransit.Messages[0]
      R-FAULT sb://dev.servicebus.windows.net/bff-queue 9ade19ec-238c-4c08-8e03-28bac695ea7b No deserializer was registered for the message content type: application/json; charset=utf-8. Supported content types include application/vnd.masstransit+json, application/vnd.masstransit+bson, application/vnd.masstransit+xml
System.Runtime.Serialization.SerializationException: No deserializer was registered for the message content type: application/json; charset=utf-8. Supported content types include application/vnd.masstransit+json, application/vnd.masstransit+bson, application/vnd.masstransit+xml
   at MassTransit.Serialization.SupportedMessageDeserializers.Deserialize(ReceiveContext receiveContext)
   at MassTransit.Pipeline.Filters.DeserializeFilter.Send(ReceiveContext context, IPipe`1 next)
   at GreenPipes.Filters.RescueFilter`2.GreenPipes.IFilter<TContext>.Send(TContext context, IPipe`1 next)

我尝试添加一个我在网上找到的 JsonConverter，但没有成功

    public class InterfaceConverter : JsonConverter
    {
        public override void WriteJson(JsonWriter writer, object value, JsonSerializer serializer)
        {
            serializer.Serialize(writer, value);
        }

        public override object ReadJson(JsonReader reader, Type objectType, object existingValue,
            JsonSerializer serializer)
        {
            // Set TypeNameHandling to Auto for deserializing objects with $type
            // Should be set directly in ConfigureJsonDeserializer when setting up MT Service bus
            serializer.TypeNameHandling = TypeNameHandling.Auto;
            return serializer.Deserialize(reader);
        }

        public override bool CanConvert(Type objectType)
        {
            return objectType.IsInterface;
        }
    }

【问题讨论】：

您的情况并非一切都清楚。在您的代码示例中，我看到MassTransit，但您没有在任何地方提到它。你是故意用的吗？
我看到您正在尝试使用EventGridClient 写入事件网格并为服务总线进行一些MassTransit 注册。它不匹配或图片中的某些内容丢失。
弗拉德，我正在尝试将服务总线用作事件网格订阅处理程序。这是一个完全有效的场景。我收到事件就好了，我只需要一种反序列化的方法
我真的很想了解用例。我不知道MassTransit 是什么。我看到你写的是application/json，但读的是application/vnd.masstransit+json。应该有一个从你的问题中不清楚的原因。云请详细说明。

标签： json azure azureservicebus json-deserialization azure-eventgrid

【解决方案1】：

我尝试了几个测试并提出了可行的解决方案。在我的测试案例中，我将消息从 EventGridTopic 重定向到 ServiceBusQueue，就像在您的案例中一样 - 如果我理解得很好。

由于 MassTransit 要求消息采用特定格式才能对其进行解释，因此我们需要确保具备以下条件：

EventGridEvent 类型的事件网格消息的自定义反序列化器
确保 MassTransit 必须使用的所有消息具有 ContentType - 没有这个，它将无法工作

因此，我构建了一个示例，如果您从 EventGrid 重定向消息，但如果您将消息直接通过管道传输到服务总线，它也可以工作。以下代码是如何实现 EventGrid 消息的反序列化器的示例：

public class EventGridMessgeDeserializer : IMessageDeserializer
    {
        private string _contentType;

        public EventGridMessgeDeserializer(string contentType)
        {
            _contentType = contentType;
        }
        public ContentType ContentType => new ContentType(_contentType);

        public ConsumeContext Deserialize(ReceiveContext receiveContext)
        {
            var body = Encoding.UTF8.GetString(receiveContext.GetBody());
            var customMessage = JsonConvert.DeserializeObject<EventGridEvent>(body);
            var serviceBusSendContext = new AzureServiceBusSendContext<EventGridEvent>(customMessage, CancellationToken.None);

            // this is the default scheme, that has to match in order messages to be processed
            // EventGrid messages type of EventGridEvent within namespace Microsoft.Azure.EventGrid.Models
            string[] messageTypes = { "urn:message:Microsoft.Azure.EventGrid.Models:EventGridEvent" };
            var serviceBusContext = receiveContext as ServiceBusReceiveContext;
            serviceBusSendContext.ContentType = new ContentType(JsonMessageSerializer.JsonContentType.ToString());
            serviceBusSendContext.SourceAddress = serviceBusContext.InputAddress;
            serviceBusSendContext.SessionId = serviceBusContext.SessionId;

            // sending JToken because we are using default Newtonsoft deserializer/serializer
            var messageEnv = new JsonMessageEnvelope(serviceBusSendContext, JObject.Parse(body), messageTypes);
            return new JsonConsumeContext(JsonSerializer.CreateDefault(), receiveContext, messageEnv);
        }

        public void Probe(ProbeContext context)
        {
        }
    }

这里的重要部分是您在自定义反序列化器中指定消息类型是什么。由于 MassTransit 需要某种格式并忽略不符合要求的消息，因此我们在此处指定 MassTransit 所需的那条信息。

string[] messageTypes = { "urn:message:Microsoft.Azure.EventGrid.Models:EventGridEvent" }

这是默认方案，必须匹配才能处理消息

最后，您可以在 Github 上找到完整代码：https://github.com/kgalic/MassTransitSample

旁注：如果您将消息直接发送到 SB 队列并想要反序列化它们，如前所述，您需要指定 ContentType，如下所示：

var message = new Message(UTF8Encoding.UTF8.GetBytes(request));
message.ContentType = "application/json"; //must have
await _senderClient.SendAsync(message);

如果你有这样的事情，你需要编写类似于 EventGridEvent 的反序列化器，你可以使用它作为示例。

【讨论】：

【解决方案2】：

MassTransit 将消息封装在 message envelope as documented 中，由于事件网格消息不是该格式，因此出现错误。

使用了您设置的 JsonSerializer 设置，但反序列化时的预期对象是 MessageEnvelope this line。

我猜你有两种方法可以解决这个问题

~~创建并使用自定义反序列化器（类似于JsonMessageDeserializer），将其反序列化为简单的JObject 或Message~~

更新：在进一步尝试时，它似乎比我最初想的要复杂得多，就像 Richard 在另一个答案中提到的那样，您最好使用 Azure 服务总线客户端本身（如果需要，甚至是通用 AMQP 客户端）
触发 Azure 函数或逻辑应用程序将事件网格有效负载包装到 Message Envelope 中，以便 MassTransit 能够对其进行反序列化，然后将其发送到服务总线队列中

【讨论】：

【解决方案3】：

我认为问题在于您试图使用 MassTransit 作为客户端库来使用通用 json 编码消息，而实际上它是一个固执己见的应用程序框架，它使用特定消息定义并需要其自己的半私有编码方案信封结构：

https://masstransittemp.readthedocs.io/en/latest/advanced/interop.html

要使用 Mass Transit，您必须在发布和订阅组件中使用该库，然后使用 Azure 服务总线作为“传输”在它们之间中继消息。这些库将处理反序列化，并将使用消息信封中的数据执行路由和异常处理。因此，“vnd.masstransit”编码消息不打算由 EventGrid 等其他框架生成或使用。

从过去运行消息库的选择过程来看，我建议切换到基于通用协议（例如 AMQP 1.0）的客户端库（请参阅https://github.com/Azure/amqpnetlite），并定义您自己的路由和异常模式。

【讨论】：