无论哪一种模型,其任务调度方法都既可以是协作式的(cooperative),也可以是抢占式的(preemptive),协作式任务调度的时机由任务自身确定,当且仅当当前任务 主动放弃执行权,调度器才会将执行权交给其他任务;而抢占式任务调度的时机由底层软件(通常是操作系统) 确定,一个任务可能在任何时候被唤醒或挂起,因此,不同任务也可能并行执行。

协作式事件驱动模型和抢占式多线程模型是最常用的两种并发任务 处理技术。它们是对偶模型。

并行工作者模型

• 在并行工作者模型中，委派者（Delegator）将传入的作业分配给不同的工作者。每个工作者完成整个任务。工作者们并行运作在不同的线程上，甚至可能在不同的CPU上。在Java应用系统中，并行工作者模型是最常见的并发模型（即使正在转变）。java.util.concurrent包中的许多并发实用工具都是设计用于这个模型的。

• 并行工作者模式的优点是，它很容易理解。你只需添加更多的工作者来提高系统的并行度。

• 共享的工作者经常需要访问一些共享数据，一旦共享状态潜入到并行工作者模型中，将会使情况变得复杂起来。

  • 线程需要以某种方式存取共享数据，以确保某个线程的修改能够对其他线程可见（数据修改需要同步到主存中，不仅仅将数据保存在执行这个线程的CPU的缓存中）。线程需要避免竟态，死锁以及很多其他共享状态的并发性问题。

    • 可持久化的数据结构是另一种选择。在修改的时候，可持久化的数据结构总是保护它的前一个版本不受影响。因此，如果多个线程指向同一个可持久化的数据结构，并且其中一个线程进行了修改，进行修改的线程会获得一个指向新结构的引用。所有其他线程保持对旧结构的引用，旧结构没有被修改并且因此保证一致性。Scala编程包含几个持久化数据结构。【注：这里的可持久化数据结构不是指持久化存储，而是一种数据结构，比如Java中的String类，以及CopyOnWriteArrayList类，具体可参考】

    • 并行工作者模式的另一个缺点是，作业执行顺序是不确定的。无法保证哪个作业最先或者最后被执行。作业A可能在作业B之前就被分配工作者了，但是作业B反而有可能在作业A之前执行。并行工作者模式的这种非确定性的特性，使得很难在任何特定的时间点推断系统的状态。这也使得它也更难（如果不是不可能的话）保证一个作业在其他作业之前被执行。

• 流水线并发模型/事件驱动系统

  • 类似于工厂中生产线上的工人们那样组织工作者。每个工作者只负责作业中的部分工作。当完成了自己的这部分工作时工作者会将作业转发给下一个工作者。每个工作者在自己的线程中运行，并且不会和其他工作者共享状态。有时也被成为无共享并行模型。

  • 工作者之间无需共享状态，意味着实现的时候无需考虑所有因并发访问共享对象而产生的并发性问题。这使得在实现工作者的时候变得非常容易。在实现工作者的时候就好像是单个线程在处理工作-基本上是一个单线程的实现。

  • 当工作者知道了没有其他线程可以修改它们的数据，工作者可以变成有状态的。对于有状态，我是指，它们可以在内存中保存它们需要操作的数据，只需在最后将更改写回到外部存储系统。因此，有状态的工作者通常比无状态的工作者具有更高的性能。

  • 流水线并发模型最大的缺点是作业的执行往往分布到多个工作者上，并因此分布到项目中的多个类上。这样导致在追踪某个作业到底被什么代码执行时变得困难。

    同样，这也加大了代码编写的难度。有时会将工作者的代码写成回调处理的形式。若在代码中嵌入过多的回调处理，往往会出现所谓的回调地狱（callback hell）现象。所谓回调地狱，就是意味着在追踪代码在回调过程中到底做了什么，以及确保每个回调只访问它需要的数据的时候，变得非常困难。

  • 事件驱动模型的主要问题可以概括为这样几类:控制流反转、函数分裂、变量重定位、调用栈重构和侵入性原则。

    • 与控制流翻转相伴的往往是函数分裂

    • 函数分裂会迫使原本在栈中的函数局部变量转移到堆中,造成变量重定位

    • 在调试事件驱动程序的时候,开发人员往往需要手工重新构造函数调用堆栈,造成调用栈重构问题

    • 函数分裂具有侵入性原则,因为一个分裂函数会强制分裂它的直接和间接调用者

    • 事件驱动模型原本应属于实现细节,然而其侵入性却导致这个细节必须暴露在模块接口之中,造成了模块 的低封装性

  • Actors 和 channels 是两种比较类似的流水线（或反应器/事件驱动）模型。

    • actors

      • 在Actor模型中每个工作者被称为actor。Actor之间可以直接异步地发送和处理消息。Actor可以被用来实现一个或多个像前文描述的那样的作业处理流水线。

      • 如akka

        object Greeter {
          final case class Greet(whom: String, replyTo: ActorRef[Greeted])
          final case class Greeted(whom: String, from: ActorRef[Greet])
        
          def apply(): Behavior[Greet] = Behaviors.receive { (context, message) =>
            context.log.info("Hello {}!", message.whom)
            message.replyTo ! Greeted(message.whom, context.self)
            Behaviors.same
          }
        }
        
        object GreeterBot {
        
          def apply(max: Int): Behavior[Greeter.Greeted] = {
            bot(0, max)
          }
        
          private def bot(greetingCounter: Int, max: Int): Behavior[Greeter.Greeted] =
            Behaviors.receive { (context, message) =>
              val n = greetingCounter + 1
              context.log.info("Greeting {} for {}", n, message.whom)
              if (n == max) {
                Behaviors.stopped
              } else {
                message.from ! Greeter.Greet(message.whom, context.self)
                bot(n, max)
              }
            }
        }
        
        
        
        object GreeterMain {
        
          final case class SayHello(name: String)
        
          def apply(): Behavior[SayHello] =
            Behaviors.setup { context =>
              val greeter = context.spawn(Greeter(), "greeter")
        
              Behaviors.receiveMessage { message =>
                val replyTo = context.spawn(GreeterBot(max = 3), message.name)
                greeter ! Greeter.Greet(message.name, replyTo)
                Behaviors.same
              }
            }
        }
        

    • Channel

      • 而在Channel模型中，工作者之间不直接进行通信。相反，它们在不同的通道中发布自己的消息（事件）。其他工作者们可以在这些通道上监听消息，发送者无需知道谁在监听。

      • 比如golang

        package main
        
        import "fmt"
        
        func sum(s []int, c chan int) {
            sum := 0
            for _, v := range s {
                sum += v
            }
            c <- sum // send sum to c
        }
        
        func main() {
            s := []int{7, 2, 8, -9, 4, 0}
        
            c := make(chan int)
            go sum(s[:len(s)/2], c)
            go sum(s[len(s)/2:], c)
            x, y := <-c, <-c // receive from c
        
            fmt.Println(x, y, x+y)
        }