Muotai-SPG 持久层实施计划

文档说明

本文档为muotai-spg项目的持久层（Mapper层）实施计划，目标是实现用户通过上传Schema文件建立知识模型，并将数据持久化到MySQL和Neo4j数据库中。

编制日期: 2025-12-15 版本: v1.0 状态: 待实施

一、项目背景与目标

1.1 项目背景

muotai-spg项目已完成以下模块的迁移和转换：

✅ reasoner模块（259个Java文件 + 64个Scala转Java文件）
✅ cloudext模块（186个文件）
✅ catalog模块（10个文件）
✅ builder/model模块（67个文件）
✅ common、core、server等基础模块（约400个文件）

总计: 约986个Java文件，核心功能已本地化

1.2 实施目标

Schema管理: 用户可以上传Schema文件，定义知识图谱的结构（实体类型、关系类型、属性等）
数据持久化: 将Schema信息持久化到MySQL数据库
图存储: 将知识图谱数据存储到Neo4j图数据库
版本管理: 支持Schema的多版本管理和发布
项目管理: 支持多项目隔离和权限管理

二、数据库架构分析

2.1 MySQL数据库设计

根据OpenSPG项目的openspg-initdb.sql分析，MySQL数据库承担以下职责：

2.1.1 核心业务表（17张表）

A. 项目与租户管理（3张表）

表名	说明	核心字段
`kg_project_info`	图谱项目信息表	id, name, description, namespace, config, visibility
`kg_biz_domain`	业务域表	id, name, description, global_config
`kg_sys_lock`	分布式锁表	id, method_name, method_value

业务说明:

一个业务域（tenant）可包含多个项目（project）
项目是Schema和数据的隔离单元
支持公有/私有项目可见性控制

B. 本体Schema管理（6张表）

表名	说明	核心字段
`kg_ontology_entity`	本体类型表	id, original_id, name, name_zh, entity_category, version, config
`kg_ontology_entity_parent`	继承关系表	id, entity_id, parent_id, path, deep_inherit
`kg_ontology_entity_property_range`	三元组表(S-P-O)	id, domain_id, property_name, range_id, property_category
`kg_ontology_property_constraint`	属性约束表	id, name, is_require, max_value, min_value, is_enum
`kg_ontology_semantic`	语义关联表	id, resource_id, semantic_type, original_resource_id
`kg_semantic_rule`	语义规则表	id, name, expression, version_id, status

Schema模型说明:

Entity (实体类型)
  ├── Properties (属性)
  │   ├── Basic Properties (基础属性: Text, Integer, Float等)
  │   └── Advanced Properties (高级属性: 对象/关系)
  ├── Parent Entities (继承关系)
  └── Semantic Rules (语义规则)

实体类型分类:

BASIC: 基本类型（Text, Integer, Float, Date等）
STANDARD: 标准类型（Email, Phone, ID Card等）
ADVANCED: 高级类型（自定义实体、概念等）

C. 项目关联（3张表）

表名	说明	核心字段
`kg_project_entity`	项目-实体关联表	id, project_id, entity_id, version, referenced
`kg_ontology_release`	Schema发布版本表	id, project_id, version, schema_view, status
`kg_ontology_ext`	Schema扩展属性表	id, resource_id, resource_type, ext_type, config

版本管理说明:

支持Schema多版本管理
版本状态: ONLINE、LATEST、EFFICIENT、HISTORY、DISCARDED
schema_view字段存储完整的Schema快照

D. 权限与配置（5张表）

表名	说明	核心字段
`kg_user`	用户表	id, user_no, token, real_name, email
`kg_role`	角色表	id, role_name, permission_detail
`kg_resource_permission`	资源权限表	id, user_no, resource_id, role_id
`kg_config`	配置表	id, project_id, config_name, config, version
`kg_ref`	关联表	id, ref_id, ref_type, refed_id, refed_type

E. 模型与应用（6张表）

表名	说明	核心字段
`kg_model_provider`	模型供应商表	id, name, provider, model_type, params
`kg_provider_param`	供应商参数表	id, provider, model_type, params
`kg_model_detail`	模型详情表	id, provider, type, name, params
`kg_user_model`	用户模型表	id, instance_id, provider, config
`kg_app`	应用表	id, name, description, logo, config
`kg_feedback`	反馈表	id, module_type, one_category, reaction_type
`kg_statistics`	统计表	id, resource_tag, statistics_type, num

2.1.2 表关系图

kg_biz_domain (业务域)
    ↓ 1:N
kg_project_info (项目)
    ↓ 1:N
kg_project_entity (项目-实体关联)
    ↓ N:1
kg_ontology_entity (实体类型)
    ↓ 1:N
kg_ontology_entity_property_range (属性-SPO三元组)
    ↓ N:1
kg_ontology_property_constraint (属性约束)

kg_ontology_entity
    ↓ 1:N
kg_ontology_entity_parent (继承关系)

kg_project_info
    ↓ 1:N
kg_ontology_release (发布版本)

2.2 Neo4j图数据库设计

2.2.1 图模型结构

Neo4j存储实际的知识图谱数据（而非Schema），结构如下：

A. 节点（Node）标签设计

cypher

// 动态标签，根据Schema定义
(:EntityType {id, name, properties...})
(:ConceptType {id, name, properties...})
(:EventType {id, name, properties...})

// 元数据标签
(:__META__ {type, version, project_id})

节点属性:

id: 实体唯一标识（必须）
name: 实体名称
其他属性根据Schema中定义的properties动态添加

B. 关系（Relationship）类型设计

cypher

// 动态关系类型，根据Schema定义
-[:RELATION_TYPE {properties...}]->

// 示例
-[:BELONGS_TO {since: '2024', weight: 0.9}]->
-[:LOCATED_IN {confidence: 0.95}]->

关系属性:

