Найти повторяющееся сообщение в темах служебной шины

Question

Я хочу найти дубликаты сообщений в ServiceBusTopics. Ниже приведена моя логика. Если я нахожу дубликаты сообщений, которые добавляются в список и отправляются в другую тему служебной шины.Но я не смог прочитать все сообщения служебной шины, которые присутствуют в подписке, перед циклом просмотра списка. Я хочу прочитать все сообщения из sevicebus, а затем просмотреть его. Я не уверен, какДолжен ли я остановить выполнение циклов, пока я не прочитал все сообщения от suscription

    private static string _serviceBusConn = "servicebusconnectionstring";
    private static string _serviceBustopic = "topic1";
    private static string _topic = "test_v1";

    static void Main(string[] args)
    {
        IList<string> items = new List<string>();
        int count; 
        IList<string> itemsRepeated = new List<string>();

        var subClient = SubscriptionClient.CreateFromConnectionString(_serviceBusConn, _serviceBustopic, "DevTest");

        subClient.OnMessage(m =>
        {
            Stream stream = m.GetBody<Stream>();
            StreamReader reader = new StreamReader(stream);
            string s = reader.ReadToEnd();

            Console.WriteLine(s);

            items.Add(s);

        });


        List<string> copy1= new List<string>(items);
        List<string> copy2 = new List<string>(items);


        foreach (var item in copy1)
        {
            count = 0;
            foreach (var itemtopic in copy2)
            {
                count++;
                Console.WriteLine("{0}{1}{2}", items, itemtopic, count);
                if (item.Equals(itemtopic))
                {
                    count++;
                }

                if (count > 1)
                {
                    Console.WriteLine(count);
                    itemsRepeated.Add(itemtopic);
                }
            }
        }
        foreach (var repeateditem in itemsRepeated)
        {
            SendMessage(repeateditem);
        }

    }

    static void SendMessage(string message)
        {
            var topicClient = TopicClient.CreateFromConnectionString(_serviceBusConn, _topic);
            var msg = new BrokeredMessage(message);
            topicClient.Send(msg);
        }

Answer 1

 if (count > 1)
                {
                    Console.WriteLine(count);
                    itemsRepeated.Add(itemtopic);
                }

Вы можете разорвать цикл, используя разрыв, как это

   if (count > 1)
                        {
                            Console.WriteLine(count);
                            itemsRepeated.Add(itemtopic);
                            break;
                        }

Answer 2

Вы можете использовать метод ReceiveBatch (Int32) для получения сообщений из тематической подписки в пакете. Вы можете получить количество сообщений в подписке на тему, используя метод Getsubscription () менеджера пространства имен.

var namespaceManager = NamespaceManager.CreateFromConnectionString(connnectionString);
var subscriptionDescription = namespaceManager.GetSubscription(topicName, subscriptionName);
var totalMessageCount= subscriptionDescription .MessageCount;

Теперь вы можете вызывать ReceiveBatch (imumMessageCount) внутри цикла и завершать цикл, когда количество полученных сообщений достигает totalMessageCount.

int receivedMessageCount = 0;
List<BrokeredMessage> MessageList = new List<BrokeredMessage>();
do{
var messageList = subClient.ReceiveBatch(100);
receivedMessageCount += messageList.Count;
MessageList.AddRange(messageList); 
}while(receivedMessageCount < totalMessageCount);

Теперь в MessageList будут все сообщения в подписке на тему. Вы можете использовать свою собственную логику в списке сообщений для обнаружения дубликатов и переадресовать ее в другую подписку.

Примечание. Чтобы получать все сообщения из ServiceBusTopic, необходимо получать все сообщения от всех подписок на темы в этой теме.