2. 程式人生 > >Scala筆記整理（九）：Actor和AKKA

Scala筆記整理（九）：Actor和AKKA

阿新 • • 發佈：2018-04-24
大數據 Scala

[TOC]

概述

? Scala的Actor有點類似於Java中的多線程編程。但是不同的是，Scala的Actor提供的模型與多線程有所不同。Scala的Actor盡可能地避免鎖和共享狀態，從而避免多線程並發時出現資源爭用的情況，進而提升多線程編程的性能。

Spark中使用的分布式多線程框架，是Akka，是Scala的一種多線程的類庫。Akka也實現了類似Scala Actor的模型，其核心概念同樣也是Actor。Scala Actor模型已經在2.1.0的時候還在用，但是在2.1.1的時候已經被遺棄了，Spark開始轉換用AKKA來替代Scala Actor，但是Scala Actor概念和原理都還是相同的。所以學習Scala Actor對我們學習AKKA，Spark還是有所幫助的

之所以學習Scala Actor,AKKA是因為在學習Spark源碼的時，我們能看懂Spark的源碼，因為在底層消息傳遞機制上大量使用AKKA的傳送機制。

scala actor

在使用前，需要先引入maven依賴：

<!--scala actor-->
<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-actors</artifactId>
    <version>2.10.5</version>
</dependency>

actor單向通信

測試代碼如下：

package cn.xpleaf.bigdata.p5.myactor
import scala.actors.Actor
/**
  * 學習scala actor的基本操作
  * 和java中的Runnable Thread幾乎一致
  *
  * 第一步：編寫一個類，擴展特質trait Actor（scala 的actor）
  * 第二步：復寫其中的act方法
  * 第三步：創建該actor的對象，調用該對象的start()方法，啟動該線程
  * 第四步：通過scala的操作符"!"，發送消息
  * 第五步：結束的話，調用close即可
  *
  * 模擬單向打招呼
  */
object ActorOps {
    def main(args: Array[String]): Unit = {
        val mFActor = new MyFirstActor()
        mFActor.start()
        // 發送消息
        mFActor ! "小美，睡了嗎？"
        mFActor ! "我去洗澡了~"
        mFActor ! "呵呵"
    }
}

class MyFirstActor extends Actor {
    override def act(): Unit = {
        while(true) {
            receive {
                case str: String => println(str)
            }
        }
    }
}

輸出結果如下：

小美，睡了嗎？
我去洗澡了~
呵呵

使用樣例類（case class）進行actor消息傳遞

測試代碼如下：

package cn.xpleaf.bigdata.p5.myactor

import scala.actors.Actor

/**
  *
  */
object GreetingActor {
    def main(args: Array[String]): Unit = {
        val ga = new GreetingActor
        ga.start()

        ga ! Greeting("小美")
        ga ! WorkContent("裝系統")
    }
}

case class Greeting(name:String)
case class WorkContent(content:String)

class GreetingActor extends Actor {
    override def act(): Unit = {
        while(true) {
            receive {
                case Greeting(name) => println(s"Hello, $name")
                case WorkContent(content) => println(s"Let‘s talk about sth. with $content")
            }
        }
    }
}

輸出結果如下：

Hello, 小美
Let‘s talk about sth. with 裝系統

actor相互通信

測試代碼如下：

package cn.xpleaf.bigdata.p5.myactor

import scala.actors.Actor

/**
  * actor之線程間，互相通信
  *
  * studentActor
  *     向老師問了一個問題
  *
  * teacherActor
  *     向學生做回應
  *
  * 通信的協議：
  * 請求，使用Request(內容)來表示
  * 響應，使用Response(內容)來表示
  */
object _03CommunicationActorOps {
    def main(args: Array[String]): Unit = {
        val teacherActor = new TeacherActor()
        teacherActor.start()
        val studentActor = new StudentActor(teacherActor)
        studentActor.start()

        studentActor ! Request("老李啊，scala學習為什麽這麽難啊")
    }
}

case class Request(req:String)
case class Response(resp:String)

class StudentActor(teacherActor: TeacherActor) extends Actor {
    override def act(): Unit = {
        while(true) {
            receive {
                case Request(req) => {
                    // 向老師請求相關的問題
                    println("學生向老師說：" + req)
                    teacherActor ! Request(req)
                }
                case Response(resp) => {
                    println(resp)
                    println("高！")
                }
            }
        }
    }
}

