本篇內(nèi)容介紹了“Flink的rpc組件有哪些”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠?qū)W有所成！

新洲網(wǎng)站建設公司創(chuàng)新互聯(lián)建站,新洲網(wǎng)站設計制作，有大型網(wǎng)站制作公司豐富經(jīng)驗。已為新洲數(shù)千家提供企業(yè)網(wǎng)站建設服務。企業(yè)網(wǎng)站搭建\成都外貿(mào)網(wǎng)站建設公司要多少錢，請找那個售后服務好的新洲做網(wǎng)站的公司定做！

Flink采用akka來實現(xiàn)rpc服務。其中有這幾個重要組件：RpcServer、RpcService、AkkaRpcActor、RpcEndpoint。

這幾個組件作用如下：

（1）RpcEndpoint

提供具體rpc服務。主要實現(xiàn)有 ResourceManager 和 TaskExecutor，

①YarnResourceManager為AM容器中啟動的服務，持有ResourceManager和NodeManager的客戶端

②TaskExecutor為NM容器中啟動taskmanager的類

（2）AkkaRpcService

提供rpc的服務類。該類內(nèi)部持有ActorSystem實例和Supervisor實例。Supervisor中含有SupervisorActor實例，SupervisorActor用于創(chuàng)建其他Actor，可以理解為根Actor。RpcEndpoint在構造時，通過AkkaRpcService的startServer()方法，獲取RpcServer實例。

	public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {
		checkNotNull(rpcEndpoint, "rpc endpoint");

		final SupervisorActor.ActorRegistration actorRegistration = registerAkkaRpcActor(rpcEndpoint);
		final ActorRef actorRef = actorRegistration.getActorRef();
		final CompletableFuture<Void> actorTerminationFuture = actorRegistration.getTerminationFuture();

		LOG.info("Starting RPC endpoint for {} at {} .", rpcEndpoint.getClass().getName(), actorRef.path());

		final String akkaAddress = AkkaUtils.getAkkaURL(actorSystem, actorRef);
		final String hostname;
		Option<String> host = actorRef.path().address().host();
		if (host.isEmpty()) {
			hostname = "localhost";
		} else {
			hostname = host.get();
		}

		Set<Class<?>> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass()));

		implementedRpcGateways.add(RpcServer.class);
		implementedRpcGateways.add(AkkaBasedEndpoint.class);

		final InvocationHandler akkaInvocationHandler;

		if (rpcEndpoint instanceof FencedRpcEndpoint) {
			// a FencedRpcEndpoint needs a FencedAkkaInvocationHandler
			akkaInvocationHandler = new FencedAkkaInvocationHandler<>(
				akkaAddress,
				hostname,
				actorRef,
				configuration.getTimeout(),
				configuration.getMaximumFramesize(),
				actorTerminationFuture,
				((FencedRpcEndpoint<?>) rpcEndpoint)::getFencingToken,
				captureAskCallstacks);

			implementedRpcGateways.add(FencedMainThreadExecutable.class);
		} else {
			akkaInvocationHandler = new AkkaInvocationHandler(
				akkaAddress,
				hostname,
				actorRef,
				configuration.getTimeout(),
				configuration.getMaximumFramesize(),
				actorTerminationFuture,
				captureAskCallstacks);
		}

		// Rather than using the System ClassLoader directly, we derive the ClassLoader
		// from this class . That works better in cases where Flink runs embedded and all Flink
		// code is loaded dynamically (for example from an OSGI bundle) through a custom ClassLoader
		ClassLoader classLoader = getClass().getClassLoader();

		@SuppressWarnings("unchecked")
		RpcServer server = (RpcServer) Proxy.newProxyInstance(
			classLoader,
			implementedRpcGateways.toArray(new Class<?>[implementedRpcGateways.size()]),
			akkaInvocationHandler);

		return server;
	}

先創(chuàng)建RpcEndpoint對應的ActorRef，然后創(chuàng)建RpcServer的代理類AkkaInvocationHandler或FencedAkkaInvocationHandler，并將ActorRef實例賦給其成員屬性 rpcEndpoint:ActorRef。這里的ActorRef即為AkkaRpcActor或FencedAkkaRpcActor實例

（3）RpcServer

用來啟動rpc服務，通常不直接調(diào)用，而是調(diào)用其動態(tài)代理類AkkaInvocationHandler或FencedAkkaInvocationHandler的start()方法

（4）AkkaInvocationHandler或FencedAkkaInvocationHandler

RpcServer的動態(tài)代理類。start()方法用來啟動服務：

	public void start() {
		rpcEndpoint.tell(ControlMessages.START, ActorRef.noSender());
	}

這里向rpcEndpoint，即AkkaRpcActor或FencedAkkaRpcActor實例發(fā)送一條ControlMessages.START消息

（5）AkkaRpcActor

響應rpc消息的actor。其createReceive()：

	public Receive createReceive() {
		return ReceiveBuilder.create()
			.match(RemoteHandshakeMessage.class, this::handleHandshakeMessage)
			.match(ControlMessages.class, this::handleControlMessage)
			.matchAny(this::handleMessage)
			.build();
	}

當消息為ControlMessages.START，調(diào)用StoppedState 的start()方法

		public State start(AkkaRpcActor<?> akkaRpcActor) {
			akkaRpcActor.mainThreadValidator.enterMainThread();

			try {
				akkaRpcActor.rpcEndpoint.internalCallOnStart();
			} catch (Throwable throwable) {
				akkaRpcActor.stop(
					RpcEndpointTerminationResult.failure(
						new AkkaRpcException(
							String.format("Could not start RpcEndpoint %s.", akkaRpcActor.rpcEndpoint.getEndpointId()),
							throwable)));
			} finally {
				akkaRpcActor.mainThreadValidator.exitMainThread();
			}

			return StartedState.STARTED;
		}

在start()方法中調(diào)用具體提供服務的RpcEndpoint實現(xiàn)類internalCallOnStart()方法來啟動服務。internalCallOnStart()方法中會調(diào)用onStart()方法。

“Flink的rpc組件有哪些”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關的知識可以關注創(chuàng)新互聯(lián)網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實用文章！

網(wǎng)站欄目：Flink的rpc組件有哪些
轉載來于：http://www.chinadenli.net/article10/ighpdo.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供動態(tài)網(wǎng)站、、網(wǎng)站收錄、響應式網(wǎng)站、網(wǎng)站改版、網(wǎng)站導航

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉載內(nèi)容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)