Событийные актеры в Groovy

Реализация возможности динамического присоединения и отсоединения акторов и потоков является сложной задачей. Однако оказалось, что жизнь намного проще, если вы можете стоять на плечах гигантов — пакета java.util.concurrent и Groovy в моем случае. Чтобы дать вам представление о том, что теперь возможно с актерами на основе событий в GParallelizer , я подготовил несколько примеров:

import static org.gparallelizer.actors.pooledActors.PooledActors.*

final def decryptor = actor {
    loop {
        react {String message->
            reply message.reverse()
        }
    }
}.start()

actor {
    decryptor.send 'suonorhcnysa si yvoorG'
    react {
        println 'Decrypted message: ' + it
    }
}.start()

Как вы можете видеть, вы создаете новых акторов с помощью метода actor (), передающего тело актера в качестве параметра замыкания. Внутри тела актера вы можете использовать loop () для итерации, реагировать () для получения сообщений и ответить (), чтобы отправить сообщение актеру, который отправил вам обработанное в данный момент сообщение. С помощью метода start () вы планируете обработчик для обработки в базовом пуле потоков.

Вот важная часть: когда актер расшифровщика не находит сообщение в своей очереди сообщений, методact () оставляет поток и возвращает его обратно в пул потоков, чтобы другие участники могли его забрать. Только после того, как новое сообщение поступит в очередь сообщений субъекта, закрытие метода реагирования () запланировано для обработки с пулом. Актеры, основанные на событиях, внутренне имитируют продолжения — работа актера разбивается на последовательно запускаемые фрагменты, которые вызываются, как только сообщение становится доступным во входящей почте. Каждый кусок для одного актера может быть выполнен другим потоком из пула потоков.

Простой калькулятор

Другой пример
актера, управляемого событиями, — калькулятор, который получает два числовых сообщения, суммирует их и отправляет результат в консоль субъекта. Я намеренно решил установить размер пула равным 1, чтобы показать, что даже однопотоковый пул может обрабатывать несколько участников. 

import static org.gparallelizer.actors.pooledActors.PooledActors.*

//not necessary, just showing that a single-threaded pool can still handle multiple actors
getPool().initialize 1

final def console = actor {
    loop {
        react {
            println 'Result: ' + it
        }
    }
}.start()

final def calculator = actor {
    react {a ->
        react {b ->
            console.send(a + b)
        }
    }
}.start()

calculator.send 2
calculator.send 3

Обратите внимание, что субъекты, управляемые событиями, требуют особого внимания в отношении метода реагирования () . Так как актеры , управляемые события , необходимо разделить код на независимые куски назначаемых для разных потоков последовательно и продолжений не изначально поддерживаются на JVM, чанки созданы искусственно за кулисами с Runnable задач и исключениями. В результате методы реагировать () и цикл () никогда не возвращаются нормально, и код акторов должен быть соответствующим образом структурирован. Я верю, что программисты Scala увидят, откуда я черпал вдохновение.

Реагируют () метод позволяет таймаут быть указан с помощью TimeCategory DSL:

import static org.gparallelizer.actors.pooledActors.PooledActors.*

def me = actor {
    friend.send('Hi')
    react(10.seconds) {message ->
        //continue conversation
    }
}

me.metaClass.onTimeout = {->friend.send('I see, busy as usual. Never mind.')}