class TeacherActor() extends Actor {
    override def act(): Unit = {
        while (true) {
            receive {
                case Request(req) => {  // 接收到學生的請求
                    sender ! Response("這個問題，需要如此搞定~")
                }
            }
        }
    }
}

輸出結果如下：

學生向老師說：老李啊，scala學習為什麽這麽難啊
這個問題，需要如此搞定~
高！

消息的同步和Future

1、Scala在默認情況下，消息都是以異步進行發送的；但是如果發送的消息是同步的，即對方接受後，一定要給自己返回結果，那麽可以使用!?的方式發送消息。即：

val response= activeActor !? activeMessage

2、如果要異步發送一個消息，但是在後續要獲得消息的返回值，那麽可以使用Future。即!!語法，如下：

val futureResponse = activeActor !! activeMessage
val activeReply = future()

AKKA actor

首先需要添加akka的maven依賴：

<!--akka actor-->
<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-actor_2.10</artifactId>
    <version>2.3.16</version>
</dependency>
<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-remote_2.10</artifactId>
    <version>2.3.16</version>
</dependency>
<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-slf4j_2.10</artifactId>
    <version>2.3.16</version>
</dependency>

AKKA消息傳遞——本地

原理如下：

技術分享圖片

_01StudentActorOps

package cn.xpleaf.bigdata.p5.myakka.p1

import akka.actor.{Actor, ActorSystem, Props}
import cn.xpleaf.bigdata.p5.myakka.MessageProtocol.{QuoteRequest, QuoteResponse}

import scala.util.Random

/**
  * 基於AKKA Actor的單向通信案例
  * 學生向老師發送請求
  */
object _01StudentActorOps {
    def main(args: Array[String]): Unit = {
        // 第一步：構建Actor操作系統
        val actorSystem = ActorSystem("StudentActorSystem")
        // 第二步：actorSystem創建TeacherActor的代理對象ActorRef
        val teacherActorRef = actorSystem.actorOf(Props[TeacherActor])
        // 第三步：發送消息
        teacherActorRef ! QuoteRequest()

        Thread.sleep(2000)
        // 第四步：關閉
        actorSystem.shutdown()
    }
}

class TeacherActor extends Actor {
    val quotes = List(
        "Moderation is for cowards",
        "Anything worth doing is worth overdoing",
        "The trouble is you think you have time",
        "You never gonna know if you never even try")

    override def receive = {
        case QuoteRequest() => {
            val random = new Random()

            val randomIndex = random.nextInt(quotes.size)
            val randomQuote = quotes(randomIndex)

            val response = QuoteResponse(randomQuote)
            println(response)
        }
    }
}

MessageProtocol

後面akka通信的幾個測試程序都會使用到這個object，只在這裏給出，後面不再給出。

package cn.xpleaf.bigdata.p5.myakka

/**
  * akka actor通信協議
  */
object MessageProtocol {

    case class QuoteRequest()

    case class QuoteResponse(resp: String)

    case class InitSign()

}

object Start extends Serializable

object Stop extends Serializable

trait Message {
    val id: String
}

case class Shutdown(waitSecs: Int) extends Serializable

case class Heartbeat(id: String, magic: Int) extends Message with Serializable

case class Header(id: String, len: Int, encrypted: Boolean) extends Message with Serializable

case class Packet(id: String, seq: Long, content: String) extends Message with Serializable

測試

輸出結果如下：

QuoteResponse(Anything worth doing is worth overdoing)

AKKA請求與響應——本地

原理如下：

技術分享圖片

TeacherActor

package cn.xpleaf.bigdata.p5.myakka.p2

import akka.actor.Actor
import cn.xpleaf.bigdata.p5.myakka.MessageProtocol.{QuoteRequest, QuoteResponse}

import scala.util.Random

/**
  * Teacher Actor
  */
class TeacherActor extends Actor {
    val quotes = List(
        "Moderation is for cowards",
        "Anything worth doing is worth overdoing",
        "The trouble is you think you have time",
        "You never gonna know if you never even try")

