📚 第1章：Neo4j基础概念与架构

1.1 Neo4j简介

什么是Neo4j

Neo4j是世界上领先的图数据库，专门为存储和查询高度连接的数据而设计。它使用图结构来表示和存储数据，其中数据实体存储为节点，实体之间的关系存储为边。

图数据库的优势

1. 直观的数据模型：图结构更接近人类思维方式
2. 高效的关系查询：在处理复杂关系时性能优异
3. 灵活的模式：支持动态添加属性和关系类型
4. ACID事务支持：保证数据一致性
5. 水平扩展能力：支持集群部署

应用场景

• 社交网络分析：用户关系、好友推荐
• 推荐系统：基于关系的个性化推荐
• 欺诈检测：识别异常交易模式
• 知识图谱：构建领域知识网络
• 网络安全：分析攻击路径和威胁传播
• 供应链管理：追踪产品流向和依赖关系

1.2 核心概念

节点（Nodes）

节点是图中的基本实体，代表现实世界中的对象。

-- 创建一个用户节点
CREATE (u:User {name: "张三", age: 30, email: "zhangsan@example.com"})

节点特性： - 可以有零个或多个标签（Labels） - 可以有零个或多个属性（Properties） - 每个节点都有唯一的内部ID

关系（Relationships）

关系连接两个节点，表示它们之间的联系。

-- 创建关系
MATCH (u1:User {name: "张三"}), (u2:User {name: "李四"})
CREATE (u1)-[:FOLLOWS {since: "2023-01-01"}]->(u2)

关系特性： - 必须有一个类型（Type） - 总是有方向的 - 可以有零个或多个属性 - 连接两个节点（可以是同一个节点）

标签（Labels）

标签用于对节点进行分类，一个节点可以有多个标签。

-- 多标签节点
CREATE (p:Person:Employee:Manager {name: "王五"})

属性（Properties）

属性是键值对，可以附加到节点和关系上。

-- 节点属性
CREATE (u:User {
  name: "赵六",
  age: 25,
  active: true,
  tags: ["developer", "python"]
})

1.3 Neo4j架构

存储架构

┌─────────────────────────────────────┐
│           Neo4j 数据库              │
├─────────────────────────────────────┤
│  节点存储    │  关系存储  │ 属性存储  │
│  (Nodes)    │ (Relations)│(Properties)│
├─────────────────────────────────────┤
│           索引存储                   │
│     (Schema & Full-text)           │
├─────────────────────────────────────┤
│           事务日志                   │
│       (Transaction Logs)           │
└─────────────────────────────────────┘

核心组件

1. 图引擎（Graph Engine）

   • 负责图遍历和查询执行
   • 优化查询计划
   • 管理缓存

2. 存储引擎（Storage Engine）

   • 管理数据持久化
   • 处理ACID事务
   • 维护数据完整性

3. 查询引擎（Query Engine）

   • 解析Cypher查询
   • 生成执行计划
   • 优化查询性能

4. 索引管理器（Index Manager）

   • 维护节点和关系索引
   • 支持全文搜索
   • 提供约束检查

1.4 Neo4j版本与部署模式

版本对比

特性	Community Edition	Enterprise Edition
基础图功能	✓	✓
Cypher查询	✓	✓
ACID事务	✓	✓
集群支持	✗	✓
在线备份	✗	✓
高级监控	✗	✓
安全功能	基础	高级

部署模式

1. 单实例部署

┌─────────────────┐
│   Neo4j Server  │
│                 │
│  ┌───────────┐  │
│  │ Database  │  │
│  └───────────┘  │
└─────────────────┘

2. 因果集群（Causal Cluster）

┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│ Core Server │  │ Core Server │  │ Core Server │
│   (Leader)  │  │ (Follower)  │  │ (Follower)  │
└─────────────┘  └─────────────┘  └─────────────┘
       │                │                │
       └────────────────┼────────────────┘
                        │
┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│Read Replica │  │Read Replica │  │Read Replica │
└─────────────┘  └─────────────┘  └─────────────┘

1.5 与其他数据库的对比

Neo4j vs 关系型数据库

