在azure worker角色中使用QueueClient.OnMessage
我有一个Azure辅助角色,负责检查4个服务总线队列。 目前,我只是循环方法来手动检查队列。
while(true) { //loop through my queues to check for messages }
Azure SDK 2.0具有监听消息而不是轮询消息的能力。 但是我见过的每个例子都使用Console.ReadKey()的控制台应用程序。 有没有办法让工人角色坐下来等待消息呢?
我试过了:
public override void Run() { _queueProcessors.ForEach(x => x.OnMessage(Process); }
其中_queueProcessors是QueueClients的列表,而Process是处理消息的私有方法。 但是,worker角色会注册它们然后重新启动。
所以任何人都知道如何让队列客户端坐下来等待消息?
以下是此代码示例:
using Microsoft.ServiceBus; using Microsoft.ServiceBus.Messaging; using Microsoft.WindowsAzure.ServiceRuntime; using System.Diagnostics; using System.Net; using System.Threading; namespace WorkerRoleWithSBQueue1 { public class WorkerRole : RoleEntryPoint { // The name of your queue const string QueueName = "demoapp"; ManualResetEvent CompletedEvent = new ManualResetEvent(false); // QueueClient is thread-safe. Recommended that you cache // rather than recreating it on every request QueueClient Client; public override void Run() { OnMessageOptions options = new OnMessageOptions(); options.AutoComplete = true; // Indicates if the message-pump should call complete on messages after the callback has completed processing. options.MaxConcurrentCalls = 1; // Indicates the maximum number of concurrent calls to the callback the pump should initiate options.ExceptionReceived += LogErrors; // Allows users to get notified of any errors encountered by the message pump Trace.WriteLine("Starting processing of messages"); // Start receiveing messages Client.OnMessage((receivedMessage) => // Initiates the message pump and callback is invoked for each message that is recieved, calling close on the client will stop the pump. { try { // Process the message Trace.WriteLine("Processing Service Bus message: " + receivedMessage.SequenceNumber.ToString()); } catch { // Handle any message processing specific exceptions here } }, options); CompletedEvent.WaitOne(); } private void LogErrors(object sender, ExceptionReceivedEventArgs e) { if (e.Exception != null) { Trace.WriteLine("Error: " + e.Exception.Message); } } public override bool OnStart() { // Set the maximum number of concurrent connections ServicePointManager.DefaultConnectionLimit = 12; // Create the queue if it does not exist already Trace.WriteLine("Creating Queue"); string connectionString = "*** provide your connection string here***"; var namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString); if (!namespaceManager.QueueExists(QueueName)) { namespaceManager.CreateQueue(QueueName); } // Initialize the connection to Service Bus Queue Client = QueueClient.CreateFromConnectionString(connectionString, QueueName); Trace.WriteLine("Sending messages..."); // populate some messages for (int ctr = 0; ctr < 10; ctr++) { Client.Send(new BrokeredMessage()); } return base.OnStart(); } public override void OnStop() { // Close the connection to Service Bus Queue Client.Close(); CompletedEvent.Set(); // complete the Run function base.OnStop(); } }
}
- 如何递归加载Azure blob文件列表?
- Azure Active Directory回复URL未按预期工作
- Azure Service Bus消息泵是否真的是事件驱动的?
- Stackexchange.Redis超时和socketfailures
- 无法加载文件或程序集“Microsoft.AI.Web”或其依赖项之一。 该系统找不到指定的文件
- 在Visual Studio中调试x64 Azurefunction
- 如何在Windows Azure(工作器)角色中捕获未处理的exception
- 无法加载文件或程序集’System.Runtime.InteropServices.RuntimeInformation
- 未调用AssemblyResolve,并且在序列化期间抛出FileNotFoundException