Я пытаюсь использовать Azure ServiceBussessionProcessess ( azure.messaging.servicebus ) для процессов, связанных с отслеживанием сообщений из приложения. По разным причинам (чтобы не утомлять вас с подробностями), мы хотели бы переработать их по SessionId, чтобы избежать убийства базы данных. Это работает - но убивает производительность. < /p>
Если я пытаюсь сделать это из другого потока, я сразу же получаю следующую ошибку: < /p>
ошибка: "ServiceBusReceiver уже закрыт и не может выполнить
Запрашиваемая операция. \ R \ natbject name: 'ServiceBusRever'. /> Я интерпретирую свойство ServiceBussessionProcessoroPtions.AutoCOcleteMessages, так как мне следует разрешить завершить сообщение, как я считаю подходящим (конечно, внутри продолжительности блокировки), но по какой -то причине мне не разрешают. /> Mikael < /p>
Вот какой -то образец кода для демонстрационных целей < /p>
using Azure.Messaging.ServiceBus;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Threading.Tasks;
using System.Timers;
namespace ServiceBus.DLQ.Reader
{
public class Demo
{
public static async Task Main(string[] args)
{
DempoSessionProcessReader reader = new DempoSessionProcessReader(
new ServiceBusClient("Endpoint=sb://*****.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=*****"), "mikes.test.queue.session");
try
{
await reader.CreateSessionProcessors(10, 10, 10, 2, false);
//To stop!
Console.ReadKey();
return;
}
catch (Exception ex)
{
}
}
}
///
/// A class for grouping all messages for a particula session.
/// The timestamp is to keep track of when to start processing them (persist)
///
public class GroupedSession
{
public DateTime Timestamp { get; set; } //When this was created (i.e. when the first message for a message was added
public List Messages { get; set; }
public GroupedSession()
{
}
}
public class DempoSessionProcessReader
{
List processorList =new List();
List guids = new List();
public ServiceBusClient _client;
public string Queue { get; set; }
private static bool _automcomplete = false;
///
/// A list of messages with the sessionId as the key
///
public static IDictionary _messages = new Dictionary();
static object _lock = new object();
private delegate void FlowHandlerDelegate(string sessionId, ProcessSessionMessageEventArgs[] messages);
static Timer _timer = new Timer();
public DempoSessionProcessReader(ServiceBusClient client, string queue)
{
_client = client;
Queue = queue;
//Generate some guids for testing purposes which we can use when preloading the queue with messages
for(int i=0;i
Подробнее здесь: https://stackoverflow.com/questions/795 ... nprocesser
