【深入淺出 Yarn 架構(gòu)與實現(xiàn)】2-2 Yarn 基礎(chǔ)庫

RPC（Remote Procedure Call）是 Hadoop 服務(wù)通信的關(guān)鍵庫，支撐上層分布式環(huán)境下復(fù)雜的進程間（Inter-Process Communication, IPC）通信邏輯，是分布式系統(tǒng)的基礎(chǔ)。允許運行于一臺計算機上的程序像調(diào)用本地方法一樣，調(diào)用另一臺計算機的子程序。
由于 RPC 服務(wù)整體知識較多，本節(jié)僅針對對 Yarn RPC 進行簡略介紹，詳細內(nèi)容會后續(xù)開專欄介紹。

創(chuàng)新互聯(lián)秉承實現(xiàn)全網(wǎng)價值營銷的理念，以專業(yè)定制企業(yè)官網(wǎng)，成都做網(wǎng)站、成都網(wǎng)站設(shè)計，小程序開發(fā)，網(wǎng)頁設(shè)計制作，手機網(wǎng)站制作，全網(wǎng)整合營銷推廣幫助傳統(tǒng)企業(yè)實現(xiàn)“互聯(lián)網(wǎng)+”轉(zhuǎn)型升級專業(yè)定制企業(yè)官網(wǎng),公司注重人才、技術(shù)和管理，匯聚了一批優(yōu)秀的互聯(lián)網(wǎng)技術(shù)人才,對客戶都以感恩的心態(tài)奉獻自己的專業(yè)和所長。

一、RPC 通信模型介紹

為什么會有 RPC 框架？
在分布式或微服務(wù)情境下，會有大量的服務(wù)間交互，如果用傳統(tǒng)的 HTTP 協(xié)議端口來通信，需要耗費大量時間處理網(wǎng)絡(luò)數(shù)據(jù)交換上，還要考慮編解碼等問題。如下圖所示。

客戶端通過 RPC 框架的動態(tài)代理得到一個代理類實例，稱為 Stub（樁）
客戶端調(diào)用接口方法（實際是 Stub 對應(yīng)的方法），Stub 會構(gòu)造一個請求，包括函數(shù)名和參數(shù)
服務(wù)端收到這個請求后，先將服務(wù)名（函數(shù)）解析出來，查找是否有對應(yīng)的服務(wù)提供者
服務(wù)端找到對應(yīng)的實現(xiàn)類后，會傳入?yún)?shù)調(diào)用
服務(wù)端 RPC 框架得到返回結(jié)果后，再進行封裝返回給客戶端
客戶端的 Stub 收到返回值后，進行解析，返回給調(diào)用者，完成 RPC 調(diào)用。

二、Hadoop RPC 介紹

一）簡介

Hadoop RPC 是 Hadoop 自己實現(xiàn)的一個 RPC 框架，主要有以下幾個特點：

透明性：像調(diào)用本地方法一樣調(diào)用遠程方法。
高性能：Hadoop 各個系統(tǒng)均采用 Master/Slave 結(jié)構(gòu)，Master 是一個 RPC Server 用于處理各個 Slave 節(jié)點發(fā)送的請求，需要有高性能。
可控性：由于 JDK 中的 RPC 框架 RMI 重量級過大，且封裝度太高，不方便控制和修改。因此實現(xiàn)了自己的 RPC 框架，以保證輕量級、高性能、可控性。

框架原理和整體執(zhí)行流程與第一節(jié)介紹的 RPC 框架一致，感興趣可深入源碼進行了解。

二）總體架構(gòu)

Hadoop RPC 架構(gòu)底層依靠 Java 的 nio、反射、動態(tài)代理等功能實現(xiàn)「客戶端 - 服務(wù)器（C/S）」通信模型。
上層封裝供程序調(diào)用的 RPC 接口。

三、案例 demo

下面兩個案例的 demo 已上傳至 github。有幫助的話點個??。
https://github.com/Simon-Ace/hadoop_rpc_demo

