Возможна ли "честная очередь" с JMS

Мне нужно реализовать систему честного ожидания, чтобы сообщения обрабатывались циклическим способом на основе значения некоторого заголовка сообщения для всех значений этого заголовка для сообщений, находящихся в настоящее время в очереди.

Сообщения в системе естественно сгруппированы по некоторому свойству, из которых существует много тысяч возможных значений, а набор значений для входящих в настоящее время сообщений изменяется со временем. Аналогами были бы сообщения с заголовком, который является миллисекундной частью времени, во время создания сообщения. Таким образом, заголовок будет иметь значение от 0 до 999, и будет некоторое распределение значения для всех сообщений, находящихся в настоящее время в очереди.

Мне нужно иметь возможность потреблять сообщения в таком порядке, чтобы никакое другое значение не было приоритетным по сравнению с любым другим. Если значения заголовков сообщений в очереди распределяются таким образом,

value | count
------|-------
  A   |   3
  B   |   3
  C   |   2

Тогда порядок потребления будет A,B,C,A,B,C,A,B.

Если сообщения с другим значением добавляются в очередь, они должны автоматически добавляться в циклическую последовательность.

Это подразумевает некоторое знание текущих сообщений в очереди, но не требует, чтобы знания были сохранены потребителем; у брокера могут быть механизмы для заказа доставки в некотором роде.

Допустимо, что существует некоторый порог, за которым начинается честная очередь. Это означает, что если порог равен 10, то приемлемо последовательно обрабатывать 10 сообщений с одинаковым значением, но обрабатываемое 11-м сообщение должно быть следующего значения в последовательности. Далее может быть одно и то же значение, если только сообщения с очередью имеют это значение.

Количество возможных значений, вероятно, исключает просто создание очереди для каждого и повторение очередей, хотя это еще не проверено.

Мы используем HornetQ, но если есть альтернативы, которые предоставляют эту семантику, я бы с удовольствием узнал.

Сообщения - это задания, а значения заголовка - идентификаторы пользователей. Что нужно искать, так это то, что в некоторых пределах никакие задания от какого-либо конкретного пользователя не будут чрезмерно задерживать работу у любого другого пользователя; Пользователь, производящий 1 миллион рабочих мест, не приводит к тому, что последующие задания от других пользователей будут ждать обработки этого миллиона рабочих заданий.

Потребители в очередях в HornetQ оцениваются в порядке создания, поэтому добавление выборочного потребителя в очередь не помешает любому пользователю, имеющему доступ ко всему, получать сообщения, соответствующие фильтру.

Группы JMS, похоже, не помогают, поскольку это связывает данную группу (пользователя?) с данным потребителем.

Потенциальное решение создает выборочных потребителей по теме, основанной на спросе (например: 10 последовательных сообщений от одного и того же пользователя), с чем-то, управляющим жизненным циклом всех выборочных потребителей, чтобы гарантировать, что catch-all не обрабатывает одно и то же сообщение, Хотя это возможно, похоже, имеет некоторые обременительные требования к синхронизации.

Ответ 1

Первым вариантом рассмотрения будет использование многопоточного приложения-потребителя. Предполагая поток на сеанс/потребитель, можно было бы установить синхронизацию или асинхронный прием с помощью селектора. Каждый селектор будет привязан к конкретному пользователю.

Работа с предположением, что JVM werereasonable справедливо с точки зрения отправки потоков (что я с удовольствием предполагаю) И не было никаких взаимоблокировок в коде приложения, я бы утверждал, что требования будут заполнены. Один поток может застрять с миллионами пользователей, остальные не будут затронуты.

Если, однако, однопоточное приложение желательно, то ничто в спецификации JMS и в нем не может помочь. Это может быть расширение поставщиков, которое может помочь. Однако другой вариант был бы для приложения, чтобы смотреть на каждое сообщение и помещать его в определенную очередь для идентификатора пользователя. Само приложение конечного потребления будет "обходить" между этими очередями, чтобы получить работу. Требуется другое приложение, но у вас очень детерминированная система.

Ответ 2

Я бы предложил использовать приоритет сообщения для этого параметра + userWindowSize = 0 и всегда иметь клиент для выбора нового сообщения с сервера.

У вас будет больше задержек в сообщениях, но у вас всегда будут сообщения, поступающие с сервера.

Обратите внимание, что есть расы, которые вам нужно будет рассмотреть. что, если вы потребляете сообщение C и сообщение B? вы затем будете потреблять C для последующего потребления B. Но это не так много, вы можете сделать это, так как C уже потребляется.

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

Ответ 3

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

Вместо этого, почему бы не поставить желаемый алгоритм очередей в ваше собственное приложение? Например: напишите приложение "добросовестный экспедитор (FQF)", который подписывается на все сообщения, в любом порядке от брокера. Попросите это приложение FQF использовать сообщения как можно быстрее, чтобы очередь брокера JMS всегда была пустой или почти пустой. Приложение FQF затем может сохранять сообщения в локальной очереди и повторно публиковать их по одному в любом порядке, который ваш желаемый алгоритм очередности определяет, в очередь или тему, на которую подписано приложение обработки сообщений. С этой целью вы, вероятно, захотите использовать транзакции или какой-то контроль потока, чтобы приложение FQF публиковало сообщения только по скорости, которую они могут обрабатывать конечной системой.

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

Таким образом, у вас будет полный контроль над порядком обработки сообщений, без необходимости каким-то образом "обмануть" брокера за то, что вы хотите. Вы бы использовали JMS просто как механизм доставки сообщений, поэтому вам не нужно писать свой собственный протокол передачи сообщений.