Как управлять интенсивным трафиком с Anypoint MQ?

<!-- Producer Flow: HTTP → Anypoint MQ -->
<flow name="order-producer-flow">

  <http:listener
    path="/orders"
    allowedMethods="POST"
    config-ref="HTTP_Listener_config"/>

  <!-- Валидировать сообщение -->
  <validation:is-not-null
    value="#[payload.orderId]"
    message="orderId обязателен"/>

  <!-- Опубликовать в MQ -->
  <anypoint-mq:publish
    config-ref="Anypoint_MQ_Config"
    destination="orders-exchange"
    messageId=#[payload.orderId]>
    <anypoint-mq:properties>
      <anypoint-mq:property
        key="priority"
        value=#[payload.priority default 'NORMAL']/>
      <anypoint-mq:property
        key="correlationId"
        value=#[correlationId]/>
    </anypoint-mq:properties>
  </anypoint-mq:publish>

  <!-- Немедленно вернуть 202 Accepted -->
  <set-payload
    value=#[output application/json --- {status: "queued", messageId: payload.orderId}]/>
  <http:response statusCode="202"/>

</flow>

<flow name="order-consumer-flow">

  <anypoint-mq:subscriber
    config-ref="Anypoint_MQ_Config"
    destination="orders-queue"
    maxConcurrency="10"
    ackMode="MANUAL"/>

  <!-- Идемпотентность: не обрабатывать одно сообщение дважды -->
  <os:retrieve
    config-ref="ObjectStore_Config"
    key=#[attributes.messageId]
    target="alreadyProcessed"
    defaultValue="false"/>

  <choice>
    <when expression=#[vars.alreadyProcessed == false]>
      <flow-ref name="process-order-subflow"/>
      <os:store
        config-ref="ObjectStore_Config"
        key=#[attributes.messageId]
        value="true"
        ttl="86400"
        ttlUnit="SECONDS"/>
      <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
    </when>
    <otherwise>
      <!-- Дубликат: только ACK, без обработки -->
      <logger message="Дубликат сообщения пропущен: #[attributes.messageId]"/>
      <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
    </otherwise>
  </choice>

</flow>

<flow name="order-consumer-flow">

  <anypoint-mq:subscriber
    destination="orders-queue"
    ackMode="MANUAL"/>

  <try>
    <flow-ref name="process-order-subflow"/>
    <anypoint-mq:ack ackToken=#[attributes.ackToken]/>

    <error-handler>
      <!-- Временная ошибка: NACK, возврат в очередь -->
      <on-error-continue type="CONNECTIVITY, TIMEOUT">
        <logger message="Временная ошибка, NACK: #[error.description]" level="WARN"/>
        <anypoint-mq:nack ackToken=#[attributes.ackToken]/>
      </on-error-continue>

      <!-- Постоянная ошибка: отправить напрямую в DLQ -->
      <on-error-continue type="VALIDATION, TRANSFORMATION">
        <anypoint-mq:publish destination="orders-dlq">
          <anypoint-mq:properties>
            <anypoint-mq:property key="errorReason" value=#[error.description]/>
            <anypoint-mq:property key="originalMessageId" value=#[attributes.messageId]/>
          </anypoint-mq:properties>
        </anypoint-mq:publish>
        <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
      </on-error-continue>
    </error-handler>
  </try>

</flow>

// Заказ из MQ → трансформация в формат SAP
%dw 2.0
output application/xml

var priorityMap = {
  "HIGH":   "01",
  "NORMAL": "02",
  "LOW":    "03"
}
---
SALESORDER: {
  HEADER: {
    ORDER_ID:    payload.orderId,
    CUSTOMER:   payload.customerId,
    PRIORITY:   priorityMap[attributes.properties."priority" default "NORMAL"],
    CREATED_AT: now() as String { format: "yyyyMMddHHmmss" },
    EXT_REF:    attributes.properties."correlationId" default ""
  },
  ITEMS: {
    (payload.items map (item, idx) -> {
      ITEM: {
        LINE_NO:  (idx + 1) * 10,
        MATERIAL: item.sku,
        QTY:      item.quantity,
        UNIT:     item.unit default "EA"
      }
    })
  }
}

<anypoint-mq:subscriber
  config-ref="Anypoint_MQ_Config"
  destination="orders-queue"
  <!-- Сколько сообщений обрабатывается параллельно -->
  maxConcurrency="20"
  <!-- Сколько сообщений забирать за один polling (1-10) -->
  fetchSize="10"
  <!-- Сколько ждать при пустой очереди (мс) -->
  pollingTime="1000"
  <!-- Таймаут обработки сообщения -->
  acknowledgementTimeout="60000"
  <!-- Режим ручного ACK -->
  ackMode="MANUAL"/>