如何让一个角色在一个进程上运行,将消息发送给在另一个进程上运行的另一个actor? [英] How can I have an actor running on one process send a message to another actor running on a separate process?
问题描述
我希望运行在各个进程(或节点)上的参与者在将消息发送给运行在不同进程(或节点)上的其他参与者的同时,又要保持容错和负载平衡.我目前正在尝试使用Akka.Cluster的分片功能来完成此操作.
I want to have actors running on various processes (or nodes) send messages to other actors running off of different processes (or nodes), all while maintaining fault-tolerance and load balancing. I am currently attempting to use Akka.Cluster's Sharding feature to accomplish this.
但是,我不确定如何做到这一点...
However, I am not sure how to accomplish this...
我有以下代码可以反映我的种子节点:
let configurePort port =
let config = Configuration.parse ("""
akka {
actor {
provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"
serializers {
hyperion = "Akka.Serialization.HyperionSerializer, Akka.Serialization.Hyperion"
}
serialization-bindings {
"System.Object" = hyperion
}
}
remote {
helios.tcp {
public-hostname = "localhost"
hostname = "localhost"
port = """ + port.ToString() + """
}
}
cluster {
auto-down-unreachable-after = 5s
seed-nodes = [ "akka.tcp://cluster-system@localhost:2551/" ]
}
persistence {
journal.plugin = "akka.persistence.journal.inmem"
snapshot-store.plugin = "akka.persistence.snapshot-store.local"
}
}
""")
config.WithFallback(ClusterSingletonManager.DefaultConfig())
let consumer (actor:Actor<_>) msg = printfn "\n%A received %s" (actor.Self.Path.ToStringWithAddress()) msg |> ignored
// spawn two separate systems with shard regions on each of them
let system1 = System.create "cluster-system" (configurePort 2551)
let shardRegion1 = spawnSharded id system1 "shardRegion1" <| props (actorOf2 consumer)
System.Threading.Thread.Sleep(1000)
let system2 = System.create "cluster-system" (configurePort 2552)
let shardRegion2 = spawnSharded id system2 "shardRegion2" <| props (actorOf2 consumer)
System.Threading.Thread.Sleep(1000)
let system3 = System.create "cluster-system" (configurePort 2553)
let shardRegion3 = spawnSharded id system3 "shardRegion3" <| props (actorOf2 consumer)
System.Threading.Thread.Sleep(3000)
// NOTE: Even thou we sent all messages through single shard region,
// some of them will be executed on the second and third one thanks to shard balancing
System.Threading.Thread.Sleep(3000)
shardRegion1 <! ("shard-1", "entity-1", "hello world 1")
shardRegion1 <! ("shard-1", "entity-2", "hello world 2")
shardRegion1 <! ("shard-2", "entity-3", "hello world 3")
shardRegion1 <! ("shard-2", "entity-4", "hello world 4")
System.Threading.Thread.Sleep(1000)
let printShards shardRegion =
async {
let! (reply:AskResult<ShardRegionStats>) = (retype shardRegion) <? GetShardRegionStats.Instance
let (stats: ShardRegionStats) = reply.Value
for kv in stats.Stats do
printfn "\tShard '%s' has %d entities on it" kv.Key kv.Value
} |> Async.RunSynchronously
let printNodes() =
printfn "\nShards active on node 'localhost:2551':"
printShards shardRegion1
printfn "\nShards active on node 'localhost:2552':"
printShards shardRegion2
printfn "\nShards active on node 'localhost:2553':"
printShards shardRegion3
printNodes()
输出看起来像这样:
Shards active on node 'localhost:2551':
Shard 'shard-1' has 2 entities on it
Shard 'shard-2' has 2 entities on it
在节点'localhost:2552'上处于活动状态的碎片:
Shards active on node 'localhost:2552':
然后我有一个单独的进程,该进程执行以下代码:
let configurePort port =
let config = Configuration.parse ("""
akka {
actor {
provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"
serializers {
hyperion = "Akka.Serialization.HyperionSerializer, Akka.Serialization.Hyperion"
}
serialization-bindings {
"System.Object" = hyperion
}
}
remote {
helios.tcp {
public-hostname = "localhost"
hostname = "localhost"
port = "0"
}
}
cluster {
auto-down-unreachable-after = 5s
seed-nodes = [ "akka.tcp://cluster-system@localhost:2551/" ]
}
persistence {
journal.plugin = "akka.persistence.journal.inmem"
snapshot-store.plugin = "akka.persistence.snapshot-store.local"
}
}
""")
config.WithFallback(ClusterSingletonManager.DefaultConfig())
let consumer (actor:Actor<_>) msg = printfn "\n%A received %s" (actor.Self.Path.ToStringWithAddress()) msg |> ignored
// spawn two separate systems with shard regions on each of them
let system1 = System.create "cluster-system" (configurePort 2554)
let shardRegion1 = spawnSharded id system1 "printer" <| props (actorOf2 consumer)
System.Threading.Thread.Sleep(1000)
let system2 = System.create "cluster-system" (configurePort 2555)
let shardRegion2 = spawnSharded id system2 "printer" <| props (actorOf2 consumer)
System.Threading.Thread.Sleep(1000)
let system3 = System.create "cluster-system" (configurePort 2556)
let shardRegion3 = spawnSharded id system3 "printer" <| props (actorOf2 consumer)
我的集群系统(在单独的进程上运行)可以识别正在加入的新节点:
> [INFO][3/15/2017 9:12:13 PM][Thread 0054][[akka://cluster-system/system/cluster/core/daemon#2086121649]] Node [akka.tcp://cluster-system@localhost:52953] is JOINING, roles []
[INFO][3/15/2017 9:12:14 PM][Thread 0006][[akka://cluster-system/system/cluster/core/daemon#2086121649]] Node [akka.tcp://cluster-system@localhost:52956] is JOINING, roles []
[INFO][3/15/2017 9:12:15 PM][Thread 0054][[akka://cluster-system/system/cluster/core/daemon#2086121649]] Node [akka.tcp://cluster-system@localhost:52961] is JOINING, roles []
[INFO][3/15/2017 9:12:18 PM][Thread 0055][[akka://cluster-system/system/cluster/core/daemon#2086121649]] Leader is moving node [akka.tcp://cluster-system@localhost:52953] to [Up]
[INFO][3/15/2017 9:12:18 PM][Thread 0055][[akka://cluster-system/system/cluster/core/daemon#2086121649]] Leader is moving node [akka.tcp://cluster-system@localhost:52956] to [Up]
[INFO][3/15/2017 9:12:18 PM][Thread 0055][[akka://cluster-system/system/cluster/core/daemon#2086121649]] Leader is moving node [akka.tcp://cluster-system@localhost:52961] to [Up]
结论:
总而言之,我想让运行在各种进程(或节点)上的参与者在保持容错和负载平衡的同时,向运行在不同进程(或节点)上的其他参与者发送消息.我目前正在尝试使用Akka.Cluster的分片功能来完成此操作.
In conclusion, I want to have actors running on various processes (or nodes) send messages to other actors running off of different processes (or nodes) while maintaining fault-tolerance and load balancing. I am currently attempting to use Akka.Cluster's Sharding feature to accomplish this.
附录:
open System
open System.IO
#if INTERACTIVE
let cd = Path.Combine(__SOURCE_DIRECTORY__, "../src/Akkling.Cluster.Sharding/bin/Debug")
System.IO.Directory.SetCurrentDirectory(cd)
#endif
#r "../src/Akkling.Cluster.Sharding/bin/Debug/System.Collections.Immutable.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akka.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Hyperion.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Newtonsoft.Json.dll"
#r @"C:\Users\Snimrod\Documents\Visual Studio 2015\Projects\Temp\packages\Akka.FSharp.1.1.3\lib\net45\Akka.FSharp.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/FSharp.PowerPack.Linq.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Helios.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/FsPickler.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Google.ProtocolBuffers.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Google.ProtocolBuffers.Serialization.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akka.Remote.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Google.ProtocolBuffers.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akka.Persistence.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akka.Cluster.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akka.Cluster.Tools.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akka.Cluster.Sharding.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akka.Serialization.Hyperion.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akkling.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akkling.Persistence.dll"
#r "../src/Akkling.Cluster.Sharding/bin/Debug/Akkling.Cluster.Sharding.dll"
open Akka.Actor
open Akka.Configuration
open Akka.Cluster
open Akka.Cluster.Tools.Singleton
open Akka.Cluster.Sharding
open Akka.Persistence
open Akkling
open Akkling.Persistence
open Akkling.Cluster
open Akkling.Cluster.Sharding
open Hyperion
推荐答案
为了保持分片及其位置的一致视图,Akka.Cluster.Sharding持久化后端必须指向一个对所有进程可见的数据库.在您的配置中,您使用的是akka.persistence.journal.inmem
,它是内存中的数据存储(仅用于测试和开发).在其他进程中将看不到它.
In order to maintain consistent view of the shards and their locations, Akka.Cluster.Sharding persistent backend must point to a database that is visible to all of the processes. In your configuration, you're using akka.persistence.journal.inmem
which is in-memory data store (used only for tests and development). It won't be visible from another processes.
您需要配置一个持久后端,以使分片在不同机器/进程上的节点之间可见.您可以使用 Akka.Persistence.SqlServer 或任何其他插件.这是仅用于分片的持久化后端的最基本配置:
You'll need to configure a persistent backend in order for shards to be visible between nodes living on different machines/processes. You can do that i.e. by using Akka.Persistence.SqlServer or any other plugin. This is the most basic configuration for your persistence backend used only by sharding:
akka.persistence {
journal {
plugin = "akka.persistence.journal.sql-server"
sql-server {
connection-string = "<connection-string>"
auto-initialize = on
}
}
snapshot-store {
plugin = "akka.persistence.snapshot-store.sql-server"
sql-server {
connection-string = "<connection-string>"
auto-initialize = on
}
}
}
有关实用性的更多信息,请参阅本文
For something more practical, please refer to this article.
还请记住,Akka.Cluster.Sharding和Akka.Persistence插件仅在预发行模式下可用(因此您需要使用-pre标志安装软件包).
Also keep in mind that both Akka.Cluster.Sharding and Akka.Persistence plugins are available only in prerelease mode (so you need to install-package with -pre flag).
这篇关于如何让一个角色在一个进程上运行,将消息发送给在另一个进程上运行的另一个actor?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!