读取azure服务总线队列中的所有活动消息

Question

我是Azure服务总线的新手，我应该将消息推送到队列中，然后有一个单独的计划任务，该任务将读取该队列中的所有活动消息，并将它们批量导入到sql我以前尝试过这个代码，当我调用时它正在工作它在发送消息后立即工作，但现在它在单独的计划任务中不工作。任何帮助为什么或什么我可以用来批量阅读信息或这是不可能的

queueClient = new QueueClient(conn, queuename, ReceiveMode.ReceiveAndDelete);

                var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
                {
                    MaxConcurrentCalls = 1,
                    AutoComplete = false
                };
                queueClient.RegisterMessageHandler(ReceiveMessagesAsync, messageHandlerOptions);


public async Task ReceiveMessagesAsync(Message message, CancellationToken token)
        {
            messages.Add(message.Body.ToString());
            Console.WriteLine($"Received message: {Encoding.UTF8.GetString(message.Body)}");
            await queueClient.CompleteAsync(message.SystemProperties.LockToken);
        }

        public Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
        {
            Console.WriteLine(exceptionReceivedEventArgs.Exception);
            return Task.CompletedTask;
        }

Answer 1

这将是很好的知道你是使用什么库来做这件事，但我建议使用https://www.nuget.org/packages/Azure.Messaging.ServiceBus/nuget包，有了它，你可以：

class Program
{
    static async Task Main(string[] args)
    {
        string queueName = "myqueue";
        var client = new ServiceBusClient("myconn");
        // create a processor that we can use to process the messages
        var processor = client.CreateProcessor(queueName, new ServiceBusProcessorOptions());
        // add handler to process messages
        processor.ProcessMessageAsync += MessageHandler;
        // add handler to process any errors
        processor.ProcessErrorAsync += ErrorHandler;
        await processor.StartProcessingAsync();
        // Process messages for 5 minutes 
        await Task.Delay(TimeSpan.FromMinutes(5));
        // stop processing 
        Console.WriteLine("Stopping the receiver...");
        await processor.StopProcessingAsync();
        Console.WriteLine("Stopped receiving messages");
    }
    private static Task ErrorHandler(ProcessErrorEventArgs arg)
    {
        // Here you can catch errors;
        return Task.CompletedTask;
    }
    static async Task MessageHandler(ProcessMessageEventArgs args)
    {
        // Do something with the message .e.g deserialize it and insert to SQL
        try
        {
            BinaryData content = args.Message.Body;
            // Here you can use :
            string contentStr = content.ToString(); // This would be your data
        }
        catch (Exception e)
        {
            // If something goes wrong you should abandon the message
            await args.AbandonMessageAsync(args.Message);
        }
        await args.CompleteMessageAsync(args.Message);
    }
}

这将处理所有消息，直到时间结束。这将是一个很好的计划任务。如果您想在处理了特定数量的消息后停止，您可以执行异步循环并检查是否已达到该数量。