Shulou(Shulou.com)06/02 Report--

This article introduces the relevant knowledge of "how to realize the wordcount of Actor parallelization". In the operation of actual cases, many people will encounter such a dilemma, so let the editor lead you to learn how to deal with these situations. I hope you can read it carefully and be able to achieve something!

She can achieve very powerful functions in scala. It 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])

This is the end of the content of "how to realize the wordcount of Actor parallelization". Thank you for your reading. If you want to know more about the industry, you can follow the website, the editor will output more high-quality practical articles for you!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.