根据Schema中定义的边属性动态设置
支持双向关系（direction: BOTH）

C. 索引设计

cypher

// 创建节点ID索引（必须）
CREATE CONSTRAINT entity_id_unique IF NOT EXISTS
FOR (n:Entity) REQUIRE n.id IS UNIQUE;

// 创建属性索引（根据Schema的index_type字段）
CREATE INDEX entity_name_index IF NOT EXISTS
FOR (n:Entity) ON (n.name);

// 全文索引（用于搜索）
CREATE FULLTEXT INDEX entity_fulltext IF NOT EXISTS
FOR (n:Entity) ON EACH [n.name, n.description];

2.2.2 数据存储模式

模式1: 基于项目的数据库隔离

每个项目使用独立的Neo4j database
- database: project_1
- database: project_2

模式2: 基于标签的项目隔离

所有项目共享一个database，通过节点属性区分
- 所有节点增加 project_id 属性
- 查询时必须过滤 project_id

推荐: 使用模式2（标签隔离），更灵活且易于管理

2.3 MySQL与Neo4j的职责划分

数据类型	MySQL	Neo4j
Schema定义	✅ 存储	❌ 不存储
元数据管理	✅ 存储	❌ 不存储
项目配置	✅ 存储	❌ 不存储
用户权限	✅ 存储	❌ 不存储
图谱实例数据	❌ 不存储	✅ 存储
实体关系	❌ 不存储	✅ 存储
图查询	❌ 不支持	✅ 支持
版本快照	✅ 存储（schema_view）	❌ 不存储

数据流:

Schema文件上传
    ↓
解析并验证Schema
    ↓
存储到MySQL (kg_ontology_entity等表)
    ↓
生成Neo4j约束和索引
    ↓
数据导入
    ↓
存储到Neo4j (节点和关系)

三、Schema 文件格式分析

3.1 Schema文件结构

基于OpenSPG的schema模型，Schema文件应包含以下部分：

3.1.1 JSON Schema示例

json

{
  "namespace": "medical",
  "entities": [
    {
      "name": "Disease",
      "nameZh": "疾病",
      "category": "ENTITY_TYPE",
      "description": "疾病实体类型",
      "properties": [
        {
          "name": "name",
          "nameZh": "疾病名称",
          "type": "Text",
          "required": true,
          "unique": true
        },
        {
          "name": "icd_code",
          "nameZh": "ICD编码",
          "type": "Text",
          "pattern": "^[A-Z]\\d{2}(\\.\\d{1,2})?$"
        },
        {
          "name": "onset_age",
          "nameZh": "发病年龄",
          "type": "Integer",
          "min": 0,
          "max": 120
        }
      ],
      "parents": ["Thing"],
      "indexes": ["name", "icd_code"]
    },
    {
      "name": "Drug",
      "nameZh": "药物",
      "category": "ENTITY_TYPE",
      "properties": [
        {
          "name": "name",
          "nameZh": "药物名称",
          "type": "Text",
          "required": true
        },
        {
          "name": "components",
          "nameZh": "成分",
          "type": "Text",
          "multiValue": true
        }
      ]
    }
  ],
  "relations": [
    {
      "name": "treats",
      "nameZh": "治疗",
      "subjectType": "Drug",
      "objectType": "Disease",
      "properties": [
        {
          "name": "effectiveness",
          "nameZh": "有效性",
          "type": "Float",
          "min": 0.0,
          "max": 1.0
        }
      ]
    }
  ],
  "concepts": [
    {
      "name": "ChronicDisease",
      "nameZh": "慢性病",
      "category": "CONCEPT_TYPE",
      "parent": "Disease",
      "rule": "Disease.duration > 90"
    }
  ]
}

3.1.2 Schema解析流程

Schema文件(JSON/YAML)
    ↓
1. 格式验证
    ↓
2. 类型检查
    ↓
3. 约束验证
    ↓
4. 继承关系验证
    ↓
5. 循环依赖检测
    ↓
6. 生成内部模型(SPGSchema)
    ↓
7. 持久化到MySQL

四、持久层技术选型

4.1 MySQL持久层

4.1.1 技术栈

技术	版本	说明
MyBatis	3.5.x	ORM框架
MyBatis-Plus	3.5.x	MyBatis增强工具
Druid	1.2.x	数据库连接池
MySQL Connector	8.0.x	MySQL驱动
Spring JDBC	6.x	事务管理

4.1.2 项目结构

muotai-spg-dao/
├── src/main/java/
│   └── com/aroundix/openspg/dao/
│       ├── dataobject/          # DO对象（对应数据库表）
│       │   ├── ProjectDO.java
│       │   ├── EntityDO.java
│       │   ├── PropertyRangeDO.java
│       │   └── ...
│       ├── mapper/              # MyBatis Mapper接口
│       │   ├── ProjectMapper.java
│       │   ├── EntityMapper.java
│       │   └── ...
│       └── repository/          # Repository封装层
│           ├── ProjectRepository.java
│           ├── EntityRepository.java
│           └── ...
└── src/main/resources/
    └── mapper/                  # MyBatis XML文件
        ├── ProjectMapper.xml
        ├── EntityMapper.xml
        └── ...

4.2 Neo4j持久层

4.2.1 技术栈

技术	版本	说明
Neo4j Java Driver	5.x	官方驱动
Spring Data Neo4j	7.x	Spring Data集成
CloudExt GraphStore	本地	已迁移的图存储抽象层

4.2.2 项目结构

muotai-spg-graph/
├── src/main/java/
│   └── com/aroundix/openspg/graph/
│       ├── entity/              # 图实体（对应Neo4j节点）
│       │   ├── GraphNode.java
│       │   ├── GraphRelation.java
│       │   └── ...
│       ├── repository/          # Neo4j Repository
│       │   ├── NodeRepository.java
│       │   ├── RelationRepository.java
│       │   └── ...
│       └── service/             # 图操作服务
│           ├── GraphStoreService.java
│           ├── SchemaService.java
│           └── ...