    override def receive = {
        case QuoteRequest() => {
            val random = new Random()

            val randomIndex = random.nextInt(quotes.size)
            val randomQuote = quotes(randomIndex)

            val response = QuoteResponse(randomQuote)
//            println(response)
            sender ! response
        }
    }
}

StudentActor

package cn.xpleaf.bigdata.p5.myakka.p2

import akka.actor.{Actor, ActorLogging, ActorRef}
import cn.xpleaf.bigdata.p5.myakka.MessageProtocol.{InitSign, QuoteRequest, QuoteResponse}

/**
  * Student Actor
  * 當學生接收到InitSign信號之後，便向老師發送一條Request請求的消息
  */
class StudentActor(teacherActorRef:ActorRef) extends Actor with ActorLogging {
    override def receive = {
        case InitSign => {
            teacherActorRef ! QuoteRequest()
//            println("student send request")
        }
        case QuoteResponse(resp) => {
            log.info(s"$resp")
        }
    }
}

DriverApp

package cn.xpleaf.bigdata.p5.myakka.p2

import akka.actor.{ActorSystem, Props}
import cn.xpleaf.bigdata.p5.myakka.MessageProtocol.InitSign

object DriverApp {
    def main(args: Array[String]): Unit = {
        val actorSystem = ActorSystem("teacherStudentSystem")
        // 老師的代理對象
        val teacherActorRef = actorSystem.actorOf(Props[TeacherActor], "teacherActor")
        // 學生的代理對象
        val studentActorRef = actorSystem.actorOf(Props[StudentActor](new StudentActor(teacherActorRef)), "studentActor")

        studentActorRef ! InitSign

        Thread.sleep(2000)

        actorSystem.shutdown()
    }
}

測試

輸出結果如下：

