聊聊flink的ActorGateway

序

本文主要研究一下flink的ActorGateway

ActorGateway

flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/instance/ActorGateway.java

public interface ActorGateway extends Serializable {/*** Sends a message asynchronously and returns its response. The response to the message is* returned as a future.** &＃64;param message Message to be sent* &＃64;param timeout Timeout until the Future is completed with an AskTimeoutException* &＃64;return Future which contains the response to the sent message*/Future ask(Object message, FiniteDuration timeout);/*** Sends a message asynchronously without a result.** &＃64;param message Message to be sent*/void tell(Object message);/*** Sends a message asynchronously without a result with sender being the sender.** &＃64;param message Message to be sent* &＃64;param sender Sender of the message*/void tell(Object message, ActorGateway sender);/*** Forwards a message. For the receiver of this message it looks as if sender has sent the* message.** &＃64;param message Message to be sent* &＃64;param sender Sender of the forwarded message*/void forward(Object message, ActorGateway sender);/*** Retries to send asynchronously a message up to numberRetries times. The response to this* message is returned as a future. The message is re-sent if the number of retries is not yet* exceeded and if an exception occurred while sending it.** &＃64;param message Message to be sent* &＃64;param numberRetries Number of times to retry sending the message* &＃64;param timeout Timeout for each sending attempt* &＃64;param executionContext ExecutionContext which is used to send the message multiple times* &＃64;return Future of the response to the sent message*/Future retry(Object message,int numberRetries,FiniteDuration timeout,ExecutionContext executionContext);/*** Returns the path of the remote instance.** &＃64;return Path of the remote instance.*/String path();/*** Returns the underlying actor with which is communicated** &＃64;return ActorRef of the target actor*/ActorRef actor();/*** Returns the leaderSessionID associated with the remote actor or null.** &＃64;return Leader session ID if its associated with this gateway, otherwise null*/UUID leaderSessionID();
}


• ActorGateway接口定义了ask、tell、forward、retry、path、actor、leaderSessionID方法&＃xff1b;它有一个实现类为AkkaActorGateway

AkkaActorGateway

flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/instance/AkkaActorGateway.java

public class AkkaActorGateway implements ActorGateway, Serializable {private static final long serialVersionUID &＃61; 42L;// ActorRef of the remote instanceprivate final ActorRef actor;// Associated leader session ID, which is used for RequiresLeaderSessionID messagesprivate final UUID leaderSessionID;// Decorator for messagesprivate final MessageDecorator decorator;public AkkaActorGateway(ActorRef actor, UUID leaderSessionID) {this.actor &＃61; Preconditions.checkNotNull(actor);this.leaderSessionID &＃61; Preconditions.checkNotNull(leaderSessionID);// we want to wrap RequiresLeaderSessionID messages in a LeaderSessionMessagethis.decorator &＃61; new LeaderSessionMessageDecorator(leaderSessionID);}/*** Sends a message asynchronously and returns its response. The response to the message is* returned as a future.** &＃64;param message Message to be sent* &＃64;param timeout Timeout until the Future is completed with an AskTimeoutException* &＃64;return Future which contains the response to the sent message*/&＃64;Overridepublic Future ask(Object message, FiniteDuration timeout) {Object newMessage &＃61; decorator.decorate(message);return Patterns.ask(actor, newMessage, new Timeout(timeout));}/*** Sends a message asynchronously without a result.** &＃64;param message Message to be sent*/&＃64;Overridepublic void tell(Object message) {Object newMessage &＃61; decorator.decorate(message);actor.tell(newMessage, ActorRef.noSender());}/*** Sends a message asynchronously without a result with sender being the sender.** &＃64;param message Message to be sent* &＃64;param sender Sender of the message*/&＃64;Overridepublic void tell(Object message, ActorGateway sender) {Object newMessage &＃61; decorator.decorate(message);actor.tell(newMessage, sender.actor());}/*** Forwards a message. For the receiver of this message it looks as if sender has sent the* message.** &＃64;param message Message to be sent* &＃64;param sender Sender of the forwarded message*/&＃64;Overridepublic void forward(Object message, ActorGateway sender) {Object newMessage &＃61; decorator.decorate(message);actor.tell(newMessage, sender.actor());}/*** Retries to send asynchronously a message up to numberRetries times. The response to this* message is returned as a future. The message is re-sent if the number of retries is not yet* exceeded and if an exception occurred while sending it.** &＃64;param message Message to be sent* &＃64;param numberRetries Number of times to retry sending the message* &＃64;param timeout Timeout for each sending attempt* &＃64;param executionContext ExecutionContext which is used to send the message multiple times* &＃64;return Future of the response to the sent message*/&＃64;Overridepublic Future retry(Object message,int numberRetries,FiniteDuration timeout,ExecutionContext executionContext) {Object newMessage &＃61; decorator.decorate(message);return AkkaUtils.retry(actor,newMessage,numberRetries,executionContext,timeout);}/*** Returns the ActorPath of the remote instance.** &＃64;return ActorPath of the remote instance.*/&＃64;Overridepublic String path() {return actor.path().toString();}/*** Returns {&＃64;link ActorRef} of the target actor** &＃64;return ActorRef of the target actor*/&＃64;Overridepublic ActorRef actor() {return actor;}&＃64;Overridepublic UUID leaderSessionID() {return leaderSessionID;}&＃64;Overridepublic String toString() {return String.format("AkkaActorGateway(%s, %s)", actor.path(), leaderSessionID);}
}


