MailboxProcessor, работающий по логике LIFO.

tl;dr Я бы попробовал следующее: взять реализацию почтового ящика из FSharp.Actor или записи в блоге Зака ​​Брея, заменить ConcurrentQueue на ConcurrentStack (плюс добавить некоторую логику с ограниченной емкостью) и использовать этот измененный агент в качестве диспетчера для передачи сообщения от источника данных к армии процессоров данных, реализованных как обычные MBP или акторы.

tl;dr2 Если воркеры — это дефицитный и медленный ресурс и нам нужно обработать сообщение, которое является последним на момент готовности воркера, то вместо этого все сводится к агенту со стеком очереди (с некоторой логикой ограниченной емкости) плюс BlockingQueue рабочих процессов. Диспетчер удаляет готовый рабочий процесс из очереди, затем извлекает сообщение из стека и отправляет это сообщение рабочему процессу. После того, как задание выполнено, рабочий ставит себя в очередь, когда становится готовым (например, до let! msg = inbox.Receive()). Поток-потребитель-диспетчер затем блокируется до тех пор, пока какой-либо рабочий процесс не будет готов, в то время как поток-производитель обновляет ограниченный стек. (ограниченный стек можно сделать с помощью массива + смещения + размера внутри блокировки, ниже слишком сложно)

Подробности

MailBoxProcessor предназначен только для одного потребителя. Это даже прокомментировано в исходном коде MBP здесь (ищите слово "ДРАКОНЫ" :) )

Если вы отправляете свои данные в MBP, то только один поток может взять их из внутренней очереди или стека. В вашем конкретном случае я бы использовал ConcurrentStack напрямую или лучше завернутый в BlockingCollection:

• Это позволит многим одновременным потребителям
• Это очень быстро и безопасно для потоков
• BlockingCollection имеет свойство BoundedCapacity, позволяющее ограничить размер коллекции. Он выдает Add, но вы можете поймать его или использовать TryAdd. Если A является основным стеком, а B — резервным, то TryAdd к A, при ошибке Add к B и поменяйте местами два с помощью Interlocked.Exchange, затем обработайте необходимые сообщения в A, очистите его, создайте новый резерв или используйте три стека, если обработка A может занять больше времени. B может снова стать полным; таким образом вы не блокируете и не теряете никаких сообщений, но можете отбрасывать ненужные контролируемым способом.

BlockingCollection имеет такие методы, как AddToAny/TakeFromAny, которые работают с массивами BlockingCollection. Это может помочь, например:

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

Что-то вроде этого:

            (data stream produces 'T)
                |
            [dispatcher's BCSC]
                |
            (a dispatcher thread consumes 'T  and pushes to processors, manages capacity of BCCS and LRU queue)
                 |                               |
            [processor1's BCCS/Actor/MBP] ... [processorN's BCCS/Actor/MBP]
                 |                               |
               (process)                         (process)

Вместо ConcurrentStack вы можете прочитать о куче. структура данных. Если вам нужны ваши последние сообщения по какому-либо свойству сообщений, например. отметке времени, а не по порядку, в котором они поступают в стек (например, если могут быть задержки в пути и порядке прибытия ‹> порядок создания), вы можете получить последнее сообщение с помощью кучи.

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

• интересная статья Зака ​​Брея об эффективной реализации Актеров. Там вам нужно заменить (под комментарием // Might want to schedule this call on another thread.) строку execute true на строку async { execute true } |> Async.Start или аналогичную, потому что в противном случае производящий поток будет потреблять поток - не очень хорошо для одного быстрого производителя. Однако для описанного выше диспетчера это как раз то, что нужно.

• FSharp.Actor (он же Fakka) разработка ветка и исходный код FSharp MPB (первая ссылка выше) здесь могут быть очень полезны для деталей реализации. Библиотека FSharp.Actors была заморожена в течение нескольких месяцев, но в ветке разработки есть некоторая активность.

• Не следует пропускать обсуждение Fakka в группах Google в этом контексте.

У меня есть несколько похожий вариант использования, и за последние два дня я исследовал все, что мог найти в F# Agents/Actors. Этот ответ — своего рода TODO для себя, чтобы попробовать эти идеи, половина из которых родилась во время его написания.

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

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

Часть 1
Часть 2
Часть 3

Вы также можете проверить код на github.

Я также писал о проблемах со сканированием в моем скрытом ужасе пост

Надеюсь, это поможет...

Самое простое решение — жадно поедать все сообщения в папке «Входящие» по мере их поступления и удалять все, кроме самых последних. Легко сделать с помощью TryReceive:

let rec readLatestLoop oldMsg =
  async { let! newMsg = inbox.TryReceive 0
          match newMsg with
          | None -> oldMsg
          | Some newMsg -> return! readLatestLoop newMsg }
let readLatest() =
  async { let! msg = inbox.Receive()
          return! readLatestLoop msg }

Столкнувшись с той же проблемой, я разработал более сложное и эффективное решение, которое я назвал отменяемой потоковой передачей и описал в статье журнала F# здесь. Идея состоит в том, чтобы начать обработку сообщений, а затем отменить эту обработку, если они заменены. Это значительно улучшает параллелизм, если выполняется значительная обработка.