五、详细实施步骤

5.1 阶段一：基础设施搭建（1-2天）

5.1.1 创建muotai-spg-dao模块

步骤:

创建Maven模块
添加依赖配置
配置MyBatis和Druid

pom.xml配置:

xml

<dependencies>
    <!-- MyBatis Plus -->
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <version>3.5.5</version>
    </dependency>

    <!-- Druid -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid-spring-boot-starter</artifactId>
        <version>1.2.20</version>
    </dependency>

    <!-- MySQL Driver -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.33</version>
    </dependency>
</dependencies>

application.yml配置:

yaml

spring:
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/openspg?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
    username: root
    password: ${DB_PASSWORD}
    druid:
      initial-size: 5
      min-idle: 5
      max-active: 20
      max-wait: 60000

mybatis-plus:
  mapper-locations: classpath:mapper/**/*.xml
  type-aliases-package: com.aroundix.openspg.dao.dataobject
  configuration:
    map-underscore-to-camel-case: true
    log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl

5.1.2 创建muotai-spg-graph模块

步骤:

创建Maven模块
添加Neo4j依赖
配置CloudExt GraphStore

pom.xml配置:

xml

<dependencies>
    <!-- Neo4j Driver -->
    <dependency>
        <groupId>org.neo4j.driver</groupId>
        <artifactId>neo4j-java-driver</artifactId>
        <version>5.15.0</version>
    </dependency>

    <!-- Spring Data Neo4j (可选) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-neo4j</artifactId>
    </dependency>

    <!-- CloudExt (本地依赖) -->
    <dependency>
        <groupId>com.aroundix.openspg</groupId>
        <artifactId>muotai-spg-biz</artifactId>
        <version>${project.version}</version>
    </dependency>
</dependencies>

application.yml配置:

yaml

spring:
  neo4j:
    uri: bolt://localhost:7687
    authentication:
      username: neo4j
      password: ${NEO4J_PASSWORD}
    max-transaction-retry-time: 30s

5.1.3 执行数据库初始化

步骤:

复制openspg-initdb.sql到muotai-spg项目
修改数据库名和表前缀
执行SQL脚本创建表

SQL文件位置:

muotai-spg-dao/src/main/resources/db/
├── schema.sql           # 表结构
├── data.sql             # 初始数据
└── migration/           # 版本迁移脚本
    ├── V1__init.sql
    └── V2__add_xxx.sql

5.2 阶段二：核心Schema持久层实现（3-5天）

5.2.1 创建DataObject类

需要创建的DO类（共17个核心DO）:

DO类名	对应表名	说明
`ProjectDO`	kg_project_info	项目信息
`TenantDO`	kg_biz_domain	业务域
`EntityDO`	kg_ontology_entity	实体类型
`EntityParentDO`	kg_ontology_entity_parent	继承关系
`PropertyRangeDO`	kg_ontology_entity_property_range	SPO三元组
`PropertyConstraintDO`	kg_ontology_property_constraint	属性约束
`ProjectEntityDO`	kg_project_entity	项目-实体关联
`OntologyReleaseDO`	kg_ontology_release	版本发布
`OntologyExtDO`	kg_ontology_ext	扩展属性
`SemanticDO`	kg_ontology_semantic	语义关联
`SemanticRuleDO`	kg_semantic_rule	语义规则
`UserDO`	kg_user	用户
`RoleDO`	kg_role	角色
`ResourcePermissionDO`	kg_resource_permission	权限
`ConfigDO`	kg_config	配置
`RefDO`	kg_ref	关联
`SysLockDO`	kg_sys_lock	分布式锁

DO类示例 (EntityDO.java):

java

@Data
@TableName("kg_ontology_entity")
public class EntityDO {
    @TableId(type = IdType.AUTO)
    private Long id;

    @TableField("original_id")
    private Long originalId;

    @TableField("name")
    private String name;

    @TableField("name_zh")
    private String nameZh;

    @TableField("entity_category")
    private String entityCategory;

    @TableField("layer")
    private String layer;

    @TableField("description")
    private String description;

    @TableField("description_zh")
    private String descriptionZh;

    @TableField("status")
    private String status;

    @TableField("with_index")
    private String withIndex;

    @TableField("scope")
    private String scope;

    @TableField("version")
    private Integer version;

    @TableField("version_status")
    private String versionStatus;

    @TableField("gmt_create")
    private LocalDateTime gmtCreate;

    @TableField("gmt_modified")
    private LocalDateTime gmtModified;

    @TableField("transformer_id")
    private Long transformerId;

    @TableField("operator_config")
    private String operatorConfig;

    @TableField("config")
    private String config;

    @TableField("unique_name")
    private String uniqueName;
}

5.2.2 创建Mapper接口

Mapper接口示例 (EntityMapper.java):

java

@Mapper
public interface EntityMapper extends BaseMapper<EntityDO> {
    /**
     * 根据名称查询实体
     */
    EntityDO selectByName(@Param("name") String name);

    /**
     * 根据项目ID查询实体列表
     */
    List<EntityDO> selectByProjectId(@Param("projectId") Long projectId);

    /**
     * 根据版本状态查询
     */
    List<EntityDO> selectByVersionStatus(
        @Param("versionStatus") String versionStatus);

    /**
     * 批量插入实体
     */
    int batchInsert(@Param("entities") List<EntityDO> entities);
}

5.2.3 创建Mapper XML

Mapper XML示例 (EntityMapper.xml):

xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.aroundix.openspg.dao.mapper.EntityMapper">

    <resultMap id="BaseResultMap" type="com.aroundix.openspg.dao.dataobject.EntityDO">
        <id column="id" property="id"/>
        <result column="original_id" property="originalId"/>
        <result column="name" property="name"/>
        <result column="name_zh" property="nameZh"/>
        <!-- ... 其他字段映射 ... -->
    </resultMap>

    <select id="selectByName" resultMap="BaseResultMap">
        SELECT * FROM kg_ontology_entity
        WHERE name = #{name} AND status = '1'
    </select>

    <select id="selectByProjectId" resultMap="BaseResultMap">
        SELECT e.* FROM kg_ontology_entity e
        INNER JOIN kg_project_entity pe ON e.id = pe.entity_id
        WHERE pe.project_id = #{projectId}
          AND e.status = '1'
          AND pe.version_status = 'ONLINE'
    </select>

    <insert id="batchInsert">
        INSERT INTO kg_ontology_entity (
            original_id, name, name_zh, entity_category,
            layer, description, status
        ) VALUES
        <foreach collection="entities" item="item" separator=",">
            (
                #{item.originalId}, #{item.name}, #{item.nameZh},
                #{item.entityCategory}, #{item.layer},
                #{item.description}, #{item.status}
            )
        </foreach>
    </insert>

</mapper>

5.2.4 创建Repository层

Repository接口 (EntityRepository.java):

java

public interface EntityRepository {
    /**
     * 保存实体类型
     */
    Long saveEntity(EntityDO entity);

    /**
     * 更新实体类型
     */
    boolean updateEntity(EntityDO entity);

    /**
     * 根据ID查询
     */
    EntityDO findById(Long id);

    /**
     * 根据名称查询
     */
    EntityDO findByName(String name);

    /**
     * 查询项目的所有实体
     */
    List<EntityDO> findByProjectId(Long projectId);

    /**
     * 批量保存
     */
    int batchSave(List<EntityDO> entities);

    /**
     * 删除实体（逻辑删除）
     */
    boolean deleteEntity(Long id);
}

Repository实现 (EntityRepositoryImpl.java):

java

@Repository
public class EntityRepositoryImpl implements EntityRepository {

    @Autowired
    private EntityMapper entityMapper;

    @Override
    @Transactional
    public Long saveEntity(EntityDO entity) {
        entity.setGmtCreate(LocalDateTime.now());
        entity.setGmtModified(LocalDateTime.now());
        entity.setStatus("1"); // 1表示有效
        entityMapper.insert(entity);
        return entity.getId();
    }

    @Override
    @Transactional
    public boolean updateEntity(EntityDO entity) {
        entity.setGmtModified(LocalDateTime.now());
        return entityMapper.updateById(entity) > 0;
    }

    @Override
    public EntityDO findById(Long id) {
        return entityMapper.selectById(id);
    }

    @Override
    public EntityDO findByName(String name) {
        return entityMapper.selectByName(name);
    }

    @Override
    public List<EntityDO> findByProjectId(Long projectId) {
        return entityMapper.selectByProjectId(projectId);
    }

    @Override
    @Transactional
    public int batchSave(List<EntityDO> entities) {
        if (entities == null || entities.isEmpty()) {
            return 0;
        }
        LocalDateTime now = LocalDateTime.now();
        entities.forEach(e -> {
            e.setGmtCreate(now);
            e.setGmtModified(now);
            e.setStatus("1");
        });
        return entityMapper.batchInsert(entities);
    }

    @Override
    @Transactional
    public boolean deleteEntity(Long id) {
        EntityDO entity = new EntityDO();
        entity.setId(id);
        entity.setStatus("9"); // 9表示删除
        entity.setGmtModified(LocalDateTime.now());
        return entityMapper.updateById(entity) > 0;
    }
}

5.3 阶段三：Schema解析与转换（2-3天）

5.3.1 创建Schema解析器

SchemaParser接口:

java

public interface SchemaParser {
    /**
     * 解析Schema文件
     */
    SPGSchema parse(InputStream inputStream) throws SchemaParseException;

    /**
     * 验证Schema
     */
    ValidationResult validate(SPGSchema schema);

    /**
     * 转换为DO对象
     */
    SchemaDataObjects convertToDataObjects(SPGSchema schema, Long projectId);
}

SchemaDataObjects类:

java

@Data
public class SchemaDataObjects {
    private List<EntityDO> entities;
    private List<EntityParentDO> parentRelations;
    private List<PropertyRangeDO> properties;
    private List<PropertyConstraintDO> constraints;
    private List<SemanticDO> semantics;
    private List<SemanticRuleDO> rules;
}

5.3.2 实现JSON Schema解析

JSONSchemaParser实现:

java

@Component
public class JSONSchemaParser implements SchemaParser {

    @Override
    public SPGSchema parse(InputStream inputStream) throws SchemaParseException {
        try {
            ObjectMapper mapper = new ObjectMapper();
            JsonNode rootNode = mapper.readTree(inputStream);

            SPGSchema.Builder builder = SPGSchema.builder();

            // 解析namespace
            String namespace = rootNode.get("namespace").asText();
            builder.namespace(namespace);

            // 解析entities
            JsonNode entitiesNode = rootNode.get("entities");
            if (entitiesNode != null && entitiesNode.isArray()) {
                for (JsonNode entityNode : entitiesNode) {
                    SPGTypeRef entity = parseEntity(entityNode);
                    builder.addType(entity);
                }
            }

            // 解析relations
            JsonNode relationsNode = rootNode.get("relations");
            if (relationsNode != null && relationsNode.isArray()) {
                for (JsonNode relationNode : relationsNode) {
                    RelationTypeRef relation = parseRelation(relationNode);
                    builder.addRelation(relation);
                }
            }

            return builder.build();

        } catch (IOException e) {
            throw new SchemaParseException("Failed to parse schema", e);
        }
    }

    private SPGTypeRef parseEntity(JsonNode entityNode) {
        // 实现实体解析逻辑
        // ...
    }

