从零实现一个分布式文件系统：HDFS的核心设计

前言在分布式存储中HDFSHadoop Distributed File System是处理海量数据的基石设计用于大文件的存储和处理。今天我们从零实现HDFS的核心功能· 元数据管理NameNode· 数据存储DataNode· 文件分块Block· 副本管理Replication· 心跳机制Heartbeat· 数据块复制与均衡· 文件读写流程---一、HDFS核心原理1. 架构图┌─────────────────────────────────────────────────────────────┐│ Client │└─────────────────────────────────────────────────────────────┘│ │▼ ▼┌─────────────────────────────────────────────────────────────┐│ NameNode ││ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ││ │ 文件元数据 │ │ 块位置 │ │ 操作日志 │ ││ │ 命名空间 │ │ (缓存) │ │ (WAL) │ ││ └─────────────┘ └─────────────┘ └─────────────┘ │└─────────────────────────────────────────────────────────────┘│ │▼ ▼┌─────────────────────────────────────────────────────────────┐│ DataNode集群 ││ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ││ │ DataNode1 │ │ DataNode2 │ │ DataNode3 │ ││ │ Block A │ │ Block A │ │ Block B │ ││ │ Block B │ │ Block C │ │ Block C │ ││ └─────────────┘ └─────────────┘ └─────────────┘ │└─────────────────────────────────────────────────────────────┘2. 核心概念概念 说明NameNode 元数据管理节点单点DataNode 数据存储节点Block 数据块默认128MBReplication 副本数默认3Heartbeat 心跳DataNode→NameNode---二、完整代码实现1. 基础数据结构c#include stdio.h#include stdlib.h#include string.h#include unistd.h#include pthread.h#include time.h#include errno.h#include sys/stat.h#include fcntl.h#include dirent.h#define MAX_FILENAME 256#define MAX_BLOCK_ID 64#define MAX_DATANODES 10#define BLOCK_SIZE (64 * 1024 * 1024) // 64MB#define REPLICATION_FACTOR 3// 数据块typedef struct block {char block_id[MAX_BLOCK_ID];long long size;long long offset;struct block *next;} block_t;// 文件元数据typedef struct hdfs_file {char filename[MAX_FILENAME];block_t *blocks;int block_count;long long file_size;time_t create_time;time_t modify_time;struct hdfs_file *next;} hdfs_file_t;// DataNode信息typedef struct datanode {char node_id[64];char host[32];int port;long long disk_free;long long disk_total;int status; // 0: offline, 1: online, 2: busytime_t last_heartbeat;char blocks[1024][64]; // 存储的块ID列表int block_count;struct datanode *next;} datanode_t;// NameNodetypedef struct hdfs_namenode {hdfs_file_t *files;datanode_t *datanodes;char block_dir[256];int block_id_counter;pthread_mutex_t mutex;int running;int port;pthread_t heartbeat_thread;} hdfs_namenode_t;// DataNodetypedef struct hdfs_datanode {char node_id[64];char data_dir[256];int port;pthread_mutex_t mutex;int running;int port_listen;pthread_t heartbeat_thread;char *blocks;} hdfs_datanode_t;// HDFS客户端typedef struct hdfs_client {char namenode_host[32];int namenode_port;} hdfs_client_t;2. NameNode实现c// 创建NameNodehdfs_namenode_t *namenode_create(int port) {hdfs_namenode_t *nn malloc(sizeof(hdfs_namenode_t));memset(nn, 0, sizeof(hdfs_namenode_t));nn-port port;nn-running 1;nn-block_id_counter 0;pthread_mutex_init(nn-mutex, NULL);mkdir(./blocks, 0755);printf([NameNode] 启动端口: %d\n, port);return nn;}// 注册DataNodeint namenode_register_datanode(hdfs_namenode_t *nn, const char *node_id,const char *host, int port, long long disk_free) {pthread_mutex_lock(nn-mutex);datanode_t *dn nn-datanodes;while (dn) {if (strcmp(dn-node_id, node_id) 0) {dn-status 1;dn-last_heartbeat time(NULL);dn-disk_free disk_free;pthread_mutex_unlock(nn-mutex);return 0;}dn dn-next;}dn malloc(sizeof(datanode_t));strcpy(dn-node_id, node_id);strcpy(dn-host, host);dn-port port;dn-disk_free disk_free;dn-disk_total disk_free;dn-status 1;dn-last_heartbeat time(NULL);dn-block_count 0;dn-next nn-datanodes;nn-datanodes dn;pthread_mutex_unlock(nn-mutex);printf([NameNode] DataNode注册: %s (%s:%d)\n, node_id, host, port);return 0;}// 心跳更新void namenode_heartbeat(hdfs_namenode_t *nn, const char *node_id,long long disk_free, char **blocks, int block_count) {pthread_mutex_lock(nn-mutex);datanode_t *dn nn-datanodes;while (dn) {if (strcmp(dn-node_id, node_id) 0) {dn-last_heartbeat time(NULL);dn-disk_free disk_free;dn-block_count block_count;for (int i 0; i block_count i 1024; i) {strcpy(dn-blocks[i], blocks[i]);}break;}dn dn-next;}pthread_mutex_unlock(nn-mutex);}// 选择DataNode存储块按磁盘空间datanode_t *namenode_select_datanode(hdfs_namenode_t *nn) {pthread_mutex_lock(nn-mutex);datanode_t *selected NULL;long long max_free -1;datanode_t *dn nn-datanodes;while (dn) {if (dn-status 1 dn-disk_free max_free) {max_free dn-disk_free;selected dn;}dn dn-next;}pthread_mutex_unlock(nn-mutex);return selected;}// 分配块IDchar *namenode_allocate_block(hdfs_namenode_t *nn) {pthread_mutex_lock(nn-mutex);nn-block_id_counter;char *block_id malloc(64);snprintf(block_id, 64, blk_%d_%ld, nn-block_id_counter, time(NULL));pthread_mutex_unlock(nn-mutex);return block_id;}3. DataNode实现c// 创建DataNodehdfs_datanode_t *datanode_create(const char *node_id, const char *data_dir, int port) {hdfs_datanode_t *dn malloc(sizeof(hdfs_datanode_t));strcpy(dn-node_id, node_id);strcpy(dn-data_dir, data_dir);dn-port port;dn-port_listen port;dn-running 1;pthread_mutex_init(dn-mutex, NULL);mkdir(data_dir, 0755);printf([DataNode] %s 启动数据目录: %s\n, node_id, data_dir);return dn;}// 存储块int datanode_store_block(hdfs_datanode_t *dn, const char *block_id,const char *data, int data_len) {pthread_mutex_lock(dn-mutex);char filepath[512];snprintf(filepath, sizeof(filepath), %s/%s.dat, dn-data_dir, block_id);FILE *fp fopen(filepath, wb);if (!fp) {pthread_mutex_unlock(dn-mutex);return -1;}fwrite(data, 1, data_len, fp);fclose(fp);pthread_mutex_unlock(dn-mutex);return 0;}// 读取块int datanode_read_block(hdfs_datanode_t *dn, const char *block_id,char *data, int *data_len) {pthread_mutex_lock(dn-mutex);char filepath[512];snprintf(filepath, sizeof(filepath), %s/%s.dat, dn-data_dir, block_id);FILE *fp fopen(filepath, rb);if (!fp) {pthread_mutex_unlock(dn-mutex);return -1;}fseek(fp, 0, SEEK_END);*data_len ftell(fp);fseek(fp, 0, SEEK_SET);fread(data, 1, *data_len, fp);fclose(fp);pthread_mutex_unlock(dn-mutex);return 0;}4. 文件操作c// 创建文件int namenode_create_file(hdfs_namenode_t *nn, const char *filename) {pthread_mutex_lock(nn-mutex);hdfs_file_t *f nn-files;while (f) {if (strcmp(f-filename, filename) 0) {pthread_mutex_unlock(nn-mutex);return -1;}f f-next;}f malloc(sizeof(hdfs_file_t));strcpy(f-filename, filename);f-blocks NULL;f-block_count 0;f-file_size 0;f-create_time time(NULL);f-modify_time time(NULL);f-next nn-files;nn-files f;pthread_mutex_unlock(nn-mutex);printf([NameNode] 创建文件: %s\n, filename);return 0;}// 写入文件分块int hdfs_write(hdfs_namenode_t *nn, const char *filename, const char *data, int data_len) {// 创建文件if (namenode_create_file(nn, filename) 0) {printf(文件已存在: %s\n, filename);return -1;}// 分块写入int offset 0;int block_num 0;int remaining data_len;while (remaining 0) {int chunk_size remaining BLOCK_SIZE ? BLOCK_SIZE : remaining;// 分配块IDchar *block_id namenode_allocate_block(nn);// 选择DataNodedatanode_t *dn namenode_select_datanode(nn);if (!dn) {printf(没有可用的DataNode\n);return -1;}// 写入数据到DataNode模拟// 实际通过RPC传输printf([写入] 块 %s 写入到 %s (大小: %d)\n, block_id, dn-node_id, chunk_size);// 更新元数据pthread_mutex_lock(nn-mutex);hdfs_file_t *f nn-files;while (f) {if (strcmp(f-filename, filename) 0) {block_t *b malloc(sizeof(block_t));strcpy(b-block_id, block_id);b-size chunk_size;b-offset offset;b-next f-blocks;f-blocks b;f-block_count;f-file_size chunk_size;break;}f f-next;}pthread_mutex_unlock(nn-mutex);offset chunk_size;remaining - chunk_size;block_num;free(block_id);}printf([HDFS] 文件 %s 写入完成共 %d 个块\n, filename, block_num);return 0;}// 读取文件int hdfs_read(hdfs_namenode_t *nn, const char *filename, char *data, int *data_len) {pthread_mutex_lock(nn-mutex);hdfs_file_t *f nn-files;while (f) {if (strcmp(f-filename, filename) 0) break;f f-next;}if (!f) {pthread_mutex_unlock(nn-mutex);return -1;}// 读取所有块block_t *b f-blocks;int total_len 0;while (b) {// 查找块所在的DataNode模拟printf([读取] 读取块: %s (大小: %lld)\n, b-block_id, b-size);// 实际需要从DataNode读取数据total_len b-size;b b-next;}*data_len total_len;pthread_mutex_unlock(nn-mutex);return 0;}5. 测试代码cvoid test_hdfs() {printf( HDFS分布式文件系统测试 \n\n);// 创建NameNodehdfs_namenode_t *nn namenode_create(9000);// 创建DataNodehdfs_datanode_t *dn1 datanode_create(dn-1, ./dn1_data, 9001);hdfs_datanode_t *dn2 datanode_create(dn-2, ./dn2_data, 9002);hdfs_datanode_t *dn3 datanode_create(dn-3, ./dn3_data, 9003);// 注册DataNodenamenode_register_datanode(nn, dn-1, 127.0.0.1, 9001, 1024*1024*1024);namenode_register_datanode(nn, dn-2, 127.0.0.1, 9002, 1024*1024*1024);namenode_register_datanode(nn, dn-3, 127.0.0.1, 9003, 1024*1024*1024);// 写入文件char test_data[1024 * 1024]; // 1MB数据for (int i 0; i 1024*1024; i) {test_data[i] A (i % 26);}printf(写入文件 /user/test.txt (1MB)...\n);hdfs_write(nn, /user/test.txt, test_data, 1024*1024);// 读取文件printf(\n读取文件 /user/test.txt...\n);char read_data[1024*1024];int read_len;hdfs_read(nn, /user/test.txt, read_data, read_len);printf(读取到 %d 字节\n, read_len);// 文件列表printf(\n文件列表:\n);hdfs_file_t *f nn-files;while (f) {printf( %s (大小: %lld, 块数: %d)\n,f-filename, f-file_size, f-block_count);f f-next;}printf(\nDataNode状态:\n);datanode_t *dn nn-datanodes;while (dn) {printf( %s: 状态%d, 块数%d\n,dn-node_id, dn-status, dn-block_count);dn dn-next;}free(nn);free(dn1);free(dn2);free(dn3);}int main() {test_hdfs();return 0;}---三、编译和运行bashgcc -o hdfs hdfs.c -lpthread./hdfs---四、HDFS vs 本实现特性 本实现 HDFSNameNode ✅ ✅DataNode ✅ ✅块存储 ✅ 64MB ✅ 128MB副本管理 ❌ ✅ 默认3心跳机制 ✅ ✅块均衡 ❌ ✅高可用 ❌ ✅ (HA)纠删码 ❌ ✅---五、总结通过这篇文章你学会了· HDFS的核心架构NameNode DataNode· 文件分块与元数据管理· DataNode注册与心跳· 文件写入流程分块存储· 文件读取流程块聚合HDFS是分布式存储的经典实现。掌握它你就理解了海量数据存储系统的核心设计。下一篇预告《从零实现一个分布式计算MapReduce的核心设计》---评论区分享一下你对HDFS的理解