Azure服务总线功能。批量接收消息

Question

我希望以批处理模式接收来自Azure ServiceBus主题的消息。

阅读https://docs.microsoft.com/en-us/Azure/Azure-functions/functions-best-practices时指出：

我有一个方法：

public static void Run([ServiceBusTrigger("mytopic name", "MySubscription",
AccessRights.Listen, Connection = TopicService.ConnectionStringName)]
string messages, TraceWriter logger)

这个方法是有效的，但它一次只需要一个消息。

根据Microsoft文档,我可以将其更改为：

public static void Run([ServiceBusTrigger("mytopic name", "MySubscription",
AccessRights.Listen, Connection = TopicService.ConnectionStringName)]
string[] messages, TraceWriter logger)

{
    "aggregator": {
        "batchSize": 10,
        "flushTimeout": "00:00:30"
    }
}

注意：主题和订阅已启用“启用批处理操作”设置。

我错过了什么？

Answer 1

这是我尝试的代码。检查一下，看看是否管用。

//---------------------------------------------------------------------------------
// Copyright (c) Microsoft Corporation. All rights reserved.  

using Microsoft.Azure;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ServiceBusTriggers
{
    class Settings
    {
        public const string TopicPath = "sbperftopicwithpartitions";
        public const string Subscription = "sub_1";
        public const string ContainerName = "sbperf-test-store2";

        internal Settings()
        {
            ServiceBusConnectionString = GetSetting("ServiceBusConnectionString", UserSettings.ServiceBusConnectionString);
            StorageAccountConnectionString = GetSetting("StorageAccountConnectionString", UserSettings.StorageAccountConnectionString);
            AzureWebJobsDashboardConnectionString = GetSetting("AzureWebJobsDashboardConnectionString", UserSettings.AzureWebJobsDashboardConnectionString);
            AzureWebJobsStorageConnectionString = GetSetting("AzureWebJobsStorageConnectionString", UserSettings.AzureWebJobsStorageConnectionString);

            NLogDatabaseConnectionString = GetSetting("NLogDatabaseConnectionString", UserSettings.NLogDatabaseConnectionString);

            PrefetchCount = GetSetting("PrefetchCount", 100);
            MaxConcurrentCalls = GetSetting("MaxConcurrentCalls",100);

            MetricsDisplayFrequency = new TimeSpan(0, 0, 30); //every 30 seconds
            TokenSource = new CancellationTokenSource();
        }

        private int GetSetting(string name, int defaultValue)
        {
            int value;
            string valueStr = CloudConfigurationManager.GetSetting(name);
            if (!int.TryParse(valueStr, out value))
            {
                Console.WriteLine("Config missing for {0}. Using default.",name);
                value = defaultValue;
            }
            return value;
        }

        private string GetSetting(string name, string defaultValue)
        {
            string valueStr = CloudConfigurationManager.GetSetting(name);
            if (string.IsNullOrEmpty(valueStr))
            {
                Console.WriteLine("Config missing for {0}. Using default.", name);
                valueStr = defaultValue;
            }
            return valueStr;
        }

        public string ServiceBusConnectionString { get; set; }

        public string StorageAccountConnectionString { get; set;  }
       
        public int PrefetchCount { get; set; }

        public int MaxConcurrentCalls { get; set; }

        public TimeSpan MetricsDisplayFrequency { get; internal set; }

        public CancellationTokenSource TokenSource { get; set; }

        public string NLogDatabaseConnectionString { get; private set; }

        public static string AzureWebJobsDashboardConnectionString { get; internal set; }

        public static string AzureWebJobsStorageConnectionString { get; internal set; }

        public void WriteSettings()
        {
            ProjectLogger.Info("1|None|{1}|DisplayFrequency|{0}|", MetricsDisplayFrequency, Thread.CurrentThread.ManagedThreadId);
            ProjectLogger.Info("1|None|{1}|PrefetchCount|{0}|", PrefetchCount, Thread.CurrentThread.ManagedThreadId);
            ProjectLogger.Info("1|None|{1}|MaxConcurrentCalls|{0}|", MaxConcurrentCalls, Thread.CurrentThread.ManagedThreadId);
        }

    }
}