    private RelationTypeRef parseRelation(JsonNode relationNode) {
        // 实现关系解析逻辑
        // ...
    }

    @Override
    public ValidationResult validate(SPGSchema schema) {
        ValidationResult result = new ValidationResult();

        // 1. 验证实体类型唯一性
        // 2. 验证属性类型合法性
        // 3. 验证继承关系无循环
        // 4. 验证约束条件合法性

        return result;
    }

    @Override
    public SchemaDataObjects convertToDataObjects(SPGSchema schema, Long projectId) {
        SchemaDataObjects dataObjects = new SchemaDataObjects();

        // 转换实体为EntityDO
        List<EntityDO> entities = new ArrayList<>();
        for (SPGTypeRef type : schema.getTypes()) {
            EntityDO entityDO = convertEntityDO(type, projectId);
            entities.add(entityDO);
        }
        dataObjects.setEntities(entities);

        // 转换属性为PropertyRangeDO
        // ...

        return dataObjects;
    }

    private EntityDO convertEntityDO(SPGTypeRef type, Long projectId) {
        EntityDO entity = new EntityDO();
        entity.setName(type.getName());
        entity.setNameZh(type.getNameZh());
        entity.setEntityCategory(type.getCategory());
        // ... 设置其他字段
        return entity;
    }
}

5.4 阶段四：Neo4j图存储实现（3-4天）

5.4.1 创建GraphStore服务

GraphStoreService接口:

java

public interface GraphStoreService {
    /**
     * 根据Schema创建图结构
     */
    void createGraphSchema(SPGSchema schema, Long projectId);

    /**
     * 创建节点
     */
    String createNode(String label, Map<String, Object> properties, Long projectId);

    /**
     * 创建关系
     */
    void createRelationship(
        String sourceId, String targetId,
        String relationType, Map<String, Object> properties,
        Long projectId
    );

    /**
     * 查询节点
     */
    Map<String, Object> getNode(String id, Long projectId);

    /**
     * 删除节点
     */
    void deleteNode(String id, Long projectId);

    /**
     * 批量导入
     */
    void batchImport(List<GraphNode> nodes, List<GraphRelation> relations, Long projectId);
}

5.4.2 使用CloudExt实现

Neo4jGraphStoreServiceImpl:

java

@Service
public class Neo4jGraphStoreServiceImpl implements GraphStoreService {

    @Autowired
    private GraphStoreClient graphStoreClient;

    @Override
    public void createGraphSchema(SPGSchema schema, Long projectId) {
        // 1. 创建唯一性约束
        for (SPGTypeRef type : schema.getTypes()) {
            String label = type.getName();
            String cypher = String.format(
                "CREATE CONSTRAINT %s_id_unique IF NOT EXISTS " +
                "FOR (n:%s) REQUIRE n.id IS UNIQUE",
                label.toLowerCase(), label
            );
            graphStoreClient.execute(cypher);
        }

        // 2. 创建索引
        for (SPGTypeRef type : schema.getTypes()) {
            for (BasePropertyRef property : type.getProperties()) {
                if (property.isIndexed()) {
                    String label = type.getName();
                    String propertyName = property.getName();
                    String cypher = String.format(
                        "CREATE INDEX %s_%s_index IF NOT EXISTS " +
                        "FOR (n:%s) ON (n.%s)",
                        label.toLowerCase(), propertyName,
                        label, propertyName
                    );
                    graphStoreClient.execute(cypher);
                }
            }
        }
    }

    @Override
    public String createNode(String label, Map<String, Object> properties, Long projectId) {
        // 添加project_id
        properties.put("project_id", projectId);

        // 生成ID（如果没有）
        String id = (String) properties.getOrDefault("id", UUID.randomUUID().toString());
        properties.put("id", id);

        // 构建Cypher
        StringBuilder cypher = new StringBuilder("CREATE (n:");
        cypher.append(label).append(" {");

        List<String> propAssignments = new ArrayList<>();
        for (Map.Entry<String, Object> entry : properties.entrySet()) {
            propAssignments.add(entry.getKey() + ": $" + entry.getKey());
        }
        cypher.append(String.join(", ", propAssignments));
        cypher.append("}) RETURN n.id");

        // 执行
        graphStoreClient.execute(cypher.toString(), properties);

        return id;
    }

    @Override
    public void createRelationship(
            String sourceId, String targetId,
            String relationType, Map<String, Object> properties,
            Long projectId) {

        properties.put("project_id", projectId);

        String cypher = String.format(
            "MATCH (a {id: $sourceId, project_id: $projectId}), " +
            "      (b {id: $targetId, project_id: $projectId}) " +
            "CREATE (a)-[r:%s]->(b) " +
            "SET r = $properties",
            relationType
        );

        Map<String, Object> params = new HashMap<>();
        params.put("sourceId", sourceId);
        params.put("targetId", targetId);
        params.put("projectId", projectId);
        params.put("properties", properties);

        graphStoreClient.execute(cypher, params);
    }

    @Override
    public Map<String, Object> getNode(String id, Long projectId) {
        String cypher = "MATCH (n {id: $id, project_id: $projectId}) RETURN n";

        Map<String, Object> params = new HashMap<>();
        params.put("id", id);
        params.put("projectId", projectId);

        // 执行查询并返回结果
        return graphStoreClient.queryForObject(cypher, params);
    }

    @Override
    public void deleteNode(String id, Long projectId) {
        String cypher = "MATCH (n {id: $id, project_id: $projectId}) " +
                       "DETACH DELETE n";

        Map<String, Object> params = new HashMap<>();
        params.put("id", id);
        params.put("projectId", projectId);

        graphStoreClient.execute(cypher, params);
    }

