Apache Ignite的分布式计算（Distributed Computing）

这段内容是关于 Apache Ignite 的 分布式计算（Distributed Computing） 功能的详细说明。它解释了如何使用 Ignite 的 API 在集群节点之间分发任务、执行并行计算，并实现容错和负载均衡。下面我将逐段为你通俗易懂地解读，帮助你理解其核心概念和用途。

🧩 一、什么是“分布式计算”？

在 Ignite 中，分布式计算 指的是：

把一个大任务拆成多个小任务，分发到集群中的不同节点上并行执行，最后汇总结果。

这类似于 MapReduce 或 Spark 的工作方式，但更轻量级，适用于内存计算场景。

🔑 二、核心入口：compute 接口

Ignite ignite = Ignition.start();
IgniteCompute compute = ignite.compute();

• ignite.compute() 是进入分布式计算的“大门”。
• 它返回一个 IgniteCompute 对象，你可以用它来：
  • 向某些节点发送任务
  • 并行执行函数
  • 广播消息
  • 异步处理等

🎯 三、指定在哪些节点上运行任务

默认情况下，ignite.compute() 会作用于所有 服务端节点（server nodes）。

但你可以自定义目标节点组：

// 只对远程节点（非本地节点）执行任务
IgniteCompute compute = ignite.compute(ignite.cluster().forRemotes());

Ignite 提供了灵活的 集群分组机制（ClusterGroup），比如你可以选择：

• 所有节点
• 某些 IP 的节点
• 带特定属性的节点
• 客户端节点 or 服务端节点

🛠️ 四、三种可执行的任务类型

Ignite 支持三种任务接口，对应不同的使用场景：

✅ 示例1：执行 Runnable 任务（无返回值）

for (String word : "Hello World".split(" ")) {
    compute.run(() -> System.out.println(word));
}

• 每个单词会被发送到集群中的某个节点上打印。
• 任务只执行一次，在任意一个节点上运行。

📌 用途：适合做日志输出、触发事件、清理资源等不需要结果的操作。

✅ 示例2：执行 Callable 任务（有返回值）

Collection<IgniteCallable<Integer>> calls = new ArrayList<>();
for (String word : "How many characters".split(" "))
    calls.add(word::length);

Collection<Integer> res = ignite.compute().call(calls);
int total = res.stream().mapToInt(Integer::intValue).sum();

• 每个单词的长度计算作为一个 Callable 任务。
• 所有任务并行执行，返回各自的长度。
• 最终收集所有结果，求和。

📌 这就是典型的 并行计算 + 结果聚合 模式。

✅ 示例3：执行 Closure 任务（带参数的函数）

Collection<Integer> res = compute.apply(String::length, Arrays.asList("How many characters".split(" ")));

• 相当于把 "How"、"many" 等作为参数传给 String::length 函数。
• 每个参数在某个节点上执行，返回长度。
• 结果集合包含所有返回值。

📌 类似于 map() 操作，非常适合做数据转换。

🔊 五、广播任务（Broadcast）

compute.broadcast(() -> System.out.println("Hello Node"));

• 将同一个任务发送到 所有关联节点 上执行。
• 适用于通知、预热缓存、初始化配置等操作。

⏱️ 六、异步执行（Async）

所有同步方法都有对应的异步版本：

异步方法返回一个 IgniteFuture<T>，你可以：

IgniteFuture<Collection<Integer>> future = compute.callAsync(calls);

// 注册回调，任务完成后自动触发
future.listen(fut -> {
    int total = fut.get().stream().mapToInt(Integer::intValue).sum();
    System.out.println("Total: " + total);
});

✅ 优势：不阻塞主线程，适合高并发系统。

⏳ 七、设置任务超时

防止某个任务长时间不结束，导致资源浪费：

compute.withTimeout(300_000).run(() -> {
    // 任务最多运行 5 分钟（300,000 毫秒）
});

• 如果超时，任务会被取消。
• 注意：withTimeout() 只对下一个任务生效，每个需要超时的任务都要单独设置。

🧠 八、节点本地状态共享（NodeLocal）

有时你想知道某个任务在一个节点上被执行了多少次，可以用 nodeLocalMap：

ConcurrentMap<String, AtomicLong> nodeLocalMap = ignite.cluster().nodeLocalMap();

AtomicLong counter = nodeLocalMap.get("counter");
if (counter == null) {
    nodeLocalMap.putIfAbsent("counter", new AtomicLong());
    counter = nodeLocalMap.get("counter");
}
counter.incrementAndGet();

• nodeLocalMap 是每个节点独有的“本地变量存储区”。
• 不会跨节点共享，类似 ThreadLocal，但是以“节点”为单位。

📌 用途：统计任务执行次数、缓存临时数据、避免重复初始化等。

💾 九、任务中访问缓存数据（Cache）

计算任务经常需要读取缓存中的数据：

IgniteCache<Long, Person> cache = ignite.cache("person");
Person person = cache.get(1L);

⚠️ 重要提示：数据亲和性（Colocation）

• 如果 key=1 的 Person 数据在 A 节点，而任务却在 B 节点执行，就需要通过网络传输数据，效率低。
• 最佳实践：让任务和数据在同一个节点上执行，这就是“任务与数据共置（Colocation）”。

Ignite 提供了 AffinityCall 和 AffinityRun 方法来实现这一点：

compute.affinityCall("person", 1L, () -> {
    // 这个任务会在存储 key=1 的节点上执行
    Person p = ignite.cache("person").get(1L); // 本地访问，无需网络
    return p.getName();
});

✅ 优势：避免网络开销，提升性能。

📦 十、类加载问题（Peer Class Loading）

当你提交一个任务类（如 MyTask.class），如果集群节点上没有这个类，会报错。

解决办法：

1. 手动部署：把 .class 或 .jar 文件放到每个节点的 classpath 中。
2. 启用对等类加载（Peer Class Loading）：
   • 默认开启（开发环境）
   • 生产环境建议关闭，改用手动部署，更安全。

<property name="peerClassLoadingEnabled" value="true"/>

⚠️ 注意：任务中使用的 Key 和 Value 类型 也必须在所有节点上可用！

✅ 总结：Ignite 分布式计算的核心能力

🎯 典型应用场景

如果你正在开发一个需要高性能并行计算的系统，比如：

• 实时数据分析
• 批量数据处理
• 分布式爬虫
• 风控引擎
• 缓存一致性检查

那么 Ignite 的分布式计算 API 是一个非常强大的工具。

📌 下一步建议：

如果你感兴趣，我可以为你提供：

• 完整的 Maven 项目示例（Java）
• 如何配置 IgniteConfiguration 支持分布式计算
• 如何使用 Spring Boot 集成 Ignite 计算
• 如何调试和监控任务执行情况

欢迎继续提问！