Dapr是为云上环境设计的跨语言, 事件驱动, 可以便捷的构建微服务的系统. balabala一堆, 有兴趣的小伙伴可以去了解一下.

Dapr提供有状态和无状态的微服务. 大部分人都是做无状态服务(微服务)的, 只是某些领域无状态并不好使, 因为开销实在是太大了; 有状态服务有固定的场景, 就是要求开销小, 延迟和吞吐都比较高. 废话少说, 直接来看Dapr是怎么实现有状态服务的.

 

先来了解一下有状态服务:

1. 稳定的路由

   发送给A服务器的请求, 不能发给B服务器, 否则就是无状态的

2. 状态

   状态保存在自己服务器内部, 而不是远程存储, 这一点和无状态有很明显的区别, 所以无状态服务需要用redis这种东西加速, 有状态不需要

3. 处理是单线程

   状态一般来讲比较复杂, 想要对一个比较复杂的东西进行并行的计算是比较困难的; 当然A和B的逻辑之间没有关系, 其实是可以并行的, 但是A自己本身的逻辑执行需要串行执行.

 

对于一个有状态服务来讲(dapr), 实现23实际上是很轻松的, 甚至有一些是用户需要实现的东西, 所以1才是关键, 当前这个消息(请求)需要被发送到哪个服务器上面处理才是最关键的, 甚至决定了他是什么系统.

决定哪个请求的目标地址, 这个东西在分布式系统里面叫Placement, 有时候也叫Naming. TiDB里面有一个Server叫PlacementDriver, 简称PD, 其实就是在干同样的事情.

好了, 开始研究Dapr的Placement是怎么实现的.

 