    @Override
    @Transactional
    public void batchImport(List<GraphNode> nodes, List<GraphRelation> relations, Long projectId) {
        // 批量创建节点
        for (GraphNode node : nodes) {
            createNode(node.getLabel(), node.getProperties(), projectId);
        }

        // 批量创建关系
        for (GraphRelation relation : relations) {
            createRelationship(
                relation.getSourceId(),
                relation.getTargetId(),
                relation.getType(),
                relation.getProperties(),
                projectId
            );
        }
    }
}

5.5 阶段五：业务服务层实现（3-4天）

5.5.1 Schema服务

SchemaService接口:

java

public interface SchemaService {
    /**
     * 上传并解析Schema文件
     */
    Long uploadSchema(MultipartFile file, Long projectId, String userId)
        throws SchemaParseException;

    /**
     * 保存Schema
     */
    Long saveSchema(SPGSchema schema, Long projectId, String userId);

    /**
     * 发布Schema版本
     */
    void publishSchema(Long projectId, Integer version, String userId);

    /**
     * 获取Schema
     */
    SPGSchema getSchema(Long projectId, Integer version);

    /**
     * 获取最新Schema
     */
    SPGSchema getLatestSchema(Long projectId);

    /**
     * 删除Schema
     */
    void deleteSchema(Long projectId, Integer version);
}

SchemaServiceImpl实现:

java

@Service
public class SchemaServiceImpl implements SchemaService {

    @Autowired
    private SchemaParser schemaParser;

    @Autowired
    private EntityRepository entityRepository;

    @Autowired
    private PropertyRangeRepository propertyRangeRepository;

    @Autowired
    private OntologyReleaseRepository releaseRepository;

    @Autowired
    private GraphStoreService graphStoreService;

    @Override
    @Transactional
    public Long uploadSchema(MultipartFile file, Long projectId, String userId)
            throws SchemaParseException {

        // 1. 解析Schema文件
        SPGSchema schema = schemaParser.parse(file.getInputStream());

        // 2. 验证Schema
        ValidationResult validationResult = schemaParser.validate(schema);
        if (!validationResult.isValid()) {
            throw new SchemaParseException(validationResult.getErrors());
        }

        // 3. 保存Schema
        return saveSchema(schema, projectId, userId);
    }

    @Override
    @Transactional
    public Long saveSchema(SPGSchema schema, Long projectId, String userId) {
        // 1. 转换为DO对象
        SchemaDataObjects dataObjects = schemaParser.convertToDataObjects(schema, projectId);

        // 2. 保存实体
        List<EntityDO> entities = dataObjects.getEntities();
        entityRepository.batchSave(entities);

        // 3. 保存属性
        List<PropertyRangeDO> properties = dataObjects.getProperties();
        propertyRangeRepository.batchSave(properties);

        // 4. 保存继承关系
        // ...

        // 5. 创建版本记录
        OntologyReleaseDO release = new OntologyReleaseDO();
        release.setProjectId(projectId);
        release.setVersion(getNextVersion(projectId));
        release.setUserId(userId);
        release.setStatus("DRAFT");
        release.setSchemaView(serializeSchema(schema));
        Long releaseId = releaseRepository.save(release);

        // 6. 在Neo4j中创建Schema结构
        graphStoreService.createGraphSchema(schema, projectId);

        return releaseId;
    }

    @Override
    @Transactional
    public void publishSchema(Long projectId, Integer version, String userId) {
        // 1. 更新发布状态
        OntologyReleaseDO release = releaseRepository.findByProjectAndVersion(projectId, version);
        release.setStatus("ONLINE");
        releaseRepository.update(release);

        // 2. 将旧版本设为HISTORY
        List<OntologyReleaseDO> oldReleases = releaseRepository.findOnlineVersions(projectId);
        oldReleases.forEach(r -> {
            if (!r.getVersion().equals(version)) {
                r.setStatus("HISTORY");
                releaseRepository.update(r);
            }
        });

        // 3. 更新实体版本状态
        // ...
    }

    // ... 其他方法实现
}

5.5.2 数据导入服务

DataImportService接口:

java

public interface DataImportService {
    /**
     * 从CSV导入数据
     */
    void importFromCSV(MultipartFile file, String entityType, Long projectId);

    /**
     * 从JSON导入数据
     */
    void importFromJSON(MultipartFile file, Long projectId);

    /**
     * 批量导入节点
     */
    void importNodes(List<Map<String, Object>> nodes, String label, Long projectId);

    /**
     * 批量导入关系
     */
    void importRelations(List<Map<String, Object>> relations, String relationType, Long projectId);
}

5.6 阶段六：测试与优化（2-3天）

5.6.1 单元测试

EntityRepositoryTest:

java

@SpringBootTest
public class EntityRepositoryTest {

    @Autowired
    private EntityRepository entityRepository;

    @Test
    public void testSaveEntity() {
        EntityDO entity = new EntityDO();
        entity.setName("TestEntity");
        entity.setNameZh("测试实体");
        entity.setEntityCategory("ENTITY_TYPE");

        Long id = entityRepository.saveEntity(entity);

        assertNotNull(id);
        assertTrue(id > 0);
    }

    @Test
    public void testFindByName() {
        EntityDO entity = entityRepository.findByName("Thing");

        assertNotNull(entity);
        assertEquals("Thing", entity.getName());
    }

    @Test
    public void testBatchSave() {
        List<EntityDO> entities = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            EntityDO entity = new EntityDO();
            entity.setName("TestEntity" + i);
            entity.setNameZh("测试实体" + i);
            entity.setEntityCategory("ENTITY_TYPE");
            entities.add(entity);
        }

        int count = entityRepository.batchSave(entities);

        assertEquals(10, count);
    }
}

5.6.2 集成测试

SchemaServiceIntegrationTest:

java

@SpringBootTest
public class SchemaServiceIntegrationTest {

    @Autowired
    private SchemaService schemaService;