特性	Neo4j	关系型数据库
数据模型	图模型	表格模型
关系查询	原生支持	JOIN操作
模式灵活性	高	低
复杂关系性能	优秀	随深度下降
ACID支持	完整	完整
查询语言	Cypher	SQL

Neo4j vs 文档数据库

特性	Neo4j	MongoDB
关系处理	原生优化	需要应用层处理
图遍历	高效	复杂
数据一致性	ACID	最终一致性
查询复杂度	适合复杂关系	适合文档查询

1.6 Python集成示例

安装Neo4j驱动

pip install neo4j

基础连接示例

from neo4j import GraphDatabase
import logging

class Neo4jConnection:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
        
    def close(self):
        if self.driver is not None:
            self.driver.close()
            
    def query(self, query, parameters=None, db=None):
        assert self.driver is not None, "Driver not initialized!"
        session = None
        response = None
        try:
            session = self.driver.session(database=db) if db is not None else self.driver.session()
            response = list(session.run(query, parameters))
        except Exception as e:
            print(f"Query failed: {e}")
        finally:
            if session is not None:
                session.close()
        return response

# 使用示例
conn = Neo4jConnection("bolt://localhost:7687", "neo4j", "password")

# 创建节点
conn.query("""
    CREATE (u:User {name: $name, age: $age})
""", parameters={"name": "张三", "age": 30})

# 查询节点
result = conn.query("""
    MATCH (u:User {name: $name})
    RETURN u.name as name, u.age as age
""", parameters={"name": "张三"})

print(result)
conn.close()

图数据模型设计示例

class SocialNetworkModel:
    def __init__(self, connection):
        self.conn = connection
        
    def create_user(self, user_id, name, email, age=None):
        """创建用户节点"""
        query = """
        CREATE (u:User {
            user_id: $user_id,
            name: $name,
            email: $email,
            age: $age,
            created_at: datetime()
        })
        RETURN u
        """
        return self.conn.query(query, {
            "user_id": user_id,
            "name": name,
            "email": email,
            "age": age
        })
    
    def create_friendship(self, user1_id, user2_id, since=None):
        """创建好友关系"""
        query = """
        MATCH (u1:User {user_id: $user1_id})
        MATCH (u2:User {user_id: $user2_id})
        CREATE (u1)-[:FRIENDS {
            since: $since,
            created_at: datetime()
        }]->(u2)
        CREATE (u2)-[:FRIENDS {
            since: $since,
            created_at: datetime()
        }]->(u1)
        """
        return self.conn.query(query, {
            "user1_id": user1_id,
            "user2_id": user2_id,
            "since": since
        })
    
    def find_mutual_friends(self, user1_id, user2_id):
        """查找共同好友"""
        query = """
        MATCH (u1:User {user_id: $user1_id})-[:FRIENDS]-(mutual)-[:FRIENDS]-(u2:User {user_id: $user2_id})
        RETURN mutual.name as mutual_friend, mutual.user_id as mutual_id
        """
        return self.conn.query(query, {
            "user1_id": user1_id,
            "user2_id": user2_id
        })
    
    def recommend_friends(self, user_id, limit=5):
        """好友推荐算法"""
        query = """
        MATCH (u:User {user_id: $user_id})-[:FRIENDS]-(friend)-[:FRIENDS]-(recommendation)
        WHERE NOT (u)-[:FRIENDS]-(recommendation) AND u <> recommendation
        WITH recommendation, count(*) as mutual_friends
        RETURN recommendation.name as name, 
               recommendation.user_id as user_id,
               mutual_friends
        ORDER BY mutual_friends DESC
        LIMIT $limit
        """
        return self.conn.query(query, {
            "user_id": user_id,
            "limit": limit
        })

# 使用示例
model = SocialNetworkModel(conn)

# 创建用户
model.create_user("001", "张三", "zhangsan@example.com", 30)
model.create_user("002", "李四", "lisi@example.com", 25)
model.create_user("003", "王五", "wangwu@example.com", 28)

# 创建好友关系
model.create_friendship("001", "002", "2023-01-01")
model.create_friendship("002", "003", "2023-02-01")

# 好友推荐
recommendations = model.recommend_friends("001")
print("推荐好友:", recommendations)

1.7 性能特点

图遍历性能

import time
import random

