Управление одновременным чтением из одного сокета в приложении .Net

Question

Я пишу многопоточное приложение .Net, в котором многим абонентам необходимо одновременно считывать данные из одного сокета.

Данные, прочитанные из сокета, разбиваются на записи / порции данных - каждая запись предназначенана определенном абоненте, идентифицированном CallerID в заголовке, так, например, поток байтов, прочитанных из сокета, мог бы быть чем-то вроде строк:

Header: CallerId = 1, Length = 5
Data:   "Hello"
Header: CallerId = 2, Length = 6
Data:   "Stack "
Header: CallerId = 1, Length = 7
Data:   " World!"
Header: CallerId = 2, Length = 8
Data:   "Overflow"

В этом случае поток данных, прочитанных абонентом1 должен читать «Hello World!», В то время как поток данных, считываемых вызывающей стороной 2, должен читаться как «Переполнение стека».

• Я думаю, что каждый поток данных вызывающей стороны должен быть открыт через некоторый класс, наследуемый отStream, чтобы вызывающие абоненты могли использовать как синхронные (Read), так и асинхронные (BeginRead) модели программирования.
• Я бы хотел избежать выделенного потока чтения для каждого сокета, так как мое приложение будетиметь дело со многими такими сокетами одновременно.
• Вызывающие не должны блокировать друг друга, например, яВ приведенном выше примере следующее не должно тупиковать:

<code>caller1Stream.Read(12) // Hello world!
caller2Stream.Read(14) // Stack Overflow

Я попытался выяснить, как это сделать, однако я обнаружил, что все быстров частности, усложнилось;поскольку данные должны читаться последовательно, и поэтому при чтении данных для вызывающих абонентов требуется буферизация - это значительно усложняет процесс чтения потока для данного вызывающего абонента:

• Читатели должны сначала проверитьих буфер, чтобы узнать, прочитал ли кто-то еще достаточно данных
  • Я изо всех сил пытаюсь найти подходящий буфер / поток памяти FIFO (уже существует или мне нужно написать один?)
• Если было прочитано недостаточно данных, читателю необходимо считывать данные из сокета до тех пор, пока не будет прочитано достаточно данных, предназначенных для данного вызывающего абонента, и буферизировать данные, предназначенные для других вызывающих абонентов.
  • Буферизация данных в порядке (я могу вести список читателей и искать правильный буфер по идентификатору вызывающего абонента), однако,
  • Как предотвратить взаимные блокировки, когда несколько потоков пытаются читать изсокет в то же время, однако считывать данные друг друга?
  • Как уведомить вызывающих абонентов, ожидающих данных, о том, что их данные были прочитаны другим абонентом?
• Как мне убедиться, что заголовок и блоки данных сообщений читаются вместе?(Блок данных имеет переменную длину, и поэтому необходимо сначала прочитать блок заголовка фиксированной длины, чтобы определить, сколько данных нужно прочитать)

Конечно, это должно быть довольно распространенной проблемой - этоне существует класса .Net, который может сделать все это намного проще?(Обратите внимание, что я не могу изменить транспортный механизм / протокол).Есть ли какой-нибудь способ использовать один из классов в стиле слушателя для выполнения вышеизложенного?

Если нет, как мне решить эти проблемы?

Answer 1

Исходя из вашей публикации, я предполагаю, что вы используете TCP / IP, так как вы имеете в виду потоковую передачу, но сокет может использоваться и для протокола без установления соединения UDP (и других). UDP очень прост, поскольку вы отправляете только фиксированные сообщения (без потоковой передачи).

В любом случае, поскольку вы имеете в виду получать поток сообщений, я должен предположить, что вы используете TCP / IP, и если вы используете сокет напрямую, у вас есть серьезные проблемы с тем, что вы ищете .

TCP / IP - потоковое решение, поэтому то, что я добавляю на одном конце, получается на другом, и оно всегда будет в порядке, но не обязательно вместе. Поэтому, если на одном конце вы напишите «Hello World», при чтении вы можете получить «Hello World», или «Hello» и «World», или даже «H» «ello World». Суть в том, что нет способа вызвать чтение и ожидать получения всего сообщения.

То, как я делал это в прошлом, это просто наличие промежуточного потока (или в моем случае я просто использовал пул потоков .net), который быстро читает сокет, повторно собирает сообщения и затем передает их в очередь для обработки. Хитрость заключается в повторной сборке, так как вы не знаете, что вы получите. Скажем, в вашем заголовке длина сообщения - int32, это 4 байта. Когда вы вызываете read, вам может даже понадобиться пересобрать эту длину, поскольку вы получили только первый байт из 4 байтов, которые вам нужны для определения длины.

Звучит так, как будто ваш заголовок имеет фиксированную длину, поэтому в основном то, что вам нужно сделать, предполагая, что callerid и length - это значения uint32. Попробуйте прочитать 8 байтов, если меньше 8, поместите в буфер, пока мы не прочитаем 8 байтов. Как только мы прочитаем 8 байтов, возьмем 4 байта и получим длину, выделим буфер для длины сообщения. Попробуйте и прочитайте, пока мы не заполним буфер, как только мы это сделаем, поместите это в очередь для определенного callerid и сообщите потоку, что есть новое сообщение, если оно спит. Затем возьмите все, что осталось, и начните сначала или дождитесь новых данных.

В прошлом я успешно использовал это, и это добавляет относительно немного накладных расходов.

Answer 2

Я не уверен, сколько у вас свободы выбора альтернатив, но вот две:

1. Используйте трубы вместо розеток. Откройте именованный канал Windows в режиме PIPE_READMODE_MESSAGE. Сообщения приходят только целыми кусками.
2. Используйте высокоуровневую абстракцию сокета, такую ​​как ZeroMQ . В итоге вы будете писать тысячи строк кода для сложных сетевых сообщений.