    @Test
    public void testUploadAndPublishSchema() throws Exception {
        // 1. 准备测试数据
        MockMultipartFile file = new MockMultipartFile(
            "schema",
            "test-schema.json",
            "application/json",
            getTestSchemaJSON().getBytes()
        );

        Long projectId = 1L;
        String userId = "test_user";

        // 2. 上传Schema
        Long releaseId = schemaService.uploadSchema(file, projectId, userId);

        assertNotNull(releaseId);

        // 3. 发布Schema
        schemaService.publishSchema(projectId, 1, userId);

        // 4. 验证Schema已发布
        SPGSchema schema = schemaService.getLatestSchema(projectId);

        assertNotNull(schema);
        assertNotNull(schema.getTypes());
        assertTrue(schema.getTypes().size() > 0);
    }

    private String getTestSchemaJSON() {
        return "{ \"namespace\": \"test\", \"entities\": [...] }";
    }
}

5.6.3 性能测试

重点测试项:

批量插入性能（10000条数据）
复杂查询性能（多表Join）
Neo4j图查询性能（多跳查询）
并发写入测试

优化方向:

使用批量插入减少数据库交互
添加合适的索引
使用缓存（Redis）减少数据库查询
使用事务批处理提高吞吐量

六、接口设计

6.1 Schema管理接口

6.1.1 上传Schema

http

POST /api/schema/upload
Content-Type: multipart/form-data

参数:
- file: Schema文件（JSON/YAML）
- projectId: 项目ID
- description: 版本描述

响应:
{
  "code": 200,
  "message": "Success",
  "data": {
    "releaseId": 123,
    "version": 1,
    "status": "DRAFT"
  }
}

6.1.2 发布Schema

http

POST /api/schema/publish

参数:
{
  "projectId": 1,
  "version": 1,
  "userId": "admin"
}

响应:
{
  "code": 200,
  "message": "Schema published successfully",
  "data": {
    "version": 1,
    "status": "ONLINE",
    "publishTime": "2025-12-15T10:00:00"
  }
}

6.1.3 获取Schema

http

GET /api/schema?projectId=1&version=1

响应:
{
  "code": 200,
  "data": {
    "namespace": "medical",
    "entities": [...],
    "relations": [...],
    "version": 1,
    "status": "ONLINE"
  }
}

6.2 数据导入接口

6.2.1 导入节点数据

http

POST /api/data/import/nodes
Content-Type: multipart/form-data

参数:
- file: CSV/JSON文件
- entityType: 实体类型名称
- projectId: 项目ID

响应:
{
  "code": 200,
  "message": "Import started",
  "data": {
    "taskId": "task_123",
    "totalCount": 10000,
    "status": "PROCESSING"
  }
}

6.2.2 查询导入任务状态

http

GET /api/data/import/status?taskId=task_123

响应:
{
  "code": 200,
  "data": {
    "taskId": "task_123",
    "status": "COMPLETED",
    "successCount": 9998,
    "failCount": 2,
    "errors": [...]
  }
}

七、风险与挑战

7.1 技术风险

风险项	风险等级	应对措施
Schema复杂度高	高	充分测试，逐步迭代
数据一致性	高	使用分布式事务（Seata）
Neo4j性能瓶颈	中	合理设计索引，使用批量操作
大数据量导入	中	使用异步任务，分批处理
版本冲突	中	使用乐观锁，冲突检测

7.2 实施挑战

挑战项	影响	解决方案
Schema格式不统一	中	定义标准格式，提供转换工具
历史数据迁移	高	提供迁移脚本和工具
并发控制	中	使用分布式锁
权限控制	中	基于RBAC的细粒度权限

八、时间计划

8.1 总体时间表

阶段	工作内容	预计时间	依赖
阶段一	基础设施搭建	1-2天	-
阶段二	Schema持久层实现	3-5天	阶段一
阶段三	Schema解析与转换	2-3天	阶段二
阶段四	Neo4j图存储实现	3-4天	阶段二
阶段五	业务服务层实现	3-4天	阶段三、四
阶段六	测试与优化	2-3天	阶段五
总计	-	14-21天	-

8.2 里程碑

里程碑	时间点	交付物
M1: 基础设施完成	第2天	MySQL表、Neo4j连接配置完成
M2: 核心DO和Mapper完成	第7天	17个核心表的CRUD完成
M3: Schema解析完成	第10天	支持JSON Schema上传和解析
M4: 图存储完成	第14天	支持Neo4j节点和关系创建
M5: 完整流程打通	第18天	支持Schema上传→解析→存储→发布
M6: 测试完成	第21天	单元测试、集成测试通过

九、后续扩展

9.1 功能扩展

Schema版本对比: 可视化展示不同版本之间的差异
Schema校验增强: 支持更复杂的业务规则校验
数据血缘: 记录数据的来源和转换过程
元数据搜索: 支持全文搜索Schema定义
Schema导入导出: 支持多种格式的导入导出

9.2 性能优化

缓存策略:
- Schema缓存（Redis）
- 查询结果缓存
- 元数据缓存
异步处理:
- 大数据量导入使用消息队列
- Schema发布异步通知
读写分离:
- MySQL主从复制
- Neo4j只读副本

9.3 监控与运维

监控指标:
- 数据库连接池状态
- Neo4j查询性能
- Schema操作审计日志
告警机制:
- 数据库慢查询告警
- 磁盘空间告警
- 异常操作告警

十、总结

本实施计划提供了muotai-spg项目持久层的完整设计方案，包括：

✅ 完整的数据库设计（MySQL 17张表 + Neo4j图模型） ✅ 清晰的技术选型（MyBatis-Plus + Neo4j Driver + CloudExt） ✅ 详细的实施步骤（6个阶段，14-21天） ✅ 完整的代码示例（DO、Mapper、Service层） ✅ 风险应对措施（技术风险和实施挑战） ✅ 扩展性设计（支持未来功能扩展）

关键成功因素:

