import java.util.*;
import java.util.concurrent.*;// 核心接口定义
interface StreamOperator {void open();void processElement(Object element);void close();
}interface SourceFunction extends StreamOperator {void run(SourceContext ctx);
}interface SinkFunction extends StreamOperator {void invoke(Object value);
}// 运行时组件
class JobGraph {private List<StreamOperator> operators = new ArrayList<>();public void addOperator(StreamOperator operator) {operators.add(operator);}public List<StreamOperator> getOperators() {return operators;}
}class ExecutionGraph {private List<ExecutionVertex> vertices = new ArrayList<>();public void addVertex(ExecutionVertex vertex) {vertices.add(vertex);}public List<ExecutionVertex> getVertices() {return vertices;}
}class ExecutionVertex {private StreamOperator operator;private int parallelism;public ExecutionVertex(StreamOperator operator, int parallelism) {this.operator = operator;this.parallelism = parallelism;}public StreamOperator getOperator() {return operator;}
}// 主控节点
class JobManager {private ResourceManager resourceManager = new ResourceManager();private Map<String, JobMaster> runningJobs = new ConcurrentHashMap<>();public String submitJob(JobGraph jobGraph) {String jobId = UUID.randomUUID().toString();JobMaster jobMaster = new JobMaster(jobId, jobGraph);runningJobs.put(jobId, jobMaster);jobMaster.start(resourceManager);return jobId;}
}class JobMaster {private String jobId;private JobGraph jobGraph;private CheckpointCoordinator checkpointCoordinator;public JobMaster(String jobId, JobGraph jobGraph) {this.jobId = jobId;this.jobGraph = jobGraph;this.checkpointCoordinator = new CheckpointCoordinator();}public void start(ResourceManager resourceManager) {// 构建执行图ExecutionGraph executionGraph = buildExecutionGraph(jobGraph);// 申请资源List<TaskSlot> slots = resourceManager.allocateResources(executionGraph);// 部署任务deployTasks(executionGraph, slots);// 启动检查点协调器checkpointCoordinator.start(jobId, executionGraph);}private ExecutionGraph buildExecutionGraph(JobGraph jobGraph) {ExecutionGraph executionGraph = new ExecutionGraph();for (StreamOperator operator : jobGraph.getOperators()) {executionGraph.addVertex(new ExecutionVertex(operator, 2)); // 默认并行度2}return executionGraph;}private void deployTasks(ExecutionGraph executionGraph, List<TaskSlot> slots) {int slotIndex = 0;for (ExecutionVertex vertex : executionGraph.getVertices()) {for (int i = 0; i < vertex.getParallelism(); i++) {Task task = new Task(vertex.getOperator());slots.get(slotIndex++ % slots.size()).deployTask(task);}}}
}// 资源管理
class ResourceManager {private List<TaskManager> taskManagers = new ArrayList<>();public ResourceManager() {// 初始化3个TaskManagerfor (int i = 0; i < 3; i++) {taskManagers.add(new TaskManager(i));}}public List<TaskSlot> allocateResources(ExecutionGraph executionGraph) {List<TaskSlot> slots = new ArrayList<>();for (TaskManager tm : taskManagers) {slots.addAll(tm.getAvailableSlots());}return slots.subList(0, Math.min(slots.size(), executionGraph.getVertices().size()));}
}// 工作节点
class TaskManager {private int id;private List<TaskSlot> slots = new ArrayList<>();public TaskManager(int id) {this.id = id;// 每个TaskManager有2个slotslots.add(new TaskSlot(id + "-1"));slots.add(new TaskSlot(id + "-2"));}public List<TaskSlot> getAvailableSlots() {return new ArrayList<>(slots);}
}class TaskSlot {private String id;private Task runningTask;public TaskSlot(String id) {this.id = id;}public void deployTask(Task task) {this.runningTask = task;task.start();}
}// 任务执行
class Task implements Runnable {private StreamOperator operator;private Thread executionThread;public Task(StreamOperator operator) {this.operator = operator;}public void start() {executionThread = new Thread(this);executionThread.start();}@Overridepublic void run() {operator.open();// 模拟数据处理循环while (true) {Object element = fetchNextElement(); // 从上游获取数据if (element != null) {operator.processElement(element);}}}private Object fetchNextElement() {// 实际从网络或本地队列获取数据return Math.random() > 0.5 ? new Object() : null;}
}// 容错机制
class CheckpointCoordinator {public void start(String jobId, ExecutionGraph executionGraph) {ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);scheduler.scheduleAtFixedRate(() -> {triggerCheckpoint(jobId, executionGraph);}, 0, 10, TimeUnit.SECONDS); // 每10秒触发检查点}private void triggerCheckpoint(String jobId, ExecutionGraph executionGraph) {System.out.println("Triggering checkpoint for job: " + jobId);// 1. 通知所有任务开始检查点for (ExecutionVertex vertex : executionGraph.getVertices()) {// 实际实现中会通过RPC通知TaskManager}// 2. 等待所有任务确认// 3. 持久化检查点元数据}
}// 示例应用
public class SimpleFlinkDemo {public static void main(String[] args) {// 1. 创建作业图JobGraph jobGraph = new JobGraph();// 创建数据源SourceFunction source = new SourceFunction() {@Override public void open() {}@Override public void close() {}@Overridepublic void run(SourceContext ctx) {// 实际产生数据流}@Overridepublic void processElement(Object element) {// 源操作符不需要处理元素}};// 创建处理算子StreamOperator mapper = new StreamOperator() {@Override public void open() {}@Override public void close() {}@Overridepublic void processElement(Object element) {System.out.println("Processing: " + element);// 实际处理逻辑}};// 创建输出算子SinkFunction sink = new SinkFunction() {@Override public void open() {}@Override public void close() {}@Overridepublic void invoke(Object value) {System.out.println("Output: " + value);}@Overridepublic void processElement(Object element) {invoke(element);}};// 构建作业图jobGraph.addOperator(source);jobGraph.addOperator(mapper);jobGraph.addOperator(sink);// 2. 提交作业JobManager jobManager = new JobManager();String jobId = jobManager.submitJob(jobGraph);System.out.println("Job submitted with ID: " + jobId);// 保持主线程运行try {Thread.sleep(60000);} catch (InterruptedException e) {e.printStackTrace();}}
}

1.创建作业图list：source数据源，mapper处理算子，sink输出算子提交

2.加入jobmanager

3.jobmaster 添加一个作业 id：job，里main含有job图

4.生成执行图，里面装的是执行ExecutionVertex

5.给执行图分配slot

6.部署task

执行检查