Big data, a good programmer, shares Actor learning notes.
Big data, a good programmer, shares Actor learning notes. She can achieve very powerful functions in scala. He is an event model based on concurrency mechanism.
The version of scala2.10.x we are learning now is the previous Actor.
Synchronization: for tasks queued on the main program, the next task can be executed only after the previous task has been executed.
Async: refers to a task that enters the "task alignment" without entering the main program. Only when the task of the main program is finished, the "task alignment" begins to request the main program and request the task to be executed, and the task will enter the main program.
Java
Shared variables-locking
There will be a lock problem.
Scala
Actor does not share data
There is no concept of lock
Message (communication) is required between Actor communications
Aactor execution order
1. First call the start () method to start Actor
two。 The act () method is executed after the start () method is called
Send messages between 3.Actor
Three ways for Actor to send messages
!-> send an asynchronous message with no return value
!?-> send a synchronization message with a return value and a thread waiting
!!-> send an asynchronous message with a return value of type FutureAny (used to get the result of asynchronous operation)
Actor parallel execution
/ / Note that the two actor are executed in parallel. When one of the for loops ends, the actor ends.
Object ActorDemo01 {
Def main (args: Array [String]): Unit = {
MyActor1.start ()
MyActor2.start ()
}
}
Object MyActor1 extends Actor {
Override def act (): Unit = {
For (I async
Actor! "stop"
Println ("send complete")
}
}
Class MyActor extends Actor {
Override def act (): Unit = {
While (true) {/ / endless loop
Receive {/ / receive
Case "start" = > {
Println ("starting")
Thread.sleep (1000)
Println ("started")
}
Case "stop" = > {
Println ("stopping")
Thread.sleep (1000)
Println ("stopped")
}
}
}
}
}
The second way: use react instead of receive, that is to say, react threads are reusable and more efficient than receive
Object ActorDemo03 {
Def main (args: Array [String]): Unit = {
Val actor: MyActor3 = new MyActor3
Actor.start ()
Actor! "start"
Actor! "stop"
Println ("succeeded")
}
}
Class MyActor3 extends Actor {
Override def act (): Unit = {
Loop {
React {
Case "start" = > {
Println ("starting")
Thread.sleep (1000)
Println ("sarted")
}
Case "stop" = > {
Println ("stoppting")
Thread.sleep (1000)
Println ("stopped")
}
}
}
}
}
Practice Actor sending messages with sample classes
/ / create a sample class
Case class AsyncMsg (id: Int, msg: String)
Case class SyncMsg (id: Int, msg: String)
Case class ReplyMsg (id: Int, msg: String)
Object ActorDemo01 extends Actor {
Override def act (): Unit = {
While (true) {
Receive {
Case "start" = > println ("starting...")
Case AsyncMsg (id, msg) = >
{
Println (s "id:$id,msg:$msg")
Sender! ReplyMsg (1, "sucess") / / returns a response message after receiving the message
}
Case SyncMsg (id,msg) = > {
Println (s "id:$id,msg:$msg")
Sender! ReplyMsg (2, "sucess")
}
}
}
}
}
Object ActorTest {
Def main (args: Array [String]): Unit = {
Val actor: Actor = ActorDemo01.start ()
/ / send messages asynchronously and no return value is returned
/ / actor! AsyncMsg (3, "heihei")
/ / println ("Asynchronous message sent completed, no return value")
/ / send messages synchronously, with a return value
/ / val text: Any = actor! SyncMsg (4, "OK")
/ / println (text)
/ / println ("synchronization message sent successfully")
/ / send messages asynchronously with a return value of Future [Any]
Val reply: Future [Any] = actor! SyncMsg (5, "OK is does not exist")
Thread.sleep (2000)
If (reply.isSet) {
Val applyMsg: Any = reply.apply ()
Println (applyMsg)
} else {
Println ("Nothing")
}
}
}
Wordcount of Actor parallelization
Class Task extends Actor {
Override def act (): Unit = {
Loop {
React {
Case SubmitTask (fileName) = > {
Val contents = Source.fromFile (new File (fileName)) .mkString
Val arr = contents.split ("\ r\ n")
Val result = arr.flatMap (_ .split (")). Map ((_, 1)). GroupBy (_. _ 1). MapValues (_ .length)
/ / val result = arr.flatMap (_ .split (")). Map ((_, 1)). GroupBy (_. _ 1). MapValues (_ .foldLeft (0) (_ + _ .2))
Sender! ResultTask (result)
}
Case StopTask = > {
Exit ()
}
}
}
}
}
Object WorkCount {
Def main (args: Array [String]) {
Val files = Array ("c://words.txt", "c://words.log")
Val replaySet = new mutable. HashSet [FutureAny]
Val resultList = new mutable.ListBuffer [ResultTask]
For (f 0) {
Val toCumpute = replaySet.filter (_ .isSet)
For (r x.foldLeft (0) (_ + _. _ 2))
Println (finalResult)
}
}
Case class SubmitTask (fileName: String)
Case object StopTask
Case class ResultTask (result: Map [String, Int])
