0
我們有一個使用輔助角色處理從Azure服務總線上設置的主題接收的消息的雲服務。Azure主題工作角色在60秒後停止處理消息
消息本身似乎完好無損，通常會被正確接收和處理。然而，在某些情況下，該消息似乎停止處理（記錄突然結束，並且在我們的WadLogsTable中沒有更多對正在處理的消息的引用）。從我的研究來看，這可能是由於工作人員的角色將其連接打開並閒置超過幾秒鐘而發生的。我將如何去阻止這些長時間處理的消息被放棄？
我們工人角色的代碼如下。
public class WorkerRole : RoleEntryPoint
{
private static StandardKernel _kernel;
private readonly ManualResetEvent _completedEvent = new ManualResetEvent(false);
private BaseRepository<CallData> _callDataRepository;
private BaseRepository<CallLog> _callLogRepository;
private SubscriptionClient _client;
private NamespaceManager _nManager;
private OnMessageOptions _options;
private BaseRepository<Site> _siteRepository;
public override void Run()
{
try
{
List<CallInformation> callInfo;
Trace.WriteLine("Starting processing of messages");
// Initiates the message pump and callback is invoked for each message that is received, calling close on the client will stop the pump.
_client.OnMessage(message =>
{
// Process message from subscription.
Trace.TraceInformation("Call Received. Ready to process message ");
message.RenewLock();
callInfo = message.GetBody<List<CallInformation>>();
writeCallData(callInfo);
Trace.TraceInformation("Call Processed. Clearing from topic.");
}, _options);
}
catch (Exception e)
{
Trace.TraceInformation("Error: " + e.Message + "---" + e.StackTrace);
}
}
private void writeCallData(List<CallInformation> callList)
{
try
{
Trace.TraceInformation("Calls received: " + callList.Count);
foreach (var callInfo in callList)
{
Trace.TraceInformation("Unwrapping call...");
var call = callInfo.CallLog.Unwrap();
Trace.TraceInformation("Begin Processing: Local Call " + call.ID + " with " + callInfo.DataPoints.Length + " datapoints");
Trace.TraceInformation("Inserting Call...");
_callLogRepository.ExecuteSqlCommand(/*SNIP: Insert call*/);
Trace.TraceInformation("Call entry written. Now building datapoint list...");
var datapoints = callInfo.DataPoints.Select(datapoint => datapoint.Unwrap()).ToList();
Trace.TraceInformation("datapoint list constructed. Processing datapoints...");
foreach (var data in datapoints)
{
/*SNIP: Long running code. Insert our datapoints one at a time. Sometimes our messages die in the middle of this foreach. */
}
Trace.TraceInformation("All datapoints written for call with dependable ID " + call.Call_ID);
Trace.TraceInformation("Call Processed successfully.");
}
}
catch (Exception e)
{
Trace.TraceInformation("Call Processing Failed. " + e.Message);
}
}
public override bool OnStart()
{
try
{
var connectionString = CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
_nManager = NamespaceManager.CreateFromConnectionString(connectionString);
_nManager.Settings.OperationTimeout = new TimeSpan(0,0,10,0);
var topic = new TopicDescription("MyTopic")
{
DuplicateDetectionHistoryTimeWindow = new TimeSpan(0, 0, 10, 0),
DefaultMessageTimeToLive = new TimeSpan(0, 0, 10, 0),
RequiresDuplicateDetection = true,
};
if (!_nManager.TopicExists("MyTopic"))
{
_nManager.CreateTopic(topic);
}
if (!_nManager.SubscriptionExists("MyTopic", "AllMessages"))
{
_nManager.CreateSubscription("MyTopic", "AllMessages");
}
_client = SubscriptionClient.CreateFromConnectionString(connectionString, "MyTopic", "AllMessages",
ReceiveMode.ReceiveAndDelete);
_options = new OnMessageOptions
{
AutoRenewTimeout = TimeSpan.FromMinutes(5),
};
_options.ExceptionReceived += LogErrors;
CreateKernel();
_callLogRepository.ExecuteSqlCommand(/*SNIP: Background processing*/);
}
catch (Exception e)
{
Trace.TraceInformation("Error on roleStart:" + e.Message + "---" + e.StackTrace);
}
return base.OnStart();
}
public override void OnStop()
{
// Close the connection to Service Bus Queue
_client.Close();
_completedEvent.Set();
}
void LogErrors(object sender, ExceptionReceivedEventArgs e)
{
if (e.Exception != null)
{
Trace.TraceInformation("Error: " + e.Exception.Message + "---" + e.Exception.StackTrace);
_client.Close();
}
}
public IKernel CreateKernel()
{
_kernel = new StandardKernel();
/*SNIP: Bind NInjectable repositories */
return _kernel;
}
}
當運行完成後，我觀察到worker角色等待幾秒鐘後再次調用OnStart並重新進入Run（） –