• AkkaActorGateway实现了ActorGateway接口&＃xff0c;它的构造器要求输入ActorRef及leaderSessionID&＃xff0c;同时基于leaderSessionID创建了LeaderSessionMessageDecorator&＃xff1b;ask、tell、forward、retry方法均首先调用LeaderSessionMessageDecorator.decorate方法包装message参数&＃xff0c;然后再去调用ActorRef的相应方法

MessageDecorator

flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/messages/MessageDecorator.java

public interface MessageDecorator extends java.io.Serializable {/*** Decorates a message** &＃64;param message Message to decorate* &＃64;return Decorated message*/Object decorate(Object message);
}


• MessageDecorator接口定义了decorate方法用于包装message&＃xff0c;它有一个实现类为LeaderSessionMessageDecorator

LeaderSessionMessageDecorator

flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/messages/LeaderSessionMessageDecorator.java

public class LeaderSessionMessageDecorator implements MessageDecorator {private static final long serialVersionUID &＃61; 5359618147408392706L;/** Leader session ID with which the RequiresLeaderSessionID messages will be decorated */private final UUID leaderSessionID;/*** Sets the leader session ID with which the messages will be decorated.** &＃64;param leaderSessionID Leader session ID to be used for decoration*/public LeaderSessionMessageDecorator(UUID leaderSessionID) {this.leaderSessionID &＃61; leaderSessionID;}&＃64;Overridepublic Object decorate(Object message) {if (message instanceof RequiresLeaderSessionID) {return new JobManagerMessages.LeaderSessionMessage(leaderSessionID, message);} else {return message;}}
}


• LeaderSessionMessageDecorator实现了MessageDecorator接口&＃xff0c;其decorate方法判断message是RequiresLeaderSessionID类型的话&＃xff0c;则返回JobManagerMessages.LeaderSessionMessage&＃xff0c;否则返回原始的message

JobManagerMessages.LeaderSessionMessage

flink-1.7.2/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala

object JobManagerMessages {/** Wrapper class for leader session messages. Leader session messages implement the* [[RequiresLeaderSessionID]] interface and have to be wrapped in a [[LeaderSessionMessage]],* which also contains the current leader session ID.** &＃64;param leaderSessionID Current leader session ID* &＃64;param message [[RequiresLeaderSessionID]] message to be wrapped in a [[LeaderSessionMessage]]*/case class LeaderSessionMessage(leaderSessionID: UUID, message: Any)//......
}


• JobManagerMessages.LeaderSessionMessage是一个case class&＃xff0c;它有两个属性&＃xff0c;分别是leaderSessionID及message

小结

• ActorGateway接口定义了ask、tell、forward、retry、path、actor、leaderSessionID方法&＃xff1b;它有一个实现类为AkkaActorGateway
• AkkaActorGateway实现了ActorGateway接口&＃xff0c;它的构造器要求输入ActorRef及leaderSessionID&＃xff0c;同时基于leaderSessionID创建了LeaderSessionMessageDecorator&＃xff1b;ask、tell、forward、retry方法均首先调用LeaderSessionMessageDecorator.decorate方法包装message参数&＃xff0c;然后再去调用ActorRef的相应方法
• MessageDecorator接口定义了decorate方法用于包装message&＃xff0c;它有一个实现类为LeaderSessionMessageDecorator&＃xff1b;LeaderSessionMessageDecorator实现了MessageDecorator接口&＃xff0c;其decorate方法判断message是RequiresLeaderSessionID类型的话&＃xff0c;则返回JobManagerMessages.LeaderSessionMessage&＃xff0c;否则返回原始的message&＃xff1b;JobManagerMessages.LeaderSessionMessage是一个case class&＃xff0c;它有两个属性&＃xff0c;分别是leaderSessionID及message

doc

• ActorGateway