从零实现一个分布式文件系统：HDFS的核心设计
前言
在分布式存储中，HDFS（Hadoop 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 数据块（默认128MB）
Replication 副本数（默认3）
Heartbeat 心跳（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: busy
time_t last_heartbeat;
char blocks[1024][64]; // 存储的块ID列表
int block_count;
struct datanode *next;
} datanode_t;
// NameNode
typedef 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;
// DataNode
typedef 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
// 创建NameNode
hdfs_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;
}
// 注册DataNode
int 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;
}
// 分配块ID
char *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
// 创建DataNode
hdfs_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;
// 分配块ID
char *block_id = namenode_allocate_block(nn);
// 选择DataNode
datanode_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. 测试代码
```c
void test_hdfs() {
printf("=== HDFS分布式文件系统测试 ===\n\n");
// 创建NameNode
hdfs_namenode_t *nn = namenode_create(9000);
// 创建DataNode
hdfs_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);
// 注册DataNode
namenode_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;
}
```
---
三、编译和运行
```bash
gcc -o hdfs hdfs.c -lpthread
./hdfs
```
---
四、HDFS vs 本实现
特性 本实现 HDFS
NameNode ✅ ✅
DataNode ✅ ✅
块存储 ✅ 64MB ✅ 128MB
副本管理 ❌ ✅ 默认3
心跳机制 ✅ ✅
块均衡 ❌ ✅
高可用 ❌ ✅ (HA)
纠删码 ❌ ✅
---
五、总结
通过这篇文章，你学会了：
· HDFS的核心架构（NameNode + DataNode）
· 文件分块与元数据管理
· DataNode注册与心跳
· 文件写入流程（分块存储）
· 文件读取流程（块聚合）
HDFS是分布式存储的经典实现。掌握它，你就理解了海量数据存储系统的核心设计。
下一篇预告：《从零实现一个分布式计算：MapReduce的核心设计》
---
评论区分享一下你对HDFS的理解～