严格按照阶段顺序实施
充分的单元测试和集成测试
及时的代码评审和重构
完善的文档和注释

下一步行动:

✅ 阅读并理解本实施计划
⏸ 准备开发环境（MySQL、Neo4j）
⏸ 开始阶段一：基础设施搭建
⏸ 按计划逐步推进各阶段工作

文档版本: v1.0 编制人: Claude AI Assistant 审核人: 待定 生效日期: 待定

Muotai-SPG 持久层实施计划 ​

文档说明 ​

一、项目背景与目标 ​

1.1 项目背景 ​

1.2 实施目标 ​

二、数据库架构分析 ​

2.1 MySQL数据库设计 ​

2.1.1 核心业务表（17张表） ​

A. 项目与租户管理（3张表） ​

B. 本体Schema管理（6张表） ​

C. 项目关联（3张表） ​

D. 权限与配置（5张表） ​

E. 模型与应用（6张表） ​

2.1.2 表关系图 ​

2.2 Neo4j图数据库设计 ​

2.2.1 图模型结构 ​

A. 节点（Node）标签设计 ​

B. 关系（Relationship）类型设计 ​

C. 索引设计 ​

2.2.2 数据存储模式 ​

2.3 MySQL与Neo4j的职责划分 ​

三、Schema 文件格式分析 ​

3.1 Schema文件结构 ​

3.1.1 JSON Schema示例 ​

3.1.2 Schema解析流程 ​

四、持久层技术选型 ​

4.1 MySQL持久层 ​

4.1.1 技术栈 ​

4.1.2 项目结构 ​

4.2 Neo4j持久层 ​

4.2.1 技术栈 ​

4.2.2 项目结构 ​

五、详细实施步骤 ​

5.1 阶段一：基础设施搭建（1-2天） ​

5.1.1 创建muotai-spg-dao模块 ​

5.1.2 创建muotai-spg-graph模块 ​

5.1.3 执行数据库初始化 ​

5.2 阶段二：核心Schema持久层实现（3-5天） ​

5.2.1 创建DataObject类 ​

5.2.2 创建Mapper接口 ​

5.2.3 创建Mapper XML ​

5.2.4 创建Repository层 ​

5.3 阶段三：Schema解析与转换（2-3天） ​

5.3.1 创建Schema解析器 ​

5.3.2 实现JSON Schema解析 ​

5.4 阶段四：Neo4j图存储实现（3-4天） ​

5.4.1 创建GraphStore服务 ​

5.4.2 使用CloudExt实现 ​

5.5 阶段五：业务服务层实现（3-4天） ​

5.5.1 Schema服务 ​

5.5.2 数据导入服务 ​

5.6 阶段六：测试与优化（2-3天） ​

5.6.1 单元测试 ​

5.6.2 集成测试 ​

5.6.3 性能测试 ​

六、接口设计 ​

6.1 Schema管理接口 ​

6.1.1 上传Schema ​

6.1.2 发布Schema ​

6.1.3 获取Schema ​

6.2 数据导入接口 ​

6.2.1 导入节点数据 ​

6.2.2 查询导入任务状态 ​

七、风险与挑战 ​

7.1 技术风险 ​

7.2 实施挑战 ​

八、时间计划 ​

8.1 总体时间表 ​

8.2 里程碑 ​

九、后续扩展 ​

9.1 功能扩展 ​

9.2 性能优化 ​

9.3 监控与运维 ​

十、总结 ​

Muotai-SPG 持久层实施计划

文档说明

一、项目背景与目标

1.1 项目背景

1.2 实施目标

二、数据库架构分析

2.1 MySQL数据库设计

2.1.1 核心业务表（17张表）

A. 项目与租户管理（3张表）

B. 本体Schema管理（6张表）

C. 项目关联（3张表）

D. 权限与配置（5张表）

E. 模型与应用（6张表）

2.1.2 表关系图

2.2 Neo4j图数据库设计

2.2.1 图模型结构

A. 节点（Node）标签设计

B. 关系（Relationship）类型设计

C. 索引设计

2.2.2 数据存储模式

2.3 MySQL与Neo4j的职责划分

三、Schema 文件格式分析

3.1 Schema文件结构

3.1.1 JSON Schema示例

3.1.2 Schema解析流程

四、持久层技术选型

4.1 MySQL持久层

4.1.1 技术栈

4.1.2 项目结构

4.2 Neo4j持久层

4.2.1 技术栈

4.2.2 项目结构

五、详细实施步骤

5.1 阶段一：基础设施搭建（1-2天）

5.1.1 创建muotai-spg-dao模块

5.1.2 创建muotai-spg-graph模块

5.1.3 执行数据库初始化

5.2 阶段二：核心Schema持久层实现（3-5天）

5.2.1 创建DataObject类

5.2.2 创建Mapper接口

5.2.3 创建Mapper XML

5.2.4 创建Repository层

5.3 阶段三：Schema解析与转换（2-3天）

5.3.1 创建Schema解析器

5.3.2 实现JSON Schema解析

5.4 阶段四：Neo4j图存储实现（3-4天）

5.4.1 创建GraphStore服务

5.4.2 使用CloudExt实现

5.5 阶段五：业务服务层实现（3-4天）

5.5.1 Schema服务

5.5.2 数据导入服务

5.6 阶段六：测试与优化（2-3天）

5.6.1 单元测试

5.6.2 集成测试

5.6.3 性能测试

六、接口设计

6.1 Schema管理接口

6.1.1 上传Schema

6.1.2 发布Schema

6.1.3 获取Schema

6.2 数据导入接口

6.2.1 导入节点数据

6.2.2 查询导入任务状态

七、风险与挑战

7.1 技术风险

7.2 实施挑战

八、时间计划

8.1 总体时间表

8.2 里程碑

九、后续扩展

9.1 功能扩展

9.2 性能优化

9.3 监控与运维

十、总结