一）RPC Writable 案例實現(xiàn)

1、新建一個 maven 工程，添加依賴

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>2.8.5</version>
</dependency>

2、定義 RPC 協(xié)議

public interface BusinessProtocol {
    void mkdir(String path);
    String getName(String name);
    long versionID = 345043000L;
}

3、定義協(xié)議實現(xiàn)

public class BusinessIMPL implements BusinessProtocol {
    @Override
    public void mkdir(String path) {
        System.out.println("成功創(chuàng)建了文件夾 ：" + path);
    }

    @Override
    public String getName(String name) {
        System.out.println("成功打了招呼： hello ：" + name);
        return "bigdata";
    }
}

4、通過 Hadoop RPC 構(gòu)建一個 RPC 服務(wù)端

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;

import java.io.IOException;

public class MyServer {
    public static void main(String[] args) {
        try {
            // 構(gòu)建一個 RPC server 端，提供了一個 BussinessProtocol 協(xié)議的 BusinessIMPL 服務(wù)實現(xiàn)
            RPC.Server server = new RPC.Builder(new Configuration())
                    .setProtocol(BusinessProtocol.class)
                    .setInstance(new BusinessIMPL())
                    .setBindAddress("localhost")
                    .setPort(6789)
                    .build();

            server.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

5、構(gòu)建一個 RPC 客戶端

import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.conf.Configuration;

import java.io.IOException;
import java.net.InetSocketAddress;

public class MyClient {
    public static void main(String[] args) {
        try {
        	// 獲取代理類實例，也就是 Stub
            BusinessProtocol proxy = RPC.getProxy(BusinessProtocol.class, BusinessProtocol.versionID,
                    new InetSocketAddress("localhost", 6789), new Configuration());

            // 通過 Stub 發(fā)送請求，實際使用就像調(diào)用本地方法一樣
            proxy.mkdir("/tmp/ABC");
            String res = proxy.getName("Simon");
            System.out.println("從 RPC 服務(wù)端接收到的返回值：" + res);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

6、測試，先啟動服務(wù)端，再啟動客戶端
服務(wù)端輸出

成功創(chuàng)建了文件夾 ：/tmp/ABC
成功打了招呼： hello ：Simon

客戶端輸出

從 RPC 服務(wù)端接收到的返回值：bigdata

二）RPC Protobuf 案例實現(xiàn)

項目結(jié)構(gòu)如下

對 proto 文件格式不熟悉的同學，參考上一篇文章《2-1 Yarn 基礎(chǔ)庫概述》

MyResourceTrackerMessage.proto 定義數(shù)據(jù)格式

syntax = "proto3";
option java_package = "com.shuofxz.protobuf_rpc.proto";
option java_outer_classname = "MyResourceTrackerMessageProto";
option java_generic_services = true;
option java_generate_equals_and_hash = true;

message MyRegisterNodeManagerRequestProto {
    string hostname = 1;
    int32 cpu = 2;
    int32 memory = 3;
}

message MyRegisterNodeManagerResponseProto {
    string flag = 1;
}

MyResourceTracker.proto 定義 rpc 接口

syntax = "proto3";

import "com/shuofxz/protobuf_rpc/proto/MyResourceTrackerMessage.proto";
option java_package = "com.shuofxz.protobuf_rpc.proto";
option java_outer_classname = "MyResourceTrackerProto";
option java_generic_services = true;
option java_generate_equals_and_hash = true;

service MyResourceTrackerService {
    rpc registerNodeManager(MyRegisterNodeManagerRequestProto) returns (MyRegisterNodeManagerResponseProto);
}

2、對 proto 文件編譯，生成 java 類

# 在項目根目錄執(zhí)行，路徑按照自己的進行修改
protoc -I=src/main/java --java_out=src/main/java src/main/java/com/shuofxz/protobuf_rpc/proto/MyResource.proto

protoc -I=src/main/java --java_out=src/main/java src/main/java/com/shuofxz/protobuf_rpc/proto/MyResourceTracker.proto

3、定義調(diào)用方法接口 MyResourceTracker

import com.shuofxz.protobuf_rpc.proto.MyResourceTrackerMessageProto.MyRegisterNodeManagerResponseProto;
import com.shuofxz.protobuf_rpc.proto.MyResourceTrackerMessageProto.MyRegisterNodeManagerRequestProto;

public interface MyResourceTracker {
    MyRegisterNodeManagerResponseProto registerNodeManager(MyRegisterNodeManagerRequestProto request) throws Exception;
}

4、對調(diào)用方法接口的實現(xiàn)（服務(wù)端）

import com.shuofxz.protobuf_rpc.interf.MyResourceTracker;
import com.shuofxz.protobuf_rpc.proto.MyResourceTrackerMessageProto;

public class MyResourceTrackerImpl implements MyResourceTracker {
    @Override
    public MyResourceTrackerMessageProto.MyRegisterNodeManagerResponseProto registerNodeManager(
            MyResourceTrackerMessageProto.MyRegisterNodeManagerRequestProto request) {

        // 輸出注冊的消息
        String hostname = request.getHostname();
        int cpu = request.getCpu();
        int memory = request.getMemory();
        System.out.println("NodeManager 的注冊消息： hostname = " + hostname + ", cpu = " + cpu + ", memory = " + memory);

        // 省略處理邏輯
        // 構(gòu)建一個響應(yīng)對象，用于返回
        MyResourceTrackerMessageProto.MyRegisterNodeManagerResponseProto.Builder builder =
                MyResourceTrackerMessageProto.MyRegisterNodeManagerResponseProto.newBuilder();
        // 直接返回 True
        builder.setFlag("true");
        MyResourceTrackerMessageProto.MyRegisterNodeManagerResponseProto response = builder.build();
        return response;
    }
}

5、編寫 proto 的協(xié)議接口

import com.shuofxz.protobuf_rpc.proto.MyResourceTrackerProto;
import org.apache.hadoop.ipc.ProtocolInfo;

@ProtocolInfo(protocolName = "com.shuofxz.blablabla", protocolVersion = 1)
public interface MyResourceTrackerPB extends MyResourceTrackerProto.MyResourceTrackerService.BlockingInterface {
}

6、編寫 proto 的協(xié)議接口實現(xiàn)（服務(wù)端）

import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
import com.shuofxz.protobuf_rpc.interf.MyResourceTracker;
import com.shuofxz.protobuf_rpc.proto.MyResourceTrackerMessageProto;
import com.shuofxz.protobuf_rpc.interf.MyResourceTrackerPB;

public class MyResourceTrackerServerSidePB implements MyResourceTrackerPB {
    final private MyResourceTracker server;

    public MyResourceTrackerServerSidePB(MyResourceTracker server) {
        this.server = server;
    }

    @Override
    public MyResourceTrackerMessageProto.MyRegisterNodeManagerResponseProto registerNodeManager(
            RpcController controller, MyResourceTrackerMessageProto.MyRegisterNodeManagerRequestProto request) throws ServiceException {
        try {
            return server.registerNodeManager(request);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
}

7、RPC Server 的實現(xiàn)

import com.shuofxz.protobuf_rpc.interf.MyResourceTrackerPB;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.ProtobufRpcEngine;
import org.apache.hadoop.ipc.RPC;
import com.shuofxz.protobuf_rpc.proto.MyResourceTrackerProto;

import java.io.IOException;

public class ProtobufRpcServer {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();

        RPC.setProtocolEngine(conf, MyResourceTrackerPB.class, ProtobufRpcEngine.class);

        // 構(gòu)建 Rpc Server
        RPC.Server server = new RPC.Builder(conf)
                .setProtocol(MyResourceTrackerPB.class)
                .setInstance(MyResourceTrackerProto.MyResourceTrackerService
                        .newReflectiveBlockingService(new MyResourceTrackerServerSidePB(new MyResourceTrackerImpl())))
                .setBindAddress("localhost")
                .setPort(9998)
                .setNumHandlers(1)
                .setVerbose(true)
                .build();

        // Rpc Server 啟動
        server.start();
    }
}

8、RPC Client 的實現(xiàn)

import com.google.protobuf.ServiceException;
import com.shuofxz.protobuf_rpc.proto.MyResourceTrackerMessageProto;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.ProtobufRpcEngine;
import org.apache.hadoop.ipc.RPC;
import com.shuofxz.protobuf_rpc.interf.MyResourceTrackerPB;

import java.io.IOException;
import java.net.InetSocketAddress;

public class ProtobufRpcClient {
    public static void main(String[] args) throws IOException {
        // 設(shè)置 RPC 引擎為 ProtobufRpcEngine
        Configuration conf = new Configuration();
        String hostname = "localhost";
        int port = 9998;
        RPC.setProtocolEngine(conf, MyResourceTrackerPB.class, ProtobufRpcEngine.class);

        // 獲取代理
        MyResourceTrackerPB protocolProxy = RPC
                .getProxy(MyResourceTrackerPB.class, 1, new InetSocketAddress(hostname, port), conf);

        // 構(gòu)建請求對象
        MyResourceTrackerMessageProto.MyRegisterNodeManagerRequestProto.Builder builder =
                MyResourceTrackerMessageProto.MyRegisterNodeManagerRequestProto.newBuilder();
        MyResourceTrackerMessageProto.MyRegisterNodeManagerRequestProto bigdata02 =
                builder.setHostname("bigdata02").setCpu(64).setMemory(128).build();

        // 發(fā)送 RPC 請求，獲取響應(yīng)
        MyResourceTrackerMessageProto.MyRegisterNodeManagerResponseProto response = null;
        try {
            response = protocolProxy.registerNodeManager(null, bigdata02);
        } catch (ServiceException e) {
            e.printStackTrace();
        }

        // 處理響應(yīng)
        String flag = response.getFlag();
        System.out.println("最終注冊結(jié)果： flag = " + flag);
    }
}

9、測試
先啟動服務(wù)端，在啟動客戶端。

四、總結(jié)

本節(jié)介紹了 Hadoop 底層通信庫 RPC。首先介紹了 RPC 的框架和原理，之后對 Hadoop 自己實現(xiàn)的 RPC 進行了介紹，并給出了兩個 demo 實踐。
強烈建議了解基礎(chǔ)知識后，跟著 demo 實現(xiàn)一個案例出來，可以更好的幫助你理解。
文中 Demo：https://github.com/Simon-Ace/hadoop_rpc_demo

參考文章：
YARN-RPC網(wǎng)絡(luò)通信架構(gòu)設(shè)計
YARN-高并發(fā)RPC源碼實現(xiàn)
Hadoop3.2.1 【 HDFS 】源碼分析 : RPC原理 [八] Client端實現(xiàn)&源碼
Hadoop RPC機制詳解
Hadoop2源碼分析－RPC探索實戰(zhàn)
《Hadoop 技術(shù)內(nèi)幕 - 深入解析 Yarn 結(jié)構(gòu)設(shè)計與實現(xiàn)原理》3.3 節(jié)

文章題目：【深入淺出 Yarn 架構(gòu)與實現(xiàn)】2-2 Yarn 基礎(chǔ)庫
標題來源：http://chinadenli.net/article48/dsoioep.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供、動態(tài)網(wǎng)站、標簽優(yōu)化、App開發(fā)、網(wǎng)站改版、企業(yè)網(wǎng)站制作

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

猜你還喜歡下面的內(nèi)容

欧美一区二区三区老妇人-欧美做爰猛烈大尺度电-99久久夜色精品国产亚洲a-亚洲福利视频一区二区