使用 Hazelcast SQL 查询内存中的数据结构：教程与实践-java教程-PHP中文网

使用 Hazelcast SQL 查询内存中的数据结构：教程与实践

花韻仙語

发布： 2025-07-12 16:22:24

原创

940人浏览过

使用 hazelcast sql 查询内存中的数据结构：教程与实践

本文旨在介绍如何利用 Hazelcast 的 SQL 功能来查询内存中的数据结构，特别是 IMap。我们将探讨 SQL over Map 的概念，并与其他查询方式（如 Java API 和 Predicates API）进行比较。此外，还将介绍如何结合 MapStore 实现数据的持久化，从而解决在内存数据结构上进行复杂查询的需求。

Hazelcast 提供了一种强大的 SQL 查询功能，可以直接对 IMap 等内存数据结构进行查询。这对于需要执行复杂查询，但又希望避免数据库瓶颈的应用场景非常有用。

Hazelcast SQL 的核心概念：SQL over Map

Hazelcast SQL 并非一个独立的数据库，而是 Hazelcast 提供的 SQL 接口的一个组成部分。其中 "SQL over Map" 允许你使用 SQL 语句查询 Hazelcast 的 IMap 数据结构。 IMap 本质上是一个分布式的、并发的 Key-Value 存储，数据完全存储在内存中，提供了非常高的读写性能。

除了 SQL over Map，Hazelcast SQL 还支持对 Kafka、文件系统等数据源进行查询。

查询 IMap 的其他方式

除了 SQL over Map，Hazelcast 还提供了其他查询 IMap 的方式：

Java API： 可以直接使用 IMap 提供的 Java API 进行查询。这种方式比较灵活，但需要编写大量的 Java 代码。
Predicates API： 可以使用 Predicates API 构建查询条件。Predicates API 提供了一种类型安全的方式来构建查询条件，但相对来说，不如 SQL 语句简洁易懂。

例如，使用 Predicates API 查询 delayed_tasks IMap 中 action_required 为 1 且 is_currently_running 为 0 的任务：

import com.hazelcast.query.Predicate;
import com.hazelcast.query.Predicates;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.map.IMap;

import java.util.Collection;
import java.util.Map;

public class PredicateExample {

    public static void main(String[] args) {
        HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
        IMap<Long, Map<String, Object>> delayedTasks = hazelcastInstance.getMap("delayed_tasks");

        // 示例数据
        Map<String, Object> task1 = Map.of("id", 1L, "type", "email", "action", "send", "retries", 3, "action_required", 1, "is_currently_running", 0);
        Map<String, Object> task2 = Map.of("id", 2L, "type", "sms", "action", "send", "retries", 1, "action_required", 0, "is_currently_running", 1);
        Map<String, Object> task3 = Map.of("id", 3L, "type", "email", "action", "resend", "retries", 0, "action_required", 1, "is_currently_running", 0);

        delayedTasks.put(1L, task1);
        delayedTasks.put(2L, task2);
        delayedTasks.put(3L, task3);

        // 构建 Predicate
        Predicate predicate = Predicates.and(
                Predicates.equal("action_required", 1),
                Predicates.equal("is_currently_running", 0)
        );

        // 执行查询
        Collection<Map<String, Object>> results = delayedTasks.values(predicate);

        // 输出结果
        System.out.println("Tasks matching the predicate:");
        for (Map<String, Object> task : results) {
            System.out.println(task);
        }

        hazelcastInstance.shutdown();
    }
}

登录后复制

相比之下，使用 SQL 语句更加直观：

SELECT id, type, action, retries, params FROM delayed_tasks WHERE action_required = 1 and is_currently_running = 0 and (penalty_date IS NULL or penalty_date < NOW()) ORDER BY insert_date ASC LIMIT 1

登录后复制

使用 Hazelcast SQL 查询 IMap

要使用 Hazelcast SQL 查询 IMap，首先需要注册 IMap 的映射。这可以通过定义 MapConfig 来实现：

import com.hazelcast.config.Config;
import com.hazelcast.config.MapConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.sql.SqlResult;
import com.hazelcast.sql.SqlService;

import java.util.List;
import java.util.Map;

public class SqlExample {

    public static void main(String[] args) {
        // 配置 Hazelcast
        Config config = new Config();
        MapConfig mapConfig = new MapConfig("delayed_tasks");
        config.addMapConfig(mapConfig);

        HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance(config);
        IMap<Long, Map<String, Object>> delayedTasks = hazelcastInstance.getMap("delayed_tasks");

        // 示例数据
        Map<String, Object> task1 = Map.of("id", 1L, "type", "email", "action", "send", "retries", 3, "action_required", 1, "is_currently_running", 0);
        Map<String, Object> task2 = Map.of("id", 2L, "type", "sms", "action", "send", "retries", 1, "action_required", 0, "is_currently_running", 1);
        Map<String, Object> task3 = Map.of("id", 3L, "type", "email", "action", "resend", "retries", 0, "action_required", 1, "is_currently_running", 0);

        delayedTasks.put(1L, task1);
        delayedTasks.put(2L, task2);
        delayedTasks.put(3L, task3);

        // 获取 SqlService
        SqlService sqlService = hazelcastInstance.getSqlService();

        // 执行 SQL 查询
        SqlResult result = sqlService.execute(
                "SELECT id, type, action, retries FROM delayed_tasks WHERE action_required = 1 and is_currently_running = 0"
        );

        // 处理查询结果
        result.forEach(row -> {
            System.out.println("ID: " + row.getObject("id") +
                    ", Type: " + row.getObject("type") +
                    ", Action: " + row.getObject("action") +
                    ", Retries: " + row.getObject("retries"));
        });

        // 关闭 Hazelcast 实例
        hazelcastInstance.shutdown();
    }
}

登录后复制

这段代码首先创建了一个 HazelcastInstance，并配置了一个名为 "delayed_tasks" 的 IMap。然后，向 IMap 中添加了一些示例数据。接下来，获取 SqlService 并执行 SQL 查询。最后，遍历查询结果并打印到控制台。

注意事项：