走進 Akka.NET

官方文檔：https://getakka.net/index.html

官網：https://petabridge.com/

 

一、Akka.NET 是什么？

Akka 是一個構建高並發、分布式和彈性消息驅動的工具包。Akka.NET 是 Akka 的一個 .NET 的移植庫。

Akka.NET 內部都是 Actor 構成的，Actor 是一個狀態、行為、郵箱、子節點和監視者策略構成的容器。

 

二、Akka.NET 的一些基礎模塊

Akka - 核心 Actor 庫
Akka.Remote - 跨節點 Actor 部署/通信
Akka.Cluster - 彈性 Actor 網絡(HA)
Akka.Persistence - 事件源, 持久化 Actor 狀態 & 恢復
Akka.Streams - 流式工作流
Akka.Cluster.Tools - 集群單例, 分布式發布/訂閱
Akka.Cluster.Sharding - 持久化狀態分區
Akka.DData - 最終一直的數據復制

 

三、Akka.NET 的架構

 

如圖所示，一個 akka 系統有一個跟節點 root，然后 root 有2個子節點，user 和 system ，你定義的 actor 都在 user 下，system 下都是系統定義的

 

四、Akka.NET 發送消息的規則

• 最多一次交付

• 每個配對的發送者、接收者對消息保持排序。

 

五、基礎 Actor 類型

• ReceiverActor：如果使用這個類型的 Actor 為基類，使用 Receive<T>(msg=>{}); 來接收消息
• UntypedActor：如果使用這個類型的 Actor 為基類，需要重寫 OnReceive（object message） 方法，然后區分 message 類型

 

六、使用 Akka.NET

• 普通模式

因為 Helloworld 的例子太多了，官網也有，這里直接略過。

代碼在這里  

代碼里有些特殊情況的處理沒寫，這里簡單說一下

比如 sender 發送給了 receiver 一個消息，但是 receiver 還有些事情沒有准備好，比如發送來一個消息，但是可能需要先登錄，那就先緩存下來，等辦完事再處理

流程是：先 Stash 消息，開啟 Scheduler ，發送 GetToken 消息，token 獲取成功返回 GetTokenSuccess 消息，取消 Scheduler ，變成 Redy 狀態，正常處理消息

