Многоагентная система, использующая шаблон производитель-потребитель?

Я пытаюсь реализовать шаблон Producer-Consumer, который использует несколько агентов в качестве рабочих вместо нескольких потоков.

Насколько я понимаю, типичная многопоточная реализация использует BlockingQueue, где один поток Producer помещает информацию в очередь, а несколько потоков Consumer извлекают данные и выполняют некоторые функции обработки.

Следуя той же логике, мой проект будет использовать агент Producer, который генерирует данные и отправляет их нескольким агентам Consumer. На первый взгляд, я подумал, что должен использовать общий BlockingQueue между агентами Consumer, чтобы агенты обращались к очереди и извлекали данные. Но я не знаю, легко ли это сделать, потому что я не думаю, что агенты имеют какую-либо общую память, и намного проще напрямую отправлять информацию агентам Consumer в виде сообщений ACL.

Это важно учитывать, потому что мой многоагентный дизайн будет обрабатывать много данных. Итак, мой вопрос: что произойдет в Jade, если я отправлю много сообщений ACL одному агенту? будет ли агент игнорировать другие сообщения?

В этом сообщении есть ответ, который предлагает ".. В среде JADE функция агентов «Входящие» для сообщений ACLMessages, в основном объект BlockingQueue, который содержит список полученных сообщений. Агент может наблюдать за своим собственным списком и обрабатывать их по ходу своего жизненного цикла. Контейнеры не имеют такой возможности..." . Это утверждение верно? Если это правда, то другие сообщения просто ожидают в очереди, и для моего проекта было бы идеально отправлять информацию напрямую агентам Consumer, но я не видел никаких BlockingQueues в классе ACLMessage.


java multithreading producer-consumer multi-agent agents-jade
person Community    schedule 23.06.2018    source источник
comment
посмотрите на jade.core.Agent#receive(). Вы увидите MessageQueue и синхронизированный, который блокирует очередь. Вы должны понимать, что если количество ваших сообщений превышает максимальный размер очереди (MessageQueue.getMaxSize), ваши первые сообщения будут удалены (InternalMessageQueue.addLast)   -  person nikelyn    schedule 25.06.2018
comment
Я понимаю, я попытался реализовать дизайн и заметил, что агенты становятся все медленнее, поскольку очередь заполняется другими сообщениями ACL. Я предполагаю, что это связано с размером или длиной очереди сообщений. Как я могу решить эту проблему ? Агенты-потребители работают медленно, потому что они обрабатывают ACLMessages, в то время как другие принимаются.   -  person    schedule 03.07.2018
comment
Обрабатывает ли агент потребителя только одно сообщение за раз? Я имею в виду на одной итерации в поведении цикла... Может быть, вам следует создать больше агентов-потребителей?   -  person nikelyn    schedule 03.07.2018
comment
Спасибо за ваш ответ. Агент producer считывает данные из файла и отправляет каждому агенту одну строку (первая строка первому агенту, вторая второму... и т. д.). Каждый агент consumer обрабатывает сообщение, как только оно получено. Я считаю, что агенты consumer отстают, потому что они не могут справиться с очередью сообщений, заполненной агентом producer. Это достаточно ясно? Итак, проблема моего кода в том, что он отправляет много сообщений нескольким агентам, я не знаю, как это решить/оптимизировать.   -  person    schedule 03.07.2018
comment
Вы пытались обрабатывать несколько сообщений одновременно в потребительском агенте? примерно так: int size = myAgent.getCurQueueSize(); если (размер == 0) {блок(); возвращение; } for(int i = 0; i ‹ size; i++){ Сообщение ACLMessage = myAgent.receive(); // код вашего процесса }   -  person nikelyn    schedule 07.07.2018

Ответы (1)


arrow_upward
1
arrow_downward

Да, сообщения будут стоять в очереди и агент не будет их игнорировать.

ACLMessage — это просто объект сообщения, который отправляется между агентами. У каждого агента есть собственная очередь сообщений (jade.core.MessageQueue) и несколько методов обработки связи.

Если вы проверите документацию класса Agent, вы можете найти такие методы, как

  • receive() - неблокирующий прием, возвращает первое сообщение в очереди или null, если очередь пуста
  • receive(MessageTemplate pattern) - ведет себя как предыдущий, но вы также можете указать шаблон для сообщения, например, конкретный AID отправителя, идентификатор разговора, а также комбинации.
  • blockingReceive() - блокировка приема, блокирует агента до тех пор, пока сообщение не появится в очереди
  • blockingReceive(MessageTemplate pattern) - блокировка приема, с шаблоном

а также есть методы блокировки приема, где можно установить таймаут.

Также важно отметить, что если вы определяете логику своего агента в классе Behaviour, вы также можете просто блокировать только поведение вместо блокировки всего агента.

ACLMessage msg = agent.receive();    
if (msg != null) {
    // your logic
} else {
    block();            
}

Разница в том, что метод block() внутри поведения просто помечает ваше поведение как заблокированное и удаляет его из пула активного поведения агента (он добавляется обратно в пул активного поведения, когда сообщение получено или поведение перезапускается методом restart()), позволяя выполнять поведение другого агента, и blockingReceive() полностью блокирует вашего агента и все его действия, пока он не получит сообщение.

person Wojciech Ćwiek    schedule 23.06.2018
comment
Спасибо за ваш ответ. Я использую класс CyclicBehaviour для обработки полученных сообщений ACL, как это рекомендуется. Это правда, что сообщения ACL блокируются в очереди и не игнорируются, но агенты постепенно замедляются по мере того, как очередь заполняется другими сообщениями ACL. Я предполагаю, что это связано с размером или длиной очереди сообщений. Как я могу решить эту проблему ? Агенты Consumer работают медленно, потому что они обрабатывают сообщения ACL, в то время как другие принимаются. - person ; 29.06.2018