使用会话在服务总线主题中强制消息的 FIFO 传递答案

【问题标题】：Enforcing FIFO delivery of messages in Service Bus topic using sessions使用会话在服务总线主题中强制消息的 FIFO 传递
【发布时间】：2022-10-07 22:00:36
【问题描述】：

我的服务总线主题有一个发布者和一个消费者。我的目标是让消费者按照发布者将消息发布到主题的顺序接收消息。这是我将消息发布到启用会话的主题的代码：

public void Publish<T>(T messageObject)
{
    var jsonString = JsonSerializer.Serialize(messageObject);
    var message = new ServiceBusMessage(jsonString)
            {
                SessionId = "12345"
            };
    
    message.ApplicationProperties["messageType"] = typeof(T).Name;

    serviceBusSender.SendMessageAsync(message);
}

这是接收器的代码：

public async Task ReceiveWithSessionsAsync()
{
    var options = new ServiceBusSessionProcessorOptions
    {
        ReceiveMode = ServiceBusReceiveMode.ReceiveAndDelete,
        AutoCompleteMessages = true,
        MaxConcurrentSessions = 1,
        MaxConcurrentCallsPerSession = 1,
    };

    await using ServiceBusSessionProcessor processor = serviceBusClient.CreateSessionProcessor("my_topic", "my_subscription__session_enabled", options);

    processor.ProcessMessageAsync += MessageHandler;

    async Task MessageHandler(ProcessSessionMessageEventArgs args)
    {
        var body = args.Message.Body.ToString();
        logger.LogInformation($"received body: {body}");
    }

    // start processing
    await processor.StartProcessingAsync();
}

我给Publish打了几个电话，然后打电话给ReceiveWithSessionsAsync，但是，收到的消息顺序仍然与发送它们的顺序不同。

我在这里想念什么？

【问题讨论】：

没声音了你在公共回购中有复制品吗？
@SeanFeldman 给你：github.com/rezamanshouri/azure-service-bus
@SeanFeldman 奇怪的是，每次我运行应用程序时，对 publish-messages-with-session 的调用都会导致以随机顺序接收消息。但是，在随后的调用中似乎保留了该顺序。我也看到了非会话订阅的确切行为。
出于好奇，如果您将ServiceBusSessionProcessorOptions.SessionIds 限制在您的硬编码会话中，那是否有效？
@SeanFeldman 同样的事情，将SessionIds = { "some_session_ID" } 添加到ServiceBusSessionProcessorOptions，以及与我之前的评论相同的行为。

标签： c# azureservicebus azure-servicebus-topics

【解决方案1】：

您确定您在接收器中接受会话吗？如果您看一下this 示例，它会在接收消息之前调用“AcceptNextSessionAsync”方法，因此请确保您正在这样做。

Ps：我实际上想提交这个作为评论，但我还不能

【讨论】：

【解决方案2】：

在您的发送者中尝试 message.setSessionId("12345") 并在您的接收者中获取会话锁定。

var options = new ServiceBusSessionProcessorOptions
{
    // By default after the message handler returns, the processor will complete the message
    // If I want more fine-grained control over settlement, I can set this to false.
    AutoCompleteMessages = false,

    // I can also allow for processing multiple sessions
    MaxConcurrentSessions = 5,

   
    MaxConcurrentCallsPerSession = 2,

    // Processing can be optionally limited to a subset of session Ids.
    SessionIds = { "12345", "67890" },
};

// create a session processor that we can use to process the messages
await using ServiceBusSessionProcessor processor = client.CreateSessionProcessor(queueName, options);

【讨论】：