[INFO] [04/24/2018 10:02:19.932] [teacherStudentSystem-akka.actor.default-dispatcher-2] [akka://teacherStudentSystem/user/studentActor] Anything worth doing is worth overdoing

AKKA請求與響應——遠程

application.conf

MyRemoteServerSideActor {
  akka {
    actor {
      provider = "akka.remote.RemoteActorRefProvider"
    }
    remote {
      enabled-transports = ["akka.remote.netty.tcp"]
      netty.tcp {
        hostname = "127.0.0.1"
        port = 2552
      }
    }
  }
}
MyRemoteClientSideActor {
  akka {
    actor {
      provider = "akka.remote.RemoteActorRefProvider"
    }
  }
}

RemoteActor

package cn.xpleaf.bigdata.p5.myakka.p3

import akka.actor.{Actor, ActorLogging}
import cn.xpleaf.bigdata.p5.myakka.{Header, Shutdown, Start, Stop}

class RemoteActor extends Actor with ActorLogging {
    def receive = {
        case Start => { // 處理Start消息
            log.info("Remote Server Start ==>RECV Start event : " + Start)
        }
        case Stop => { // 處理Stop消息
            log.info("Remote Server Stop ==>RECV Stop event: " + Stop)
        }
        case Shutdown(waitSecs) => { // 處理Shutdown消息
            log.info("Remote Server Shutdown ==>Wait to shutdown: waitSecs=" + waitSecs)
            Thread.sleep(waitSecs)
            log.info("Remote Server Shutdown ==>Shutdown this system.")
            context.system.shutdown // 停止當前ActorSystem系統
        }
        case Header(id, len, encrypted) => log.info("Remote Server => RECV header: " + (id, len, encrypted)) // 處理Header消息
        case _ =>
    }
}

AkkaServerApplication

package cn.xpleaf.bigdata.p5.myakka.p3

import akka.actor.{ActorSystem, Props}
import com.typesafe.config.ConfigFactory

object AkkaServerApplication extends App {
    // 創建名稱為remote-system的ActorSystem：從配置文件application.conf中獲取該Actor的配置內容
    val system = ActorSystem("remote-system",
        ConfigFactory.load().getConfig("MyRemoteServerSideActor"))

    val log = system.log
    log.info("===>Remote server actor started: " + system)
    // 創建一個名稱為remoteActor的Actor，返回一個ActorRef，這裏我們不需要使用這個返回值
    system.actorOf(Props[RemoteActor], "remoteActor")

}

ClientActor

package cn.xpleaf.bigdata.p5.myakka.p3

import akka.actor.SupervisorStrategy.Stop
import akka.actor.{Actor, ActorLogging}
import cn.xpleaf.bigdata.p5.myakka.{Header, Start}

class ClientActor extends Actor with ActorLogging {

    // akka.<protocol>://<actor system>@<hostname>:<port>/<actor path>
    val path = "akka.tcp://[email protected]:2552/user/remoteActor" // 遠程Actor的路徑，通過該路徑能夠獲取到遠程Actor的一個引用
    val remoteServerRef = context.actorSelection(path) // 獲取到遠程Actor的一個引用，通過該引用可以向遠程Actor發送消息
    @volatile var connected = false
    @volatile var stop = false

    def receive = {
        case Start => { // 發送Start消息表示要與遠程Actor進行後續業務邏輯處理的通信，可以指示遠程Actor初始化一些滿足業務處理的操作或數據
            send(Start)
            if (!connected) {
                connected = true
                log.info("ClientActor==> Actor connected: " + this)
            }
        }
        case Stop => {
            send(Stop)
            stop = true
            connected = false
            log.info("ClientActor=> Stopped")
        }
        case header: Header => {
            log.info("ClientActor=> Header")
            send(header)
        }
        case (seq, result) => log.info("RESULT: seq=" + seq + ", result=" + result) // 用於接收遠程Actor處理一個Packet消息的結果
        case m => log.info("Unknown message: " + m)
    }

    private def send(cmd: Serializable): Unit = {
        log.info("Send command to server: " + cmd)
        try {
            remoteServerRef ! cmd // 發送一個消息到遠程Actor，消息必須是可序列化的，因為消息對象要經過網絡傳輸
        } catch {
            case e: Exception => {
                connected = false
                log.info("Try to connect by sending Start command...")
                send(Start)
            }
        }
    }

}

AkkaClientApplication

package cn.xpleaf.bigdata.p5.myakka.p3

import akka.actor.{ActorSystem, Props}
import cn.xpleaf.bigdata.p5.myakka.{Header, Start}
import com.typesafe.config.ConfigFactory

object AkkaClientApplication extends App {
    // 通過配置文件application.conf配置創建ActorSystem系統
    val system = ActorSystem("client-system",
        ConfigFactory.load().getConfig("MyRemoteClientSideActor"))
    val log = system.log
    val clientActor = system.actorOf(Props[ClientActor], "clientActor") // 獲取到ClientActor的一個引用
    clientActor ! Start // 發送一個Start消息，第一次與遠程Actor握手（通過本地ClientActor進行轉發）
    Thread.sleep(2000)
    clientActor ! Header("What‘s your name: Can you tell me ", 20, encrypted = false) // 發送一個Header消息到遠程Actor（通過本地ClientActor進行轉發）
    Thread.sleep(2000)

}

測試

服務端輸出結果如下：

[INFO] [04/24/2018 09:39:49.271] [main] [Remoting] Starting remoting
[INFO] [04/24/2018 09:39:49.508] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://[email protected]:2552]
[INFO] [04/24/2018 09:39:49.509] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://[email protected]:2552]
[INFO] [04/24/2018 09:39:49.517] [main] [ActorSystem(remote-system)] ===>Remote server actor started: akka://remote-system
[INFO] [04/24/2018 09:46:01.872] [remote-system-akka.actor.default-dispatcher-3] [akka.tcp://[email protected]:2552/user/remoteActor] Remote Server Start ==>RECV Start event : cn.xpleaf.bigdata.p5.myakka.Start$@325737b3
[INFO] [04/24/2018 09:46:03.501] [remote-system-akka.actor.default-dispatcher-3] [akka.tcp://[email protected]:2552/user/remoteActor] Remote Server => RECV header: (What‘s your name: Can you tell me ,20,false)

客戶端輸出結果如下：

[INFO] [04/24/2018 09:46:01.274] [main] [Remoting] Starting remoting
[INFO] [04/24/2018 09:46:01.479] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://[email protected]:2552]
[INFO] [04/24/2018 09:46:01.480] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://[email protected]:2552]
[INFO] [04/24/2018 09:46:01.493] [client-system-akka.actor.default-dispatcher-4] [akka.tcp://[email protected]:2552/user/clientActor] Send command to server: cn.xpleaf.bigdata.p5.myakka.Start$@4f00805d
[INFO] [04/24/2018 09:46:01.496] [client-system-akka.actor.default-dispatcher-4] [akka.tcp://[email protected]:2552/user/clientActor] ClientActor==> Actor connected: cn.xpleaf.bigdata.p5.myakka.p3.ClientActor@5a85b576
[INFO] [04/24/2018 09:46:03.490] [client-system-akka.actor.default-dispatcher-2] [akka.tcp://[email protected]:2552/user/clientActor] ClientActor=> Header
[INFO] [04/24/2018 09:46:03.491] [client-system-akka.actor.default-dispatcher-2] [akka.tcp://[email protected]:2552/user/clientActor] Send command to server: Header(What‘s your name: Can you tell me ,20,false)

Scala筆記整理（九）：Actor和AKKA

相關推薦

Scala筆記整理（九）：Actor和AKKA

大數據 Scala [TOC] 概述 ? Scala的Actor有點類似於Java中的多線程編程。但是不同的是，Scala的Actor提供的模型與多線程有所不同。Scala的Actor盡可能地避免鎖和共享狀態，從而避免多線程並發時出現資源爭用的情況，進而提升多線程編程的性能。 Spark中使用

Cris 的 Scala 筆記整理（九）：面向物件高階

9. 面向物件高階 9.1 靜態屬性和靜態方法 ① 回顧 Java 的靜態概念 public static 返回值型別 方法名(引數列表) {方法體} Java 中靜態方法並不是通過物件呼叫的，而是通過類物件呼叫的，所以靜態操作並不是面向物件的 ② Scala 中靜態的概念-伴生物件 Scala

Scala筆記整理（一）：scala基本知識

大數據 Scala [TOC] Scala簡介 Scala是一門多範式（multi-paradigm）的編程語言，設計初衷是要集成面向對象編程和函數式編程的各種特性。 Scala運行在Java虛擬機上，並兼容現有的Java程序。 Scala源代碼被編譯成Java字節碼，所以它可以運行於JVM之上，並

Scala筆記整理（二）：Scala數據結構—數組、map與tuple

大數據 Scala [TOC] 數組 定長數組 如果你需要一個長度不變的數組，可以用Scala中的Array。例如： val numsArray = new Array[Int] (30) //長度為30的整數數組，所有元素初始化為0 val stringArrays = new Array [St

Scala筆記整理（三）：Scala面向對象—類詳解1

大數據 Scala [TOC] getter和setter 簡單類和無參方法 需要說明的是，官方給出的參考資料對於這方面講解得十分詳細，目前入門來說，不需要達到這樣的理解，只需要可以進行開發即可，這一部分我給出的是官方的一些文檔說明，但是後面的定制和註解，就直接簡單說明能用就好了。 class

Scala筆記整理（四）：Scala面向對象—類詳解2（繼承相關）

大數據 Scala [TOC] 單例 希望某個類只存在一個使用的對象，而不管有多少個調用者在使用它，就是單例的概念。 Java中的單例 package cn.xpleaf.single; /** * 單例模式-餓漢式（在屬性中先創建好對象，不管是否調用getInstance方法） * @auth

Scala筆記整理（五）：函數式編程

大數據 Scala [TOC] 作為值傳遞的函數 測試代碼如下： package cn.xpleaf.bigdata.p4.function /** * scala中關於函數的操作 */ object _01FunctionOps { def main(args: Array[Stri

Scala筆記整理（六）：Scala集合庫

大數據 Scala [TOC] Scala主要集合結構 1、Scala中的集合體系主要包括：Iterable、Seq（IndexSeq）、Set（SortedSet）、Map（SortedMap）。其中Iterable是所有集合trait的根trait。實際上Seq、Set、和Map都是子trait

Scala筆記整理（八）：類型參數（泛型）與隱士轉換

大數據 Scala [TOC] 概述 類型參數是什麽？類型參數其實就是Java中的泛型。大家對Java中的泛型應該有所了解，比如我們有List list = new ArrayList()，接著list.add(1)，沒問題，list.add("2")，然後我們list.get(

Scala筆記整理（七）：模式匹配和樣例類

大數據 Scala [TOC] 可以用到switch語句 1、Scala強大的模式匹配機制，可以應用在switch語句、類型檢查以及“析構”等場合。 def swithOps: Unit ={ var sign = 0 val ch: Char = ‘+‘ ch match

Cris 的 Scala 筆記整理（七）：面向物件

7. 面向物件（重點） 7.1 Scala 面向物件基礎 類 [修飾符] class 類名 { 類體 } scala語法中，類並不宣告為public，所有這些類都具有公有可見性(即預設就是public) 一個Scala原始檔可以包含多個類 定義一個最簡單的類 object

Cris 的 Scala 筆記整理（十）：隱式轉換

10. 隱式轉換 隱式轉換函式是以implicit關鍵字宣告的帶有單個引數的函式。這種函式將會自動應用，將值從一種型別轉換為另一種型別 ① 隱式轉換最佳入門案例 首先看下面的程式碼 如果想要轉換成功，就需要藉助隱式轉換函式 為了更好的理解，我們看看反編譯後的位元組

Kafka筆記整理（二）：Kafka Java API使用

大數據 Kafka Java [TOC] Kafka筆記整理（二）：Kafka Java API使用 下面的測試代碼使用的都是下面的topic： $ kafka-topics.sh --describe hadoop --zookeeper uplooking01:2181,uplooking0

Kafka筆記整理（三）：消費形式驗證與性能測試

大數據 Kafka 性能測試 [TOC] Kafka筆記整理（三）：消費形式驗證與性能測試 Kafka消費形式驗證 前面的《Kafka筆記整理（一）》中有提到消費者的消費形式，說明如下： 1、每個consumer屬於一個consumer group，可以指定組id。group.id 2、消費形

Redis筆記整理（一）：Redis安裝配置與數據類型操作

數據庫 NoSQL Redis [TOC] Redis筆記整理（一）：Redis安裝配置與數據類型操作 Redis簡介 Redis是一個開源（BSD許可），內存存儲的數據結構服務器，可用作數據庫，高速緩存和消息隊列代理。 它支持字符串、哈希表、列表、集合、有序集合，位圖，hyperloglo

Redis筆記整理（二）：Java API使用與Redis分布式集群環境搭建

數據庫 NoSQL Redis [TOC] Redis筆記整理（二）：Java API使用與Redis分布式集群環境搭建 Redis Java API使用（一）：單機版本Redis API使用 Redis的Java API通過Jedis來進行操作，因此首先需要Jedis的第三方庫，因為使用的是M

Redis筆記整理（三）：進階操作與高級部分

數據庫 NoSQL Redis [TOC] Redis筆記整理（三）：進階操作與高級部分 Redis發布訂閱 Redis發布訂閱（pub/sub）是一種消息通信模式：發送者(pub)發送消息，訂閱者(sub)接收消息。 Redis客戶端可以訂閱任意數量的頻道。 下圖展示了頻道channel1，以

ElasticSearch筆記整理（一）：簡介、REST與安裝配置

大數據 ElasticSearch ELK [TOC] ElasticSearch簡介 ElasticSearch是一款基於Apache Lucene構建的開源搜索引擎，它采用Java編寫並使用Lucene構建索引、提供搜索功能，ElasticSearch的目標是讓全文搜索變得簡單，開發者可以通

ElasticSearch筆記整理（二）：CURL操作、ES插件、集群安裝與核心概念

大數據 ElasticSearch ELK [TOC] CURL操作 CURL簡介 curl是利用URL語法在命令行方式下工作的開源文件傳輸工具，使用curl可以簡單實現常見的get/post請求。簡單的認為是可以在命令行下面訪問url的一個工具。在centos的默認庫裏面是有curl工具的，如

Storm筆記整理（一）：簡介與設計思想

大數據 實時計算 Storm [TOC] 實時計算概述 有別於傳統的離線批處理操作(對很多數據的集合進行的操作)，實時處理，說白就是針對一條一條的數據/記錄進行操作，所有的這些操作進行一個匯總(截止到目前為止的所有的統計總和)。 實時計算與離線計算比較 Bounded：有界 離線計算面臨