Возможна ли «справедливая очередь» с помощью JMS

Question

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

Сообщения в системе естественно сгруппированы по некоторому свойству, из которых существует много тысяч возможных значений, а набор значений для входящих в настоящее время сообщений изменяется со временем. Аналогами были бы сообщения с заголовком, который является миллисекундной частью времени, во время создания сообщения. Таким образом, заголовок будет иметь значение от 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 последовательных сообщений от одного и того же пользователя), с чем-то управляющим жизненным циклом всех выборочных потребителей, чтобы гарантировать, что уловки не обрабатывают одинаковые сообщение. Хотя это возможно, похоже, имеет некоторые обременительные требования к синхронизации.

Answer 1

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

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

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

Answer 2

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

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

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

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

Answer 3

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

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

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

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