有一个Placement的进程, 2333, 目录cmd/placement, 就看他了

  1. func main() {
  2. log.Infof("starting Dapr Placement Service -- version %s -- commit %s", version.Version(), version.Commit())
  3.  
  4. cfg := newConfig()
  5.  
  6. // Apply options to all loggers.
  7. if err := logger.ApplyOptionsToLoggers(&cfg.loggerOptions); err != nil {
  8. log.Fatal(err)
  9. }
  10. log.Infof("log level set to: %s", cfg.loggerOptions.OutputLevel)
  11.  
  12. // Initialize dapr metrics for placement.
  13. if err := cfg.metricsExporter.Init(); err != nil {
  14. log.Fatal(err)
  15. }
  16.  
  17. if err := monitoring.InitMetrics(); err != nil {
  18. log.Fatal(err)
  19. }
  20.  
  21. // Start Raft cluster.
  22. raftServer := raft.New(cfg.raftID, cfg.raftInMemEnabled, cfg.raftBootStrap, cfg.raftPeers)
  23. if raftServer == nil {
  24. log.Fatal("failed to create raft server.")
  25. }
  26.  
  27. if err := raftServer.StartRaft(nil); err != nil {
  28. log.Fatalf("failed to start Raft Server: %v", err)
  29. }
  30.  
  31. // Start Placement gRPC server.
  32. hashing.SetReplicationFactor(cfg.replicationFactor)
  33. apiServer := placement.NewPlacementService(raftServer)

可以看到main函数里面启动了一个raft server, 一般这样的话, 就说明在某些能力方面做到了强一致性.

raft库用的是consul实现的raft, 而不是etcd, 因为etcd的raft不是库, 只能是一个服务器(包括etcd embed), 你不能定制里面的协议, 你只能使用etcd提供给你的client来访问他. 这一点etcd做的非常不友好.

 

如果用raft库来做placement, 那么协议可以定制, 可以找Apply相关的函数, 因为raft状态机只是负责log的一致性, log即消息, 消息的处理则表现出来状态, Apply函数就是需要用户做消息处理的地方. 幸亏之前有做过MIT 6.824的lab, 对这个稍微有一点了解.

  1. // Apply log is invoked once a log entry is committed.
  2. func (c *FSM) Apply(log *raft.Log) interface{} {
  3. buf := log.Data
  4. cmdType := CommandType(buf[0])
  5.  
  6. if log.Index < c.state.Index {
  7. logging.Warnf("old: %d, new index: %d. skip apply", c.state.Index, log.Index)
  8. return nil
  9. }
  10.  
  11. var err error
  12. var updated bool
  13. switch cmdType {
  14. case MemberUpsert:
  15. updated, err = c.upsertMember(buf[1:])
  16. case MemberRemove:
  17. updated, err = c.removeMember(buf[1:])
  18. default:
  19. err = errors.New("unimplemented command")
  20. }
  21.  
  22. if err != nil {
  23. return err
  24. }
  25.  
  26. return updated
  27. }

在pkg/placement/raft文件夹下面找到raft相关的代码, fsm.go里面有对消息的处理函数.

可以看到, 消息的处理非常简单, 里面只有MemberUpsert, 和MemberRemove两个消息.  FSM状态机内保存的状态只有:

  1. // DaprHostMemberState is the state to store Dapr runtime host and
  2. // consistent hashing tables.
  3. type DaprHostMemberState struct {
  4. // Index is the index number of raft log.
  5. Index uint64
  6. // Members includes Dapr runtime hosts.
  7. Members map[string]*DaprHostMember
  8.  
  9. // TableGeneration is the generation of hashingTableMap.
  10. // This is increased whenever hashingTableMap is updated.
  11. TableGeneration uint64
  12.  
  13. // hashingTableMap is the map for storing consistent hashing data
  14. // per Actor types.
  15. hashingTableMap map[string]*hashing.Consistent
  16. }

很明显, 这里面只有DaprHostMember这个有用的信息, 而DaprHostMember就是集群内的节点.

 

这里可以分析出来, Dapr通过Raft协议来维护了一个强一致性的Membership, 除此之外什么也没干….据我的朋友说, 跟Orleans是有一点类似的, 只是Orleans是AP系统.

 

再通过对一致性Hash的分析, 可以看到:

  1. func (a *actorsRuntime) lookupActorAddress(actorType, actorID string) (string, string) {
  2. if a.placementTables == nil {
  3. return "", ""
  4. }
  5.  
  6. t := a.placementTables.Entries[actorType]
  7. if t == nil {
  8. return "", ""
  9. }
  10. host, err := t.GetHost(actorID)
  11. if err != nil || host == nil {
  12. return "", ""
  13. }
  14. return host.Name, host.AppID
  15. }

通过 ActorType和ActorID到一致性的Hash表中去找host, 那个GetHost实现就是一致性Hash表实现的.

Actor RPC Call的实现:

  1. func (a *actorsRuntime) Call(ctx context.Context, req *invokev1.InvokeMethodRequest) (*invokev1.InvokeMethodResponse, error) {
  2. if a.placementBlock {
  3. <-a.placementSignal
  4. }
  5.  
  6. actor := req.Actor()
  7. targetActorAddress, appID := a.lookupActorAddress(actor.GetActorType(), actor.GetActorId())
  8. if targetActorAddress == "" {
  9. return nil, errors.Errorf("error finding address for actor type %s with id %s", actor.GetActorType(), actor.GetActorId())
  10. }
  11.  
  12. var resp *invokev1.InvokeMethodResponse
  13. var err error
  14.  
  15. if a.isActorLocal(targetActorAddress, a.config.HostAddress, a.config.Port) {
  16. resp, err = a.callLocalActor(ctx, req)
  17. } else {
  18. resp, err = a.callRemoteActorWithRetry(ctx, retry.DefaultLinearRetryCount, retry.DefaultLinearBackoffInterval, a.callRemoteActor, targetActorAddress, appID, req)
  19. }
  20.  
  21. if err != nil {
  22. return nil, err
  23. }
  24. return resp, nil
  25. }

通过刚才我们看到loopupActorAddress函数找到的Host, 然后判断是否是在当前Host宿主内, 否则就发送到远程, 对当前宿主做了优化, 实际上没鸡儿用, 因为分布式系统里面, 一般都会有很多个host, 在当前host内的概率实际上是非常低的.

 

从这边, 我们大概就能分析到全貌, 即Dapr实现分布式有状态服务的细节:

1. 通过Consul Raft库维护Membership

2. 集群和Placement组件通讯, 获取到Membership

3. 寻找Actor的算法实现在Host内, 而不是Placement组件. 通过ActorType找到可以提供某种服务的Host, 然后组成一个一致性Hash表, 到该表内查找Host, 进而转发请求

 

对Host内一致性Hash表的查找引用, 找到了修改内容的地方:

  1. func (a *actorsRuntime) updatePlacements(in *placementv1pb.PlacementTables) {
  2. a.placementTableLock.Lock()
  3. defer a.placementTableLock.Unlock()
  4.  
  5. if in.Version != a.placementTables.Version {
  6. for k, v := range in.Entries {
  7. loadMap := map[string]*hashing.Host{}
  8. for lk, lv := range v.LoadMap {
  9. loadMap[lk] = hashing.NewHost(lv.Name, lv.Id, lv.Load, lv.Port)
  10. }
  11. c := hashing.NewFromExisting(v.Hosts, v.SortedSet, loadMap)
  12. a.placementTables.Entries[k] = c
  13. }
  14.  
  15. a.placementTables.Version = in.Version
  16. a.drainRebalancedActors()
  17.  
  18. log.Infof("placement tables updated, version: %s", in.GetVersion())
  19.  
  20. a.evaluateReminders()
  21. }
  22. }

从这几行代码可以看出, 版本不不一样, 就会全更新, 而且还会进行rehash, 就是a.drainRebalanceActors. 

如果学过数据结构, 那么肯定学到过一种东西叫HashTable, HashTable在扩容的时候需要rehash, 需要构建一个更大的table, 然后把所有元素重新放进去, 位置会和原先的大不一样. 而一致性Hash可以解决全rehash的情况, 只让部分内容rehash, 失效的内容会比较少.

但是, 凡事都有一个但是, 所有的节点都同时rehash还好, 可一个分布式系统怎么做到所有node都同时rehash, 很显然是做不到的, 所以Dapr维护的Actor Address目录, 是最终一致的, 也就是系统里面会存在多个ID相同的Actor(短暂的), 还是会导致不一致.

 

对dapr/proto/placement/v1/placement.proto查看, 验证了我的猜想

  1. // Placement service is used to report Dapr runtime host status.
  2. service Placement {
  3. rpc ReportDaprStatus(stream Host) returns (stream PlacementOrder) {}
  4. }
  5.  
  6. message PlacementOrder {
  7. PlacementTables tables = 1;
  8. string operation = 2;
  9. }

Host启动, 就去placement那边通过gRPC Stream订阅了集群的变动. 懒到极点了, 居然是把整个membership发送过来, 而不是发送的diff.

 

总结一下, 从上面的源码分析我们可以知道, Dapr的Membership是CP系统, 但是Actor的Placement不是, 是一个最终一致的AP系统. 而TiDB的PD是一个CP系统, 只不过是通过etcd embed做的. 希望对大家有一点帮助.

对我有帮助的, 可能就是Dapr对于Consul raft的使用.

 

参考:

1. Dapr

2. Etcd Embed

3. Consul Raft

 

版权声明:本文为egmkang原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://www.cnblogs.com/egmkang/p/13934225.html