快捷搜索:  汽车  科技

akka 原理(终于有人把AkkaCluster原理与应用讲明白了)

akka 原理(终于有人把AkkaCluster原理与应用讲明白了)我们基于Akka实现了一个简单的模拟日志实时处理的集群系统,可以从任何数据源输入数据,如文件、消息中间件Kafka、数据库,还可以是一个远程调用请求,我们收集数据,然数据经过一个拦截器层,最后解析处理数据为特定格式,最后数据写入Kafka。具体实现逻辑如下图所示:Akka支持在每个成员节点加入集群的时候,设置成员自己的角色。通过角色划分,可以将使用Akka集群处理业务的系统划分为多个处理逻辑独立的子系统,每个子系统处理自己的业务逻辑,而且,划分得到的多个子系统都处于一个统一的Akka集群中。因此,每个子系统也具备了Akka集群所具有的特性,如故障检测、状态转移、状态传播等等。节点状态发生转移会触发某个事件,我们可以根据不同类型的事件来进行相应的处理,为了能够详细捕获到各种事件,我们先看一下Akka定义的事件集合,如图所示:通常,在基于Akka Cluster的应用中实现Actor时,可以重

Akka集群原理

Akka集群支持去中心化的基于P2P的集群服务,没有单点故障(SPOF)问题,它主要是通过Gossip协议来实现。对于集群成员的状态,Akka提供了一种故障检测机制,能够自动发现出现故障而离开集群的成员节点,通过事件驱动的方式,将状态传播到整个集群的其它成员节点。

  • 状态转移与故障检测

Akka内部为集群成员定义了一组有限状态(6种状态),并给出了一个状态转移矩阵,代码如下所示:

private[cluster] val allowedTransitions: Map[MemberStatus Set[MemberStatus]] = Map( Joining -> Set(Up Down Removed) Up -> Set(Leaving Down Removed) Leaving -> Set(Exiting Down Removed) Down -> Set(Removed) Exiting -> Set(Removed Down) Removed -> Set.empty[MemberStatus]) }

Akka集群中的每个成员节点,都有可能处于上面的一种状态,在发生某些事件以后,会发生状态转移。需要注意的是,除了Down和Removed状态以外,节点处于其它任何一个状态时都有可能变成Down状态,即节点故障而无法提供服务,而在变成Down状态之前有一个虚拟的Unreachable状态,因为在Gossip收敛过程中,是无法到达或者经由Unreachable状态的节点,这个状态是由Akka实现的故障探测器(Failure Detector)来检测到的。处于Down状态的节点如果想要再次加入Akka集群,需要重新启动,并进入Joining状态,然后才能进行后续状态的转移变化。Akka集群成员节点状态及其转移情况,如下图所示:

akka 原理(终于有人把AkkaCluster原理与应用讲明白了)(1)

我们说明一下Akka中的故障检测机制。在Akka中,集群中每一个成员节点M会被集群中的其他另一组节点(默认是5个)G监控,这一组节点G并不是整个集群中的其他所有节点,只是整个集群全部节点的一个子集,组G中的节点会检测节点M是否处于Unreachable状态,这是通过发送心跳来确认节点M是否可达,如果不可达则组G中的节点会将节点M的Unreachable状态向集群中组G之外的其它节点传播,最终使得集群中的每个成员节点都知道节点M故障。

  • Akka事件集合

节点状态发生转移会触发某个事件,我们可以根据不同类型的事件来进行相应的处理,为了能够详细捕获到各种事件,我们先看一下Akka定义的事件集合,如图所示:

akka 原理(终于有人把AkkaCluster原理与应用讲明白了)(2)

通常,在基于Akka Cluster的应用中实现Actor时,可以重写Actor的preStart方法,通过Cluster来订阅集群事件,代码示例如下所示:

val cluster = Cluster(context.system) override def preStart(): Unit = { cluster.subscribe(self initialStateMode = InitialStateAsEvents classOf[MemberUp] classOf[MemberRemoved] classOf[UnreachableMember]) }