class PerformanceTest:
    def __init__(self, connection):
        self.conn = connection
        
    def create_test_data(self, num_users=1000, num_relationships=5000):
        """创建测试数据"""
        print(f"创建 {num_users} 个用户...")
        
        # 批量创建用户
        for i in range(num_users):
            self.conn.query("""
                CREATE (u:User {
                    user_id: $user_id,
                    name: $name,
                    created_at: datetime()
                })
            """, {
                "user_id": f"user_{i:04d}",
                "name": f"User {i}"
            })
            
            if i % 100 == 0:
                print(f"已创建 {i} 个用户")
        
        print(f"创建 {num_relationships} 个关系...")
        
        # 随机创建关系
        for i in range(num_relationships):
            user1 = f"user_{random.randint(0, num_users-1):04d}"
            user2 = f"user_{random.randint(0, num_users-1):04d}"
            
            if user1 != user2:
                self.conn.query("""
                    MATCH (u1:User {user_id: $user1})
                    MATCH (u2:User {user_id: $user2})
                    MERGE (u1)-[:KNOWS]->(u2)
                """, {"user1": user1, "user2": user2})
                
            if i % 500 == 0:
                print(f"已创建 {i} 个关系")
    
    def test_graph_traversal(self, start_user, max_depth=4):
        """测试图遍历性能"""
        start_time = time.time()
        
        result = self.conn.query("""
            MATCH path = (start:User {user_id: $start_user})-[:KNOWS*1..$max_depth]-(connected)
            RETURN connected.user_id as user_id, length(path) as distance
            ORDER BY distance
        """, {
            "start_user": start_user,
            "max_depth": max_depth
        })
        
        end_time = time.time()
        
        print(f"图遍历完成:")
        print(f"- 起始用户: {start_user}")
        print(f"- 最大深度: {max_depth}")
        print(f"- 找到连接: {len(result)} 个")
        print(f"- 执行时间: {end_time - start_time:.3f} 秒")
        
        return result
    
    def test_shortest_path(self, user1, user2):
        """测试最短路径查询"""
        start_time = time.time()
        
        result = self.conn.query("""
            MATCH (u1:User {user_id: $user1}), (u2:User {user_id: $user2})
            MATCH path = shortestPath((u1)-[:KNOWS*]-(u2))
            RETURN [node in nodes(path) | node.user_id] as path,
                   length(path) as distance
        """, {"user1": user1, "user2": user2})
        
        end_time = time.time()
        
        print(f"最短路径查询:")
        print(f"- 用户1: {user1}")
        print(f"- 用户2: {user2}")
        if result:
            print(f"- 路径: {' -> '.join(result[0]['path'])}")
            print(f"- 距离: {result[0]['distance']}")
        else:
            print(f"- 无连接路径")
        print(f"- 执行时间: {end_time - start_time:.3f} 秒")
        
        return result

# 性能测试示例
perf_test = PerformanceTest(conn)

# 创建测试数据（小规模测试）
# perf_test.create_test_data(100, 500)

# 测试图遍历
# perf_test.test_graph_traversal("user_0000", 3)

# 测试最短路径
# perf_test.test_shortest_path("user_0000", "user_0050")

1.8 章节总结

核心知识点

1. 图数据模型：节点、关系、标签、属性
2. Neo4j架构：存储引擎、查询引擎、索引管理
3. 部署模式：单实例、因果集群
4. 性能优势：图遍历、关系查询优化
5. 应用场景：社交网络、推荐系统、欺诈检测

最佳实践

1. 数据建模：合理设计节点标签和关系类型
2. 索引策略：为常用查询字段创建索引
3. 查询优化：使用EXPLAIN分析查询计划
4. 批量操作：大量数据导入时使用批处理
5. 监控性能：定期检查查询性能和内存使用

练习题

1. 设计一个电商平台的图数据模型，包含用户、商品、订单、评价等实体
2. 实现一个简单的好友推荐算法
3. 比较图数据库和关系型数据库在处理多层关系查询时的性能差异
4. 设计一个知识图谱的基础结构
5. 实现一个简单的路径查找算法

---

下一章预告：在下一章中，我们将学习如何安装和配置Neo4j环境，包括单机部署和集群配置。