public ReceiveActor()
{
        private ICancelable _cancel;

        public ReceiveActor()
        {
            Receive<TestMsg>(msg =>
            {
                Stash.Stash();
            });

            Receive<GetToken>(msg =>
            {
                GetTokenProcess();
            }

            Receive<GetTokenSuccess>(msg =>
            {
                _cancel?.Cancel();
                _cancel = null;

                Become(Ready);
            });

            if (_cancel == null)
            {
                _cancel = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(TimeSpan.Zero, TimeSpan.FromSeconds(10), Self, new GetToken(), Self);
            }
        }

        private void Ready()
        {
            Receive<TestMsg>(msg =>
            {
                //Todo
            });

            Stash.UnstashAll();
        }

        private void GetTokenProcess()
        {
            var tokenActor = Context.ActorOf(Props.Create(() => new TokenActor("url")));
            tokenActor.Ask<GetTokenSuccess>(new GetToken
            {
                Token = "",
            }).ContinueWith(tr =>
            {
                if (tr.IsCanceled || tr.IsFaulted)
                {
                    return new GetTokenSuccess();
                }

                return tr.Result;
            }).PipeTo(Self);
        }

        public class GetToken
        {
            public string Token { get; set; }
        }
}

 

 

• Router

Actor 內部其實是單線程的處理消息的，但是用了 Router 可以把消息分出去給多個節點處理，這里分為2種模式，group 和 pool

先來看 group 配置

akka {
   actor{
      provider = cluster
      deployment {
        /api/myClusterGroupRouter {
          router = broadcast-group # routing strategy
          routees.paths = ["/user/api"] # path of routee on each node
          nr-of-instances = 3 # max number of total routees
          cluster {
             enabled = on
             allow-local-routees = on
             use-role = crawler
          }
        }
      }
   }
}

 

 如圖看到在 89節點上注冊了 90 和 91 節點的 Routees

 

pool 模式

akka {
   actor{
      provider = cluster
      deployment {
        /api/myClusterPoolRouter {
          router = round-robin-pool # routing strategy
          nr-of-instances = 10 # max number of total routees
          cluster {
             enabled = on
             allow-local-routees = on
             use-role = crawler
             max-nr-of-instances-per-node = 1
          }
        }
      }
   }
}

這種模式其實是在 90 和 91 的節點上分別注冊了 89 的 routee，89上是akka.tcp://Sys@localhost:90/remote/akka/tcp/localhost:89/$a 和 akka.tcp://Sys@localhost:91/remote/akka/tcp/localhost:89/$a

 具體代碼在這里 

• Shard

用邏輯 id 聯系，不關心集群中的物理位置或管理他們的創建，可以 rebalance （消耗資源比較大，只有差異比較大的時候才會，默認 rebalance-threshold = 10，shard 數量設置為集群節點最大數量的10倍，shard 數量太大消耗大）

整個 ActorSystem 中只能有一個Codinator，在最老的節點上，因為最老的節點被認為是安全的。

如圖所示整個系統分為了 Coordinator ，Shard region（typeName）， Shard，Entity 部分，節點的路徑是  /user/sharding/<typeName>/<shardId>/<entityId> 和普通的路徑 /user/<actorName> 不一樣

這里使用了 Persistent ，每種數據庫的配置完全不同，我是 Redis 的配置，建議使用即時更新的數據庫，不更新的很可能是棄用的，比如 mysql。

具體代碼在這里

這里需要注意的是：

1. 只能有一個 Lighthourse —— 否在爆 ID 13 error
2. 創建 ActorSystem 的時候后邊加上 _config.WithFallback(ClusterClientReceptionist.DefaultConfig()).WithFallback(DistributedPubSub.DefaultConfig()); —— 否在爆 ID 9 error

• 區別

Shard 看着好像也能處理 Router 的問題，但這里又有些區別，此處摘自官方文檔，用的自動翻譯，請自行理解下：

在一致的散列場景中，整個密鑰空間被拼接在我們在路由器配置中定義的參與者數量之間。這究竟意味着什么？單個參與者實例負責處理整個范圍的密鑰。例如：假設在一個一致的哈希路由器后面，我們有一個數字1-100的密鑰空間，在5個參與者之間平均共享。這意味着，第一個參與者將負責處理ID為1-20、第二個參與者21-40、第三個參與者41-60等的所有消息。如果您的目標是每個標識符有一個唯一的參與者，那么對於您的情況來說，這不是一個有效的場景。在集群分片場景的另一邊，每個實體（這就是我們如何引用分片參與者）由對（ShardId，EntityId）標識。它是1-1關系，因此用同一對標識的消息保證總是路由到相同的單個實體。 另一個區別是靈活性。一致的哈希路由器幾乎是靜態的，在您想要調整集群大小的情況下，它們不能很好地工作。你記得以前提到過的拼接鍵空間的概念嗎？想象一下，一旦我們改變分配給它的參與者/節點的實際數量，它將如何工作。當然，這些關鍵范圍也必須改變。因此，在集群大小調整之后，以前處理ID為1-20的消息的演員現在可以處理不同的范圍（即，1-15）。這稱為“分區切換”。當使用集群切分時，我們的參與者將能夠隨着集群大小的改變遷移到其他節點，即使在集群大小改變時，也能夠保持消息id和分割實體之間的關系。

 

七、總結

除了 stream 沒研究，其他的多少都看了一點，因為我沒有使用 stream 的場景，其他發現 Helloworld 的入門真的沒什么用，只是讓你理解 Actor 的開發的思想和模式，實際使用需要多看文檔，一點點研究，還有就是官網的博文，再就是一些大神寫的例子，

這里放幾個網址供大家學習

GitHub：

https://github.com/cgstevens/FileProcessor

https://github.com/Horusiath/AkkaDemos

https://github.com/petabridge/akkadotnet-code-samples

https://github.com/Lutando/Akkatecture

 

文檔：

Streams：https://petabridge.com/blog/akkadotnet-11-cluster-streams/

Distributed Pub-Sub：https://petabridge.com/blog/distributed-pub-sub-intro-akkadotnet/

Shard：https://petabridge.com/blog/introduction-to-cluster-sharding-akkadotnet/

Shard：https://petabridge.com/blog/cluster-sharding-technical-overview-akkadotnet/

Petabridge.Cmd：https://petabridge.com/blog/petabridgecmd-release/

Moonitoring：https://github.com/petabridge/akka-monitoring

你是不是感覺很奇怪，沒有 Dashboard 呢，其實是有，但是收費，在這里：https://phobos.petabridge.com/