c# - 连续的C# WebJob可以自动处理单个ServiceBus队列中的不同消息类型吗

Question

我创建了一个 Azure WebJob 来处理单个 ServiceBus 队列中的两个不同的自定义消息。

自动反序列化是否支持此功能？

即我可以只创建两个函数，每个函数都具有相同的 ServiceBusTrigger 但消息参数不同:

public static void ProcessMessageA([ServiceBusTrigger("MessageQueue")] MessageA message) { ... }
public static void ProcessMessageB([ServiceBusTrigger("MessageQueue")] MessageB message) { ... }

我已经尝试过此操作，但它不起作用，所以我是否必须处理 BrokeredMessage 并根据需要手动反序列化和处理消息？

Answer 1

这里的问题是 webjob SDK 不知道哪条消息是 A 类型还是 B 类型。

因此，正如您所说，您可以选择手动反序列化中转消息。我认为您应该在代理消息上添加一个属性来指定消息的类型。

因此，要发送消息，可能是这样的:

// Create the object you want to send
var messageA = new MessageA();
... 

// Create the borkered message
var message = new BrokeredMessage(messageA);

// Add a property 
message.Properties["messageType"] = "MessageA";

// Send the message
...

所以现在你的网络作业中只有一个函数

public static void ProcessMessageA([ServiceBusTrigger("MessageQueue")] BrokeredMessage message)
{
    var messageType = (string) message.Properties["messageType"];
    if (messageType == "MessageA")
    {
        // Deserialize the message
        var messageA = message.GetBody<MessageA>();
        // Process the message
    }
    else if (messageType == "MessageB")
    {
        // Deserialize the message
        var messageB = message.GetBody<MessageB>();;
        // Process the message
    }
}

否则，您可以为每种类型的消息创建单独的队列或者创建一个service bus topic有两个订阅。

创建订阅时，您可以指定过滤器。

string connectionString = CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");

var namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);

// Create the topic
namespaceManager.CreateTopic("TestTopic");

// Create subscription to handle message A
namespaceManager.CreateSubscription("TestTopic", "MessageA", new SqlFilter("messageType = 'MessageA'"));
// Create subscription to handle message A
namespaceManager.CreateSubscription("TestTopic", "MessageB", new SqlFilter("messageType = 'MessageB'"));

现在您可以在 webjob 中使用两个不同的函数来处理 MessageA 和 MessageB。

public static void ProcessMessageA([ServiceBusTrigger("TestTopic", "MessageA")] MessageA message) { }

public static void ProcessMessageB([ServiceBusTrigger("TestTopic", "MessageB")] MessageB message) { }