Hadoop 實戰二：HDFS Java API 操作與 RPC 遠程調用機制深度解析

Hadoop 實戰二：HDFS Java API 操作與 RPC 遠程調用機制深度解析

基於 Hadoop 偽分布式環境，本文實現 HDFS 核心文件操作 (上傳/下載/列表/移動/目錄管理) 完整 Java 客戶端代碼，並深度解析 Hadoop RPC 遠程調用底層原理、動態代理生成流程、FileSystem 獲取源碼鏈路，為理解分佈式文件系統客戶端交互打下堅實基礎。

一、HDFS Java API 完整操作類

package cn.itcast.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.commons.compress.utils.IOUtils;
import org.junit.Before;
import org.junit.Test;

import java.io.*;
import java.net.URI;

public class HdfsClient {
    private FileSystem fs;
    private static final String HDFS_URI = "hdfs://zpfbuaa:9000/";

    @Before
    public void init() throws IOException {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", HDFS_URI);
        // 可選：Kerberos 認證、代理用戶等
        fs = FileSystem.get(URI.create(HDFS_URI), conf);
    }

    /** 1. 上傳文件 (流式複製，大文件友好) */
    @Test
    public void upload() throws IOException {
        try (FSDataOutputStream os = fs.create(new Path("/user/hadoop/my.jar"));
             FileInputStream is = new FileInputStream("/home/hadoop/download/my.jar")) {
            IOUtils.copy(is, os); // Commons IO 工具類
        }
    }

    /** 2. 便捷上傳 API (內部同流式複製) */
    @Test
    public void upload2() throws IOException {
        fs.copyFromLocalFile(new Path("/home/hadoop/download/my.jar"), 
                             new Path("/aaa/bbb/ccc/my3.jar")); // 父目錄需存在或自動創建
    }

    /** 3. 下載文件 */
    @Test
    public void download() throws IOException {
        try (FSDataInputStream is = fs.open(new Path("/jdk-7u65-linux-i586.tar.gz"));
             FileOutputStream os = new FileOutputStream("/home/hadoop/download/jdk.tar.gz")) {
            IOUtils.copy(is, os);
        }
    }

    /** 4. 列出文件/目錄 (遞歸) */
    @Test
    public void listFiles() throws IOException {
        System.out.println("=== listFiles (遞歸, 含文件元數據) ===");
        RemoteIterator<LocatedFileStatus> it = fs.listFiles(new Path("/"), true);
        while (it.hasNext()) {
            LocatedFileStatus f = it.next();
            System.out.printf("文件: %s, 大小: %d, 塊: %s%n", 
                f.getPath(), f.getLen(), Arrays.toString(f.getBlockLocations()));
        }

        System.out.println("\n=== listStatus (單級, 區分文件/目錄) ===");
        FileStatus[] statuses = fs.listStatus(new Path("/"));
        for (FileStatus s : statuses) {
            System.out.println(s.getPath().getName() + (s.isDirectory() ? " [DIR]" : " [FILE]"));
        }
    }

    /** 5. 創建多級目錄 */
    @Test
    public void mkdirs() throws IOException {
        fs.mkdirs(new Path("/aaa/bbb/ccc/ddd"));
    }

    /** 6. 刪除文件/目錄 (遞歸刪除) */
    @Test
    public void delete() throws IOException {
        // 第二參數 true: 遞歸刪除非空目錄
        fs.delete(new Path("/aaa/bbb"), true);
    }

    /** 7. 移動/重命名 (同文件系統內即時完成) */
    @Test
    public void rename() throws IOException {
        fs.rename(new Path("/old/path"), new Path("/new/path"));
    }

    /** 8. 讀取文件內容 (小文件) */
    @Test
    public void cat() throws IOException {
        try (FSDataInputStream in = fs.open(new Path("/user/hadoop/wordcount/input.txt"))) {
            BufferedReader br = new BufferedReader(new InputStreamReader(in));
            String line;
            while ((line = br.readLine()) != null) System.out.println(line);
        }
    }
}

二、Hadoop RPC 遠程調用機制深度解析

1. RPC 架構角色

2. 核心代碼示例

共享接口 (含版本號)

public interface LoginServiceInterface {
    long versionID = 1L; // 版本號，客戶端/服務端必須一致
    String login(String username, String password);
}

服務端實現與啟動

public class LoginServiceImpl implements LoginServiceInterface {
    @Override
    public String login(String username, String password) {
        return username + " logged in successfully!";
    }
}

public class RpcServer {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        RPC.Builder builder = new RPC.Builder(conf);
        builder.setBindAddress("zpfbuaa")      // 綁定 IP
               .setPort(10000)                 // 綁定端口
               .setProtocol(LoginServiceInterface.class) // 協議接口
               .setInstance(new LoginServiceImpl());     // 實現實例
        
        RPC.Server server = builder.build();
        server.start(); // 非阻塞啟動
        System.out.println("RPC Server started on port 10000");
    }
}

客戶端調用

public class LoginController {
    public static void main(String[] args) throws IOException {
        // 獲取動態代理
        LoginServiceInterface proxy = RPC.getProxy(
            LoginServiceInterface.class, 
            LoginServiceInterface.versionID, // 版本號
            new InetSocketAddress("zpfbuaa", 10000),
            new Configuration()
        );
        
        // 像調用本地方法一樣調用遠程服務
        String result = proxy.login("zpfbuaa", "123456789");
        System.out.println("Server response: " + result);
    }
}

3. RPC 調用完整流程 (9 步)

1. 客戶端調用 proxy.login(...)。
2. 動態代理 (InvocationHandler) 攔截，構建 Invocation 對象 (方法名、參數、版本號)。
3. 客戶端 Connection 發送序列化後的 Invocation 到服務端 (TCP, 默認 Hadoop Writable 序列化)。
4. 服務端 Listener 線程接收請求，交給 Handler 線程池處理。
5. Handler 反序列化 Invocation，根據接口類找到實現實例。
6. 服務端動態代理調用實際實現 impl.login(...)。
7. 實現類執行邏輯，返回結果。
8. 服務端序列化結果，發送回客戶端。
9. 客戶端反序列化結果，返回給調用者。

FileSystem.get(Configuration)
  → FileSystem.get(URI, Configuration)
    → FileSystem.createFileSystem(URI, Configuration)  // 反射實例化
      → DistributedFileSystem 初始化
        → initialize(URI, Configuration)
          → DFSClient 初始化 (核心客戶端)
            → NameNode 連接建立 (RPC Client)
            → LeaseRenewer 租約續期線程啟動
            → 心跳/緩存初始化