例如,对于MemberUp事件,我们可以获取到对应Actor的引用ActorRef,然后通过与其进行消息交换,一起协同完成特定任务。

  • Akka成员角色(Node Role)

Akka支持在每个成员节点加入集群的时候,设置成员自己的角色。通过角色划分,可以将使用Akka集群处理业务的系统划分为多个处理逻辑独立的子系统,每个子系统处理自己的业务逻辑,而且,划分得到的多个子系统都处于一个统一的Akka集群中。因此,每个子系统也具备了Akka集群所具有的特性,如故障检测、状态转移、状态传播等等。

Akka集群应用实践

我们基于Akka实现了一个简单的模拟日志实时处理的集群系统,可以从任何数据源输入数据,如文件、消息中间件Kafka、数据库,还可以是一个远程调用请求,我们收集数据,然数据经过一个拦截器层,最后解析处理数据为特定格式,最后数据写入Kafka。具体实现逻辑如下图所示:

akka 原理(终于有人把AkkaCluster原理与应用讲明白了)(3)

上图中,我们将日志实时处理系统分为3个子系统,通过Akka的Role来进行划分,3个角色分别为collector、interceptor、processor,3个子系统中的节点都是整个Akka集群的成员。整个集群系统中的数据流向是:collector接收数据(或者直接对接特定数据源而产生数据),我们这里模式发送Nginx日志记录行,将数据发送到interceptor;interceptor收到collector发送的日志记录行,解析出请求的真是IP地址,拦截在黑名单IP列表中的请求,如果IP地址不在黑名单,则发送给processor去处理;processor对整个日志记录行进行处理,最后保存到Kakfa中。我们抽象出用来订阅集群事件相关的逻辑,实现抽象类为ClusterRoledWorker,代码如下所示:

package org.shirdrn.scala.akka.cluster import akka.actor._ import akka.cluster.ClusterEvent.{InitialStateAsEvents MemberEvent MemberUp UnreachableMember} import akka.cluster.{Cluster Member} abstract class ClusterRoledWorker extends Actor with ActorLogging { // 创建一个Cluster实例 val cluster = Cluster(context.system) // 用来缓存下游注册过来的子系统ActorRef var workers = IndexedSeq.empty[ActorRef] override def preStart(): Unit = { // 订阅集群事件 cluster.subscribe(self initialStateMode = InitialStateAsEvents classOf[MemberUp] classOf[UnreachableMember] classOf[MemberEvent]) } override def postStop(): Unit = cluster.unsubscribe(self) /** * 下游子系统节点发送注册消息 */ def register(member: Member createPath: (Member) => ActorPath): Unit = { val actorPath = createPath(member) log.info("Actor path: " actorPath) val actorSelection = context.actorSelection(actorPath) actorSelection ! Registration } }

另外,定义了一些case class作为消息,方便在各个Actor之间进行发送/接收,代码如下所示:

package org.shirdrn.scala.akka.cluster object Registration extends Serializable trait EventMessage extends Serializable case class RawNginxRecord(sourceHost: String line: String) extends EventMessage case class NginxRecord(sourceHost: String eventCode: String line: String) extends EventMessage case class FilteredRecord(sourceHost: String eventCode: String line: String logDate: String realIp: String) extends EventMessage

Akka Cluster使用一个配置文件,用来指定一些有关Actor的配置,我们使用的配置文件为application.conf,配置内容如下所示:

akka { loglevel = INFO stdout-loglevel = INFO event-handlers = ["akka.event.Logging$DefaultLogger"] actor { provider = "akka.cluster.ClusterActorRefProvider" } remote { enabled-transports = ["akka.remote.netty.tcp"] log-remote-lifecycle-events = off netty.tcp { hostname = "127.0.0.1" port = 0 } } cluster { seed-nodes = [ "akka.tcp://event-cluster-system@127.0.0.1:2751" "akka.tcp://event-cluster-system@127.0.0.1:2752" "akka.tcp://event-cluster-system@127.0.0.1:2753" ] seed-node-timeout = 60s auto-down-unreachable-after = 10s } }

上述配置中,我们创建的Akka Cluster的名称为event-cluster-system,初始指定了3个seed节点,实际上这3个节点是我们实现的collector角色的节点,用来收集数据。下面,我们依次说明collector、interceptor、processor这3中角色的集群节点的处理逻辑:

  • collector实现

我们实现的collector实现类为EventCollector,它是一个Actor,该实现类继承自ClusterRoledWorker抽象类,具体实现代码如下所示:

package org.shirdrn.scala.akka.cluster import akka.actor._ import akka.cluster.ClusterEvent._ import com.typesafe.config.ConfigFactory import scala.concurrent.ExecutionContext import scala.concurrent.duration._ import scala.concurrent.forkjoin.ForkJoinPool class EventCollector extends ClusterRoledWorker { @volatile var recordCounter : Int = 0 def receive = { case MemberUp(member) => log.info("Member is Up: {}" member.address) case UnreachableMember(member) => log.info("Member detected as Unreachable: {}" member) case MemberRemoved(member previousStatus) => log.info("Member is Removed: {} after {}" member.address previousStatus) case _: MemberEvent => // ignore case Registration => { // watch发送注册消息的interceptor,如果对应的Actor终止了,会发送一个Terminated消息 context watch sender workers = workers : sender log.info("Interceptor registered: " sender) log.info("Registered interceptors: " workers.size) } case Terminated(interceptingActorRef) => // interceptor终止,更新缓存的ActorRef workers = workers.filterNot(_ == interceptingActorRef) case RawNginxRecord(sourceHost line) => { // 构造NginxRecord消息,发送到下游interceptor val eventCode = "eventcode=(\\d )".r.findFirstIn(line).get log.info("Raw message: eventCode=" eventCode " sourceHost=" sourceHost " line=" line) recordCounter = 1 if(workers.size > 0) { // 模拟Roudrobin方式,将日志记录消息发送给下游一组interceptor中的一个 val interceptorIndex = (if(recordCounter < 0) 0 else recordCounter) % workers.size workers(interceptorIndex) ! NginxRecord(sourceHost eventCode line) log.info("Details: interceptorIndex=" interceptorIndex " interceptors=" workers.size) } } } } /** * 用来模拟发送日志记录消息的Actor */ class EventClientActor extends Actor with ActorLogging { implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(new ForkJoinPool()) def receive = { case _=> } val events = Map( "2751" -> List( """10.10.2.72 [21/Aug/2015:18:29:19 0800] "GET /t.gif?installid=0000lAOX&udid=25371384b2eb1a5dc5643e14626ecbd4&sessionid=25371384b2eb1a5dc5643e14626ecbd41440152875362&imsi=460002830862833&operator=1&network=1×tamp=1440152954&action=14&eventcode=300039&page=200002& HTTP/1.0" "-" 204 0 "-" "Dalvik/1.6.0 (Linux; U; Android 4.4.4; R8207 Build/KTU84P)" "121.25.190.146"""" """10.10.2.8 [21/Aug/2015:18:29:19 0800] "GET /t.gif?installid=0000VACO&udid=f6b0520cbc36fda6f63a72d91bf305c0&imsi=460012927613645&operator=2&network=1×tamp=1440152956&action=1840&eventcode=100003&type=1&result=0& HTTP/1.0" "-" 204 0 "-" "Dalvik/1.6.0 (Linux; U; Android 4.4.2; GT-I9500 Build/KOT49H)" "61.175.219.69"""" ) "2752" -> List( """10.10.2.72 [21/Aug/2015:18:29:19 0800] "GET /t.gif?installid=0000gCo4&udid=636d127f4936109a22347b239a0ce73f&sessionid=636d127f4936109a22347b239a0ce73f1440150695096&imsi=460036010038180&operator=3&network=4×tamp=1440152902&action=1566&eventcode=101010&playid=99d5a59f100cb778b64b5234a189e1f4&radioid=1100000048450&audioid=1000001535718&playtime=3& HTTP/1.0" "-" 204 0 "-" "Dalvik/1.6.0 (Linux; U; Android 4.4.4; R8205 Build/KTU84P)" "106.38.128.67"""" """10.10.2.72 [21/Aug/2015:18:29:19 0800] "GET /t.gif?installid=0000kPSC&udid=2ee585cde388ac57c0e81f9a76f5b797&operator=0&network=1×tamp=1440152968&action=6423&eventcode=100003&type=1&result=0& HTTP/1.0" "-" 204 0 "-" "Dalvik/v3.3.85 (Linux; U; Android L; P8 Build/KOT49H)" "202.103.133.112"""" """10.10.2.72 [21/Aug/2015:18:29:19 0800] "GET /t.gif?installid=0000lABW&udid=face1161d739abacca913dcb82576e9d&sessionid=face1161d739abacca913dcb82576e9d1440151582673&operator=0&network=1×tamp=1440152520&action=1911&eventcode=101010&playid=b07c241010f8691284c68186c42ab006&radioid=1100000000762&audioid=1000001751983&playtime=158& HTTP/1.0" "-" 204 0 "-" "Dalvik/1.6.0 (Linux; U; Android 4.1; H5 Build/JZO54K)" "221.232.36.250"""" ) "2753" -> List( """10.10.2.8 [21/Aug/2015:18:29:19 0800] "GET /t.gif?installid=0000krJw&udid=939488333889f18e2b406d2ece8f938a&sessionid=939488333889f18e2b406d2ece8f938a1440137301421&imsi=460028180045362&operator=1&network=1×tamp=1440152947&action=1431&eventcode=300030&playid=e1fd5467085475dc4483d2795f112717&radioid=1100000001123&audioid=1000000094911&playtime=951992& HTTP/1.0" "-" 204 0 "-" "Dalvik/1.6.0 (Linux; U; Android 4.0.4; R813T Build/IMM76D)" "5.45.64.205"""" """10.10.2.72 [21/Aug/2015:18:29:19 0800] "GET /t.gif?installid=0000kcpz&udid=cbc7bbb560914c374cb7a29eef8c2144&sessionid=cbc7bbb560914c374cb7a29eef8c21441440152816008&imsi=460008782944219&operator=1&network=1×tamp=1440152873&action=360&eventcode=200003&page=200003&radioid=1100000046018& HTTP/1.0" "-" 204 0 "-" "Dalvik/v3.3.85 (Linux; U; Android 4.4.2; MX4S Build/KOT49H)" "119.128.106.232"""" """10.10.2.8 [21/Aug/2015:18:29:19 0800] "GET /t.gif?installid=0000juRL&udid=3f9a5ffa69a5cd5f0754d2ba98c0aeb2&imsi=460023744091238&operator=1&network=1×tamp=1440152957&action=78&eventcode=100003&type=1&result=0& HTTP/1.0" "-" 204 0 "-" "Dalvik/v3.3.85 (Linux; U; Android 4.4.3; S?MSUNG. Build/KOT49H)" "223.153.72.78"""" ) ) val ports = Seq("2751" "2752" "2753") val actors = scala.collection.mutable.HashMap[String ActorRef]() ports.foreach { port => // 创建一个Config对象 val config = ConfigFactory.parseString("akka.remote.netty.tcp.port=" port) .withFallback(ConfigFactory.parseString("akka.cluster.roles = [collector]")) .withFallback(ConfigFactory.load()) // 创建一个ActorSystem实例 val system = ActorSystem("event-cluster-system" config) actors(port) = system.actorOf(Props[EventCollector] name = "collectingActor") } Thread.sleep(30000) context.system.scheduler.schedule(0 millis 5000 millis) { // 使用Akka的Scheduler,模拟定时发送日志记录消息 ports.foreach { port => events(port).foreach { line => println("RAW: port=" port " line=" line) actors(port) ! RawNginxRecord("host.me:" port line) } } } } object EventClient extends App { val system = ActorSystem("client") // 创建EventClientActor实例 val clientActorRef = system.actorOf(Props[EventClientActor] name = "clientActor") system.log.info("Client actor started: " clientActorRef) }

上面代码中,EventClientActor并不是属于我们创建的Akka集群event-cluster-system,它是一个位于集群外部的节点,它模拟向各个collector角色的节点发送消息。

  • interceptor实现

与编写collector类似,实现的interceptor的Actor实现类为EventInterceptor,代码如下所示:

package org.shirdrn.scala.akka.cluster import akka.actor._ import akka.cluster.ClusterEvent._ import akka.cluster.Member import akka.cluster.protobuf.msg.ClusterMessages.MemberStatus import com.typesafe.config.ConfigFactory import net.sf.json.JSONObject import org.shirdrn.scala.akka.cluster.utils.DatetimeUtils class EventInterceptor extends ClusterRoledWorker { @volatile var interceptedRecords : Int = 0 val IP_PATTERN = "[^\\s] \\s \\[([^\\]] )\\]. \"(\\d \\.\\d \\.\\d \\.\\d )\"".r val blackIpList = Array( "5.9.116.101" "103.42.176.138" "123.182.148.65" "5.45.64.205" "27.159.226.192" "76.164.228.218" "77.79.178.186" "104.200.31.117" "104.200.31.32" "104.200.31.238" "123.182.129.108" "220.161.98.39" "59.58.152.90" "117.26.221.236" "59.58.150.110" "123.180.229.156" "59.60.123.239" "117.26.222.6" "117.26.220.88" "59.60.124.227" "142.54.161.50" "59.58.148.52" "59.58.150.85" "202.105.90.142" ).toSet log.info("Black IP count: " blackIpList.size) blackIpList.foreach(log.info(_)) def receive = { case MemberUp(member) => log.info("Member is Up: {}" member.address) register(member getCollectorPath) case state: CurrentClusterState => // 如果加入Akka集群的成员节点是Up状态,并且是collector角色,则调用register向collector进行注册 state.members.filter(_.status == MemberStatus.Up) foreach(register(_ getCollectorPath)) case UnreachableMember(member) => log.info("Member detected as Unreachable: {}" member) case MemberRemoved(member previousStatus) => log.info("Member is Removed: {} after {}" member.address previousStatus) case _: MemberEvent => // ignore case Registration => { context watch sender workers = workers : sender log.info("Processor registered: " sender) log.info("Registered processors: " workers.size) } case Terminated(processingActorRef) => workers = workers.filterNot(_ == processingActorRef) case NginxRecord(sourceHost eventCode line) => { val (isIpInBlackList data) = checkRecord(eventCode line) if(!isIpInBlackList) { interceptedRecords = 1 if(workers.size > 0) { val processorIndex = (if (interceptedRecords < 0) 0 else interceptedRecords) % workers.size workers(processorIndex) ! FilteredRecord(sourceHost eventCode line data.getString("eventdate") data.getString("realip")) log.info("Details: processorIndex=" processorIndex " processors=" workers.size) } log.info("Intercepted data: data=" data) } else { log.info("Discarded: " line) } } } def getCollectorPath(member: Member): ActorPath = { RootActorPath(member.address) / "user" / "collectingActor" } /** * 检查collector发送的消息所对应的IP是否在黑名单列表中 */ private def checkRecord(eventCode: String line: String): (Boolean JSONObject) = { val data: JSONObject = new JSONObject() var isIpInBlackList = false IP_PATTERN.findFirstMatchIn(line).foreach { m => val rawDt = m.group(1) val dt = DatetimeUtils.format(rawDt) val realIp = m.group(2) data.put("eventdate" dt) data.put("realip" realIp) data.put("eventcode" eventCode) isIpInBlackList = blackIpList.contains(realIp) } (isIpInBlackList data) } } object EventInterceptor extends App { Seq("2851" "2852").foreach { port => val config = ConfigFactory.parseString("akka.remote.netty.tcp.port=" port) .withFallback(ConfigFactory.parseString("akka.cluster.roles = [interceptor]")) .withFallback(ConfigFactory.load()) val system = ActorSystem("event-cluster-system" config) val processingActor = system.actorOf(Props[EventInterceptor] name = "interceptingActor") system.log.info("Processing Actor: " processingActor) } }

上述代码中,解析出Nginx日志记录中的IP地址,查看其是否在IP黑名单列表中,如果在内名单中则直接丢掉该记录数据。

  • processor实现

EventProcessor的实现代码,如下所示:

package org.shirdrn.scala.akka.cluster import java.util.Properties import akka.actor._ import akka.cluster.ClusterEvent._ import akka.cluster.Member import akka.cluster.protobuf.msg.ClusterMessages.MemberStatus import com.typesafe.config.ConfigFactory import kafka.producer.{KeyedMessage Producer ProducerConfig} import net.sf.json.JSONObject class EventProcessor extends ClusterRoledWorker { val topic = "app_events" val producer = KakfaUtils.createProcuder def receive = { case MemberUp(member) => log.info("Member is Up: {}" member.address) // 将processor注册到上游的collector中 register(member getProcessorPath) case state: CurrentClusterState => state.members.filter(_.status == MemberStatus.Up).foreach(register(_ getProcessorPath)) case UnreachableMember(member) => log.info("Member detected as Unreachable: {}" member) case MemberRemoved(member previousStatus) => log.info("Member is Removed: {} after {}" member.address previousStatus) case _: MemberEvent => // ignore case FilteredRecord(sourceHost eventCode line nginxDate realIp) => { val data = process(eventCode line nginxDate realIp) log.info("Processed: data=" data) // 将解析后的消息一JSON字符串的格式,保存到Kafka中 producer.send(new KeyedMessage[String String](topic sourceHost data.toString)) } } def getProcessorPath(member: Member): ActorPath = { RootActorPath(member.address) / "user" / "interceptingActor" } private def process(eventCode: String line: String eventDate: String realIp: String): JSONObject = { val data: JSONObject = new JSONObject() "[\\?|&]{1}([^=] )=([^&] )&".r.findAllMatchIn(line) foreach { m => val key = m.group(1) val value = m.group(2) data.put(key value) } data.put("eventdate" eventDate) data.put("realip" realIp) data } } object KakfaUtils { // bin/kafka-topics.sh --create -zookeeper zk1:2181 zk2:2181 zk3:2181/data-dept/kafka --replication-factor 2 --partitions 2 --topic app_events val props = new Properties() val config = Map( "metadata.broker.list" -> "hadoop2:9092 hadoop3:9092" "serializer.class" -> "kafka.serializer.StringEncoder" "producer.type" -> "async" ) config.foreach(entry => props.put(entry._1 entry._2)) val producerConfig = new ProducerConfig(props) def createProcuder() : Producer[String String] = { new Producer[String String](producerConfig) } } object EventProcessor extends App { // 启动了5个EventProcessor Seq("2951" "2952" "2953" "2954" "2955") foreach { port => val config = ConfigFactory.parseString("akka.remote.netty.tcp.port=" port) .withFallback(ConfigFactory.parseString("akka.cluster.roles = [processor]")) .withFallback(ConfigFactory.load()) val system = ActorSystem("event-cluster-system" config) val processingActor = system.actorOf(Props[EventProcessor] name = "processingActor") system.log.info("Processing Actor: " processingActor) } }

角色为processor的Actor的实现类为EventProcessor,我们在其伴生对象中创建了5个实例,分别对应不同的端口。解析的Nginx日志记录最后保存到Kafka,示例如下所示:

{"installid":"0000VACO" "imsi":"460012927613645" "network":"1" "action":"1840" "type":"1" "eventdate":"2015-08-21 18:29:19" "realip":"61.175.219.69"} {"installid":"0000kcpz" "sessionid":"cbc7bbb560914c374cb7a29eef8c21441440152816008" "operator":"1" "timestamp":"1440152873" "eventcode":"200003" "radioid":"1100000046018" "eventdate":"2015-08-21 18:29:19" "realip":"119.128.106.232"} {"installid":"0000lAOX" "sessionid":"25371384b2eb1a5dc5643e14626ecbd41440152875362" "operator":"1" "timestamp":"1440152954" "eventcode":"300039" "eventdate":"2015-08-21 18:29:19" "realip":"121.25.190.146"}

(原创 时延军 ,链接:http://shiyanjun.cn)

感谢大家支持,多多转发关注小编不迷路~每天更新好文!!!

猜您喜欢: