scala – 阿卡事件匯流排教程

不知道是否有或沒有任何好的教程,但我可以給你一個可能的使用者案例的快速例子,使用事件流可能是有幫助的.但是,在高級別上,事件流是滿足您的應用可能具有的pub / sub型別要求的良好機制.假設您有一個用例來更新系統中的使用者餘額.平衡經常被訪問,所以你決定快取它以獲得更好的效能.當餘額更新時,您還需要檢視使用者是否跨越了餘額的門檻,如果是這樣,請傳送電子郵件.您不想將快取丟棄或平衡閾值檢查直接繫結到主平衡更新呼叫中,因為它們可能重量很大,並且減慢使用者的響應.您可以對這樣的特定要求進行建模：

//Message and event classes
case class UpdateAccountBalance(userId:Long, amount:Long)
case class BalanceUpdated(userId:Long)

//Actor that performs account updates
class AccountManager extends Actor{
val dao = new AccountManagerDao

def receive = {
case UpdateAccountBalance(userId, amount) =>
val res = for(result <- dao.updateBalance(userId, amount)) yield{
context.system.eventStream.publish(BalanceUpdated(userId))
result
}

sender ! res
}
}

//Actor that manages a cache of account balance data
class AccountCacher extends Actor{
val cache = new AccountCache

override def preStart = {
context.system.eventStream.subscribe(context.self, classOf[BalanceUpdated])
}

def receive = {
case BalanceUpdated(userId) =>
cache.remove(userId)
}
}

//Actor that checks balance after an update to warn of low balance
class LowBalanceChecker extends Actor{
val dao = new LowBalanceDao

override def preStart = {
context.system.eventStream.subscribe(context.self, classOf[BalanceUpdated])
}

def receive = {
case BalanceUpdated(userId) =>
for{
balance <- dao.getBalance(userId)
theshold <- dao.getBalanceThreshold(userId)
if (balance < threshold)
}{
sendBalanceEmail(userId, balance)
}
}
}

在這個例子中,AccountCacher和LowBalanceChecker參與者根據BalanceUpdated事件的類型別來訂閱eventStream.如果事件釋出到流,則將由兩個這些actor例項接收.然後,在AccountManager中,當餘額更新成功時,會為使用者引發一個BalanceUpdated事件.當這種情況發生時,並行地將該訊息傳遞給AccountCacher和LowBalanceChecker的郵箱,導致餘額從快取中刪除並檢查帳戶閾值,並且可能傳送電子郵件.

現在,您可以直接將(！)呼叫到AccountManager中直接與其他兩個演員進行通訊,但是可以認為可能會太平滑地將這兩種“副作用”耦合到平衡更新中,而這些型別的細節不一定屬於AccountManager.如果您有一些條件可能會導致一些額外的事情(檢查,更新等),需要純粹作為副作用發生(不是核心業務流程本身的一部分),則事件流可能是一種很好的方式解除事件的發生和誰可能需要對事件做出反應.