Как надежно обрабатывать сообщения служебной шины Azure с помощью рабочего процесса без сохранения состояния с помощью стандартных приложений логики Azure

Question

У нас есть стандартное приложение логики с рабочим процессом без сохранения состояния. Триггер - «Когда сообщения доступны в очереди» для служебной шины Azure, а следующим шагом является цикл «Для каждого». Эта комбинация имеет недостаток из-за ограничений и приводит к двум проблемам.

1. Похоже, что триггер без сохранения состояния разрешает только автозаполнение, поэтому при ошибке все сообщения теряются.
2. Похоже, что триггер без сохранения состояния не позволяет настроить пакетную обработку, поэтому любой пакет, превышающий 100, приводит к следующей ошибке.

InvalidTemplate. Невозможно обработать выражения языка шаблонов для действия 'For_each' в строке '{line}' и столбце '{column}': 'Превышено ограничение на количество элементов foreach для действия' For_each': максимальное' 100 'и фактическое' {messageCount}'.'.

Мне что-то здесь не хватает, или рабочие процессы с отслеживанием состояния - единственный надежный способ обработки сообщений служебной шины Azure?

Answer 1

Я нигде не нашел его в документации Microsoft, но можно получить правильное поведение блокировки просмотра с помощью встроенного стандартного триггера приложения логики. Этот параметр не отображается в конструкторе приложений логики, но можно изменить workflow.json и изменить идентификатор операции на «peekLockTopicMessages». Это дает ожидаемое поведение блокировки просмотра. Я не уверен, почему это не является частью какого-либо документированного решения. Затем вы можете использовать встроенное действие http для вызова REST API служебной шины для завершения сообщения. Таким образом, нет необходимости использовать какие-либо управляемые действия или триггеры Azure.

      {
"definition": {
    "$schema": "https://schema.management.azure.com/providers/Microsoft.Logic/schemas/2016-06-01/workflowdefinition.json#",
    "actions": {
        "Placeholder_Logic": {
            "type": "Scope",
            "actions": {
                "Placeholder_for_logic": {
                    "type": "Wait",
                    "inputs": {
                        "interval": {
                            "count": 5,
                            "unit": "Second"
                        }
                    },
                    "runAfter": {}
                }
            },
            "runAfter": {}
        },
        "Complete_SB_message": {
            "type": "Http",
            "inputs": {
                "method": "DELETE",
                "uri": "https:/sb-namespace.servicebus.windows.net/topic-name/subscriptions/subscription-name/messages/@{triggerOutputs()?['body']?['messageId']}/@{triggerOutputs()?['body']?['lockToken']}",
                "authentication": {
                    "audience": "https://servicebus.azure.net",
                    "type": "ManagedServiceIdentity"
                }
            },
            "runAfter": {
                "Placeholder_Logic": [
                    "Succeeded"
                ]
            }
        }
    },
    "triggers": {
        "When_messages_are_available_in_a_topic_subscription_(Peek-Lock)": {
            "type": "ServiceProvider",
            "inputs": {
                "parameters": {
                    "topicName": "topic-name",
                    "subscriptionName": "subscription-name",
                    "isSessionsEnabled": false
                },
                "serviceProviderConfiguration": {
                    "connectionName": "serviceBus",
                    "operationId": "peekLockTopicMessages",
                    "serviceProviderId": "/serviceProviders/serviceBus"
                }
            },
            "splitOn": "@triggerOutputs()?['body']"
        }
    },
    "contentVersion": "1.0.0.0",
    "outputs": {}
},
"kind": "Stateful"

}

Answer 2

Мы протестировали в нашей местной среде, нижеприведенные утверждения основаны на нашем анализе.

1. Похоже, что триггер без сохранения состояния разрешает только автозаполнение, поэтому при ошибке все сообщения теряются.

Согласно документации Azure , очередь служебной шины поддерживает 2 триггера.

1. Когда сообщение получено в очереди (автозаполнение)
2. Когда сообщение получено в очереди (блокировка просмотра)

В приложениях логики на основе потребления у вас будет возможность выбрать любой из вышеперечисленных триггеров в вашем конструкторе логики.

Вот снимок экрана для справки:

Если вы используете стандартное приложение логики плана, независимо от рабочего потока с сохранением состояния или без него, у вас будет триггер очереди служебной шины, указанный ниже. When a message is received in a queue и у него нет возможности подтвердить, является ли он автозаполнением или блокировкой взгляда.

Вот снимок экрана для справки:

2. Похоже, что триггер без сохранения состояния не позволяет настроить пакетную обработку, поэтому любой пакет, превышающий 100, приводит к следующей ошибке.

InvalidTemplate. Невозможно обработать выражения языка шаблонов для действия 'For_each' в строке '{line}' и столбце '{column}': 'Превышено ограничение на количество элементов foreach для действия' For_each ': максимальное'100'и фактическое'{messageCount}'.'.

Согласно документации Azure , For_each loopв приложении логики поддерживает только 100 элементов по умолчанию, если вы используете рабочий процесс без сохранения состояния. он поддерживает 100000 элементов, если вы используете рабочие процессы с отслеживанием состояния.