Golang超高性能分布式Actor模型开发插件库ProtoActor的使用

ProtoActor是一个跨平台的Actor模型实现，支持Go和C#之间的交互。以下是关于ProtoActor的详细介绍和使用示例。

设计原则

• 最小化API：API小巧易用，避免企业级JVM式的容器和配置
• 基于现有技术：利用gRPC流进行网络传输，Protobuf进行序列化
• 传递数据而非对象：显式处理序列化问题
• 高性能：不因API魔法而牺牲性能

ProtoActor的远程通信性能极佳，在两个节点之间每秒可传递超过200万条消息，同时保持消息顺序。

基础使用示例

Hello World示例

type Hello struct{ Who string }
type HelloActor struct{}

func (state *HelloActor) Receive(context actor.Context) {
    switch msg := context.Message().(type) {
    case Hello:
        fmt.Printf("Hello %v\n", msg.Who)
    }
}

func main() {
    context := actor.EmptyRootContext
    props := actor.PropsFromProducer(func() actor.Actor { return &HelloActor{} })
    pid, err := context.Spawn(props)
    if err != nil {
        panic(err)
    }
    context.Send(pid, Hello{Who: "Roger"})
    console.ReadLine()
}

状态机示例

type Hello struct{ Who string }
type SetBehaviorActor struct{}

func (state *SetBehaviorActor) Receive(context actor.Context) {
    switch msg := context.Message().(type) {
    case Hello:
        fmt.Printf("Hello %v\n", msg.Who)
        context.SetBehavior(state.Other)
    }
}

func (state *SetBehaviorActor) Other(context actor.Context) {
    switch msg := context.Message().(type) {
    case Hello:
        fmt.Printf("%v, ey we are now handling messages in another behavior", msg.Who)
    }
}

func NewSetBehaviorActor() actor.Actor {
    return &SetBehaviorActor{}
}

func main() {
    context := actor.EmptyRootContext
    props := actor.PropsFromProducer(NewSetBehaviorActor)
    pid, err := context.Spawn(props)
    if err != nil {
        panic(err)
    }
    context.Send(pid, Hello{Who: "Roger"})
    context.Send(pid, Hello{Who: "Roger"})
    console.ReadLine()
}

生命周期事件

type Hello struct{ Who string }
type HelloActor struct{}

func (state *HelloActor) Receive(context actor.Context) {
    switch msg := context.Message().(type) {
    case *actor.Started:
        fmt.Println("Started, initialize actor here")
    case *actor.Stopping:
        fmt.Println("Stopping, actor is about shut down")
    case *actor.Stopped:
        fmt.Println("Stopped, actor and its children are stopped")
    case *actor.Restarting:
        fmt.Println("Restarting, actor is about restart")
    case Hello:
        fmt.Printf("Hello %v\n", msg.Who)
    }
}

func main() {
    context := actor.EmptyRootContext
    props := actor.PropsFromProducer(func() actor.Actor { return &HelloActor{} })
    pid, err := context.Spawn(props)
    if err != nil {
        panic(err)
    }
    context.Send(pid, Hello{Who: "Roger"})

    // 等待1秒以展示正确的生命周期事件顺序
    time.Sleep(1 * time.Second)
    context.Stop(pid)

    console.ReadLine()
}

监督策略示例

type Hello struct{ Who string }
type ParentActor struct{}

func (state *ParentActor) Receive(context actor.Context) {
    switch msg := context.Message().(type) {
    case Hello:
        props := actor.PropsFromProducer(NewChildActor)
        child := context.Spawn(props)
        context.Send(child, msg)
    }
}

func NewParentActor() actor.Actor {
    return &ParentActor{}
}

type ChildActor struct{}

func (state *ChildActor) Receive(context actor.Context) {
    switch msg := context.Message().(type) {
    case *actor.Started:
        fmt.Println("Starting, initialize actor here")
    case *actor.Stopping:
        fmt.Println("Stopping, actor is about shut down")
    case *actor.Stopped:
        fmt.Println("Stopped, actor and its children are stopped")
    case *actor.Restarting:
        fmt.Println("Restarting, actor is about restart")
    case Hello:
        fmt.Printf("Hello %v\n", msg.Who)
        panic("Ouch")
    }
}

func NewChildActor() actor.Actor {
    return &ChildActor{}
}

func main() {
    decider := func(reason interface{}) actor.Directive {
        log.Printf("handling failure for child. reason:%v", reason)
        return actor.RestartDirective
    }
    supervisor := actor.NewOneForOneStrategy(10, 1000, decider)

    ctx := actor.NewActorSystem().Root
    props := actor.PropsFromProducer(NewParentActor).WithSupervisor(supervisor)

    pid := ctx.Spawn(props)
    ctx.Send(pid, Hello{Who: "Roger"})

    console.ReadLine()
}

网络/远程通信示例

节点1代码

type MyActor struct {
    count int
}

func (state *MyActor) Receive(context actor.Context) {
    switch context.Message().(type) {
    case *messages.Response:
        state.count++
        fmt.Println(state.count)
    }
}

func main() {
    remote.Start("localhost:8090")

    context := actor.EmptyRootContext
    props := actor.PropsFromProducer(func() actor.Actor { return &MyActor{} })
    pid, _ := context.Spawn(props)
    message := &messages.Echo{Message: "hej", Sender: pid}

    // 创建远程actor
    spawnResponse, _ := remote.SpawnNamed("localhost:8091", "myactor", "hello", time.Second)

    // 获取创建的PID
    spawnedPID := spawnResponse.Pid
    for i := 0; i < 10; i++ {
        context.Send(spawnedPID, message)
    }

    console.ReadLine()
}

节点2代码

type MyActor struct{}

func (*MyActor) Receive(context actor.Context) {
    switch msg := context.Message().(type) {
    case *messages.Echo:
        context.Send(msg.Sender, &messages.Response{
            SomeValue: "result",
        })
    }
}

func main() {
    remote.Start("localhost:8091")

    // 注册本地actor以便远程创建
    remote.Register("hello", actor.PropsFromProducer(func() actor.Actor { return &MyActor{} }))
    console.ReadLine()
}

消息协议定义

syntax = "proto3";
package messages;
import "actor.proto"; // 需要导入actor.proto以便消息可以包含PID

// 节点1发送给节点2的消息
message Echo {
  actor.PID Sender = 1; // 远程actor应回复的PID
  string Message = 2;
}

// 远程actor应回复的消息
message Response {
  string SomeValue = 1;
}

构建和测试

确保$GOPATH变量正确设置后，执行以下命令：

go get github.com/asynkron/protoactor-go/...
cd $GOPATH/src/github.com/asynkron/protoactor-go
go get ./...
make

运行测试：

go test `go list ./... | grep -v "/examples/" | grep -v "/persistence" | grep -v "/scheduler"`

ProtoActor是一个仍在beta阶段的Go实现，但已经有一些生产环境用户。请注意API在1.0版本前可能会发生变化。