Scala语法(五) Actor与线程池

浅浅的花香味﹌ 2022-02-09 08:21 525阅读 0赞

### 前言 ###

在前面的章节内, 我们介绍了Scala基础、集合、方法与函数体, 以及模式匹配的相关知识. 本章简单的介绍了Actor的相关知识.

--------------------

### 正文 ###

*  创建Actor  
    个人理解`Actor`对象与Java内的`Thread`对象相似. 创建时通过继承的方式创建`Actor`类, 并通过`start()`方法启动`Actor`子线程.

import scala.actors.Actor
    
    
    /**
     * 使用Actor的方式启动线程.
     * 1. 调用start()方法启动Actor
     * 2. 调用start()方法后, act方法会被执行
     * 3. 向Actor发送消息.
     * 
     * | 发送异步消息 没有返回值
     * |? 发送同步消息,等待返回值
     * |! 发送异步消息,返回值为Future[Any]
     * */
    // Actor已经2.11.x版本已经被移除了 使用的是Akka内的Actor.
    object QuickActorA extends  Actor{
      def act(){
        for( i <- 1 to 100){
          println("actor1-"+i)
          Thread.sleep(1000)
        }
      }
    }
    object QuickActorB extends  Actor{
      def act(){
        for( i <- 1 to 100){
          println("actor2-"+i)
          Thread.sleep(1000)
        }
      }
    }
    object QucikActor {
        def main(args: Array[String]): Unit = {
          QuickActorA.start()
          QuickActorB.start()
        }
    }

*  持久通信

import scala.actors.Actor
    
    /**
     * Actor不断的接收消息.
     * 
     * */
    class QuickActorReceiveA extends Actor {
      def act():Unit = {
        while(true){
          receive{
            case "start" => {
              println("Start")
            }
            case "stop" => {
              println("Stop")
            }
          }
        }
      }
    }
    object QuickActorReceive {
      def main(args: Array[String]): Unit = {
        val actor = new QuickActorReceiveA()
        actor.start()
        actor ! "start"
        actor ! "stop"
      }
    }

*  React方法通信

import scala.actors.Actor
    
    class QuickActorReactA extends Actor {
      def act():Unit ={
        loop{
          // 复用线程池
          react{
            case "start" => {println("start"); Thread.sleep(1000); }
            case "stop" => {println("stop") ;Thread.sleep(1000);}
            case "exit" => {exit()}
          }
        }
      }
    }
    object QuickActorReact {
      def main(args: Array[String]): Unit = {
          val quickActorReactA = new QuickActorReactA()
          quickActorReactA.start()
          quickActorReactA ! "start"
          quickActorReactA ! "stop"
          quickActorReactA ! "exit"
    
      }
    }

*  创建线程池(Java)

import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    
    
    /***
     * Java内进行线程池声明代码.
     */
    public class QuickJavaThreadPool {
    	public static void main(String[] args) {
    		// 定义线程池 并向其中塞任务
    		ExecutorService pool = Executors.newCachedThreadPool();
    		for(int i=0;i<10;i++) {
    			pool.execute(new Runnable() {
    				
    				@Override
    				public void run() {
    					System.out.println("Thread Name:"+Thread.currentThread().getName());
    					try {
    						Thread.sleep(1000);
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}
    				}
    			});
    		}
    		
    		Future<Integer> future = pool.submit(new Callable<Integer>() {
    
    			@Override
    			public Integer call() throws Exception {
    				return 1;
    			}
    			
    		});
    		
    //		pool.shutdown();
    	}
    }

*  创建线程池(Scala)

import java.util.concurrent.Executors
    import java.util.concurrent.Future
    import java.util.concurrent.Callable
    
    /**
     * 在Scala内使用线程池技术 创建 线程池.
     *
     */
    object QuickThreadPool {
      def main(args: Array[String]): Unit = {
        val pool = Executors.newCachedThreadPool()
        for (i <- 1 to 10) {
          pool.execute(new Runnable() {
            def run(): Unit = {
              println("Thread-Name:" + Thread.currentThread().getName())
              Thread.sleep(1000)
            }
          })
        }
    
        
        val future:Future[Int] = pool.submit(new Callable[Int](){
          override def call():Int ={
            Thread.sleep(1000)
            100
          }
        }) 
        var status = future.isDone()
        Thread.sleep(1000)
        status = future.isDone()
        if(status){
          println(future.get())
        }
    //    pool.shutdown()
      }
    }