📚 第2章：安装与环境配置

2.1 系统要求

硬件要求

组件	最低要求	推荐配置	生产环境
CPU	2核	4核+	8核+
内存	2GB	8GB+	16GB+
磁盘	10GB	100GB+	SSD 500GB+
网络	100Mbps	1Gbps	10Gbps

软件要求

• Java: OpenJDK 11 或 Oracle JDK 11+
• 操作系统:
  • Linux: Ubuntu 18.04+, CentOS 7+, RHEL 7+
  • Windows: Windows 10+, Windows Server 2016+
  • macOS: 10.14+

2.2 Java环境配置

Linux/macOS安装Java

# Ubuntu/Debian
sudo apt update
sudo apt install openjdk-11-jdk

# CentOS/RHEL
sudo yum install java-11-openjdk-devel

# macOS (使用Homebrew)
brew install openjdk@11

Windows安装Java

1. 下载OpenJDK 11: https://adoptopenjdk.net/
2. 运行安装程序
3. 配置环境变量

# 设置JAVA_HOME
set JAVA_HOME=C:\Program Files\AdoptOpenJDK\jdk-11.0.x
set PATH=%JAVA_HOME%\bin;%PATH%

验证Java安装

java -version
javac -version

2.3 Neo4j安装

方式一：官方安装包

Linux安装

# 下载Neo4j Community Edition
wget https://neo4j.com/artifact.php?name=neo4j-community-4.4.12-unix.tar.gz

# 解压
tar -xzf neo4j-community-4.4.12-unix.tar.gz

# 移动到安装目录
sudo mv neo4j-community-4.4.12 /opt/neo4j

# 创建符号链接
sudo ln -s /opt/neo4j/bin/neo4j /usr/local/bin/neo4j
sudo ln -s /opt/neo4j/bin/cypher-shell /usr/local/bin/cypher-shell

# 设置权限
sudo chown -R $USER:$USER /opt/neo4j

Windows安装

# 下载并解压Neo4j
# 解压到 C:\neo4j

# 添加到PATH环境变量
$env:PATH += ";C:\neo4j\bin"

# 或者通过系统设置永久添加

方式二：包管理器安装

Ubuntu/Debian

# 添加Neo4j仓库
wget -O - https://debian.neo4j.com/neotechnology.gpg.key | sudo apt-key add -
echo 'deb https://debian.neo4j.com stable 4.4' | sudo tee /etc/apt/sources.list.d/neo4j.list

# 安装
sudo apt update
sudo apt install neo4j

CentOS/RHEL

# 添加Neo4j仓库
sudo rpm --import https://debian.neo4j.com/neotechnology.gpg.key

cat <<EOF | sudo tee /etc/yum.repos.d/neo4j.repo
[neo4j]
name=Neo4j RPM Repository
baseurl=https://yum.neo4j.com/stable/4.4
enabled=1
gpgcheck=1
EOF

# 安装
sudo yum install neo4j

方式三：Docker安装

# 拉取Neo4j镜像
docker pull neo4j:4.4-community

# 运行Neo4j容器
docker run \
    --name neo4j-container \
    -p 7474:7474 -p 7687:7687 \
    -d \
    -v $HOME/neo4j/data:/data \
    -v $HOME/neo4j/logs:/logs \
    -v $HOME/neo4j/import:/var/lib/neo4j/import \
    -v $HOME/neo4j/plugins:/plugins \
    --env NEO4J_AUTH=neo4j/password \
    neo4j:4.4-community

2.4 基础配置

配置文件位置

# Linux包管理器安装
/etc/neo4j/neo4j.conf

# Linux手动安装
/opt/neo4j/conf/neo4j.conf

# Windows
C:\neo4j\conf\neo4j.conf

# Docker
/var/lib/neo4j/conf/neo4j.conf

核心配置项

# neo4j.conf

# 数据库连接配置
dbms.default_listen_address=0.0.0.0
dbms.default_advertised_address=localhost

# HTTP连接器配置
dbms.connector.http.enabled=true
dbms.connector.http.listen_address=:7474
dbms.connector.http.advertised_address=:7474

# HTTPS连接器配置
dbms.connector.https.enabled=false
dbms.connector.https.listen_address=:7473

# Bolt连接器配置
dbms.connector.bolt.enabled=true
dbms.connector.bolt.listen_address=:7687
dbms.connector.bolt.advertised_address=:7687

# 内存配置
dbms.memory.heap.initial_size=512m
dbms.memory.heap.max_size=1G
dbms.memory.pagecache.size=512m

# 数据目录
dbms.directories.data=/var/lib/neo4j/data
dbms.directories.logs=/var/lib/neo4j/logs
dbms.directories.import=/var/lib/neo4j/import

# 安全配置
dbms.security.auth_enabled=true
dbms.security.allow_csv_import_from_file_urls=true

# 日志配置
dbms.logs.query.enabled=true
dbms.logs.query.threshold=0
dbms.logs.query.parameter_logging_enabled=true

Python配置管理脚本

import os
import configparser
from pathlib import Path

class Neo4jConfigManager:
    def __init__(self, config_path=None):
        if config_path is None:
            # 自动检测配置文件路径
            possible_paths = [
                "/etc/neo4j/neo4j.conf",
                "/opt/neo4j/conf/neo4j.conf",
                "C:\\neo4j\\conf\\neo4j.conf",
                os.path.expanduser("~/neo4j/conf/neo4j.conf")
            ]
            
            for path in possible_paths:
                if os.path.exists(path):
                    config_path = path
                    break
            
            if config_path is None:
                raise FileNotFoundError("Neo4j配置文件未找到")
        
        self.config_path = config_path
        self.config = {}
        self.load_config()
    
    def load_config(self):
        """加载配置文件"""
        with open(self.config_path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    if '=' in line:
                        key, value = line.split('=', 1)
                        self.config[key.strip()] = value.strip()
    
    def get_config(self, key, default=None):
        """获取配置值"""
        return self.config.get(key, default)
    
    def set_config(self, key, value):
        """设置配置值"""
        self.config[key] = str(value)
    
    def save_config(self, backup=True):
        """保存配置文件"""
        if backup:
            backup_path = f"{self.config_path}.backup"
            os.system(f"cp {self.config_path} {backup_path}")
        
        with open(self.config_path, 'w', encoding='utf-8') as f:
            f.write("# Neo4j Configuration\n")
            f.write("# Generated by Neo4jConfigManager\n\n")
            
            for key, value in self.config.items():
                f.write(f"{key}={value}\n")
    
    def configure_memory(self, heap_size="1G", pagecache_size="512m"):
        """配置内存设置"""
        self.set_config("dbms.memory.heap.initial_size", heap_size)
        self.set_config("dbms.memory.heap.max_size", heap_size)
        self.set_config("dbms.memory.pagecache.size", pagecache_size)
    
    def configure_network(self, http_port=7474, bolt_port=7687, bind_address="0.0.0.0"):
        """配置网络设置"""
        self.set_config("dbms.default_listen_address", bind_address)
        self.set_config("dbms.connector.http.listen_address", f":{http_port}")
        self.set_config("dbms.connector.bolt.listen_address", f":{bolt_port}")
    
    def enable_query_logging(self, enabled=True, threshold="0"):
        """启用查询日志"""
        self.set_config("dbms.logs.query.enabled", str(enabled).lower())
        if enabled:
            self.set_config("dbms.logs.query.threshold", threshold)
            self.set_config("dbms.logs.query.parameter_logging_enabled", "true")
    
    def configure_security(self, auth_enabled=True, csv_import=True):
        """配置安全设置"""
        self.set_config("dbms.security.auth_enabled", str(auth_enabled).lower())
        self.set_config("dbms.security.allow_csv_import_from_file_urls", str(csv_import).lower())
    
    def print_config(self):
        """打印当前配置"""
        print("Neo4j Configuration:")
        print("=" * 50)
        for key, value in sorted(self.config.items()):
            print(f"{key:<40} = {value}")

# 使用示例
if __name__ == "__main__":
    try:
        config_manager = Neo4jConfigManager()
        
        # 配置内存（适合开发环境）
        config_manager.configure_memory("2G", "1G")
        
        # 配置网络
        config_manager.configure_network()
        
        # 启用查询日志
        config_manager.enable_query_logging(True)
        
        # 配置安全
        config_manager.configure_security()
        
        # 打印配置
        config_manager.print_config()
        
        # 保存配置（创建备份）
        # config_manager.save_config()
        
    except FileNotFoundError as e:
        print(f"错误: {e}")
        print("请确保Neo4j已正确安装")

2.5 启动和停止服务

Linux系统服务

# 启动Neo4j服务
sudo systemctl start neo4j

# 停止Neo4j服务
sudo systemctl stop neo4j

# 重启Neo4j服务
sudo systemctl restart neo4j

# 设置开机自启
sudo systemctl enable neo4j

# 查看服务状态
sudo systemctl status neo4j

# 查看日志
sudo journalctl -u neo4j -f

手动启动

# 前台启动（调试模式）
neo4j console

# 后台启动
neo4j start

# 停止服务
neo4j stop

# 重启服务
neo4j restart

# 查看状态
neo4j status

Docker管理

# 启动容器
docker start neo4j-container

# 停止容器
docker stop neo4j-container

# 重启容器
docker restart neo4j-container

# 查看日志
docker logs -f neo4j-container

# 进入容器
docker exec -it neo4j-container bash

Python服务管理脚本

import subprocess
import time
import requests
import psutil
import os
from pathlib import Path

class Neo4jServiceManager:
    def __init__(self, neo4j_home=None, service_name="neo4j"):
        self.neo4j_home = neo4j_home or self.find_neo4j_home()
        self.service_name = service_name
        self.http_url = "http://localhost:7474"
        self.bolt_url = "bolt://localhost:7687"
    
    def find_neo4j_home(self):
        """自动查找Neo4j安装目录"""
        possible_paths = [
            "/opt/neo4j",
            "/usr/share/neo4j",
            "C:\\neo4j",
            os.path.expanduser("~/neo4j")
        ]
        
        for path in possible_paths:
            if os.path.exists(os.path.join(path, "bin", "neo4j")):
                return path
        
        return None
    
    def is_running(self):
        """检查Neo4j是否运行"""
        try:
            # 检查HTTP端口
            response = requests.get(f"{self.http_url}/", timeout=5)
            return response.status_code == 200
        except:
            return False
    
    def get_process_info(self):
        """获取Neo4j进程信息"""
        for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
            try:
                if 'java' in proc.info['name'].lower():
                    cmdline = ' '.join(proc.info['cmdline'])
                    if 'neo4j' in cmdline.lower():
                        return {
                            'pid': proc.info['pid'],
                            'memory': proc.memory_info().rss / 1024 / 1024,  # MB
                            'cpu': proc.cpu_percent(),
                            'create_time': proc.create_time()
                        }
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        return None
    
    def start_service(self, method="systemctl"):
        """启动Neo4j服务"""
        if self.is_running():
            print("Neo4j已经在运行中")
            return True
        
        print("启动Neo4j服务...")
        
        try:
            if method == "systemctl" and os.path.exists("/bin/systemctl"):
                result = subprocess.run(["sudo", "systemctl", "start", self.service_name], 
                                      capture_output=True, text=True)
            elif self.neo4j_home:
                neo4j_bin = os.path.join(self.neo4j_home, "bin", "neo4j")
                result = subprocess.run([neo4j_bin, "start"], 
                                      capture_output=True, text=True)
            else:
                result = subprocess.run(["neo4j", "start"], 
                                      capture_output=True, text=True)
            
            if result.returncode == 0:
                # 等待服务启动
                for i in range(30):
                    if self.is_running():
                        print(f"Neo4j服务启动成功 (耗时 {i+1} 秒)")
                        return True
                    time.sleep(1)
                
                print("服务启动超时")
                return False
            else:
                print(f"启动失败: {result.stderr}")
                return False
                
        except Exception as e:
            print(f"启动异常: {e}")
            return False
    
    def stop_service(self, method="systemctl"):
        """停止Neo4j服务"""
        if not self.is_running():
            print("Neo4j服务未运行")
            return True
        
        print("停止Neo4j服务...")
        
        try:
            if method == "systemctl" and os.path.exists("/bin/systemctl"):
                result = subprocess.run(["sudo", "systemctl", "stop", self.service_name], 
                                      capture_output=True, text=True)
            elif self.neo4j_home:
                neo4j_bin = os.path.join(self.neo4j_home, "bin", "neo4j")
                result = subprocess.run([neo4j_bin, "stop"], 
                                      capture_output=True, text=True)
            else:
                result = subprocess.run(["neo4j", "stop"], 
                                      capture_output=True, text=True)
            
            if result.returncode == 0:
                # 等待服务停止
                for i in range(15):
                    if not self.is_running():
                        print(f"Neo4j服务停止成功 (耗时 {i+1} 秒)")
                        return True
                    time.sleep(1)
                
                print("服务停止超时")
                return False
            else:
                print(f"停止失败: {result.stderr}")
                return False
                
        except Exception as e:
            print(f"停止异常: {e}")
            return False
    
    def restart_service(self, method="systemctl"):
        """重启Neo4j服务"""
        print("重启Neo4j服务...")
        if self.stop_service(method):
            time.sleep(2)
            return self.start_service(method)
        return False
    
    def get_status(self):
        """获取服务状态"""
        status = {
            'running': self.is_running(),
            'http_url': self.http_url,
            'bolt_url': self.bolt_url,
            'process_info': self.get_process_info()
        }
        
        if status['running']:
            try:
                # 获取数据库信息
                response = requests.get(f"{self.http_url}/db/data/", timeout=5)
                if response.status_code == 200:
                    data = response.json()
                    status['neo4j_version'] = data.get('neo4j_version')
            except:
                pass
        
        return status
    
    def print_status(self):
        """打印服务状态"""
        status = self.get_status()
        
        print("Neo4j服务状态:")
        print("=" * 40)
        print(f"运行状态: {'运行中' if status['running'] else '已停止'}")
        print(f"HTTP地址: {status['http_url']}")
        print(f"Bolt地址: {status['bolt_url']}")
        
        if status.get('neo4j_version'):
            print(f"Neo4j版本: {status['neo4j_version']}")
        
        if status['process_info']:
            proc = status['process_info']
            print(f"进程ID: {proc['pid']}")
            print(f"内存使用: {proc['memory']:.1f} MB")
            print(f"CPU使用: {proc['cpu']:.1f}%")
            
            create_time = time.strftime('%Y-%m-%d %H:%M:%S', 
                                      time.localtime(proc['create_time']))
            print(f"启动时间: {create_time}")

# 使用示例
if __name__ == "__main__":
    service_manager = Neo4jServiceManager()
    
    # 打印当前状态
    service_manager.print_status()
    
    # 启动服务
    # service_manager.start_service()
    
    # 停止服务
    # service_manager.stop_service()
    
    # 重启服务
    # service_manager.restart_service()

2.6 初始化设置

首次访问

1. 启动Neo4j服务后，访问 http://localhost:7474
2. 默认用户名: neo4j
3. 默认密码: neo4j
4. 首次登录需要修改密码

命令行初始化

# 使用cypher-shell连接
cypher-shell -u neo4j -p neo4j

# 修改密码
ALTER CURRENT USER SET PASSWORD FROM 'neo4j' TO 'newpassword';

# 退出
:exit

Python初始化脚本

from neo4j import GraphDatabase
import getpass

class Neo4jInitializer:
    def __init__(self, uri="bolt://localhost:7687"):
        self.uri = uri
        self.driver = None
    
    def connect(self, username, password):
        """连接到Neo4j"""
        try:
            self.driver = GraphDatabase.driver(self.uri, auth=(username, password))
            # 测试连接
            with self.driver.session() as session:
                result = session.run("RETURN 1 as test")
                result.single()
            print("连接成功!")
            return True
        except Exception as e:
            print(f"连接失败: {e}")
            return False
    
    def change_password(self, old_password, new_password):
        """修改密码"""
        if not self.driver:
            print("请先连接到数据库")
            return False
        
        try:
            with self.driver.session() as session:
                session.run(
                    "ALTER CURRENT USER SET PASSWORD FROM $old_password TO $new_password",
                    old_password=old_password,
                    new_password=new_password
                )
            print("密码修改成功!")
            return True
        except Exception as e:
            print(f"密码修改失败: {e}")
            return False
    
    def create_sample_data(self):
        """创建示例数据"""
        if not self.driver:
            print("请先连接到数据库")
            return False
        
        try:
            with self.driver.session() as session:
                # 清空数据库
                session.run("MATCH (n) DETACH DELETE n")
                
                # 创建示例用户
                session.run("""
                    CREATE (alice:Person {name: 'Alice', age: 30, city: 'Beijing'})
                    CREATE (bob:Person {name: 'Bob', age: 25, city: 'Shanghai'})
                    CREATE (charlie:Person {name: 'Charlie', age: 35, city: 'Guangzhou'})
                    CREATE (diana:Person {name: 'Diana', age: 28, city: 'Shenzhen'})
                """)
                
                # 创建关系
                session.run("""
                    MATCH (alice:Person {name: 'Alice'})
                    MATCH (bob:Person {name: 'Bob'})
                    MATCH (charlie:Person {name: 'Charlie'})
                    MATCH (diana:Person {name: 'Diana'})
                    
                    CREATE (alice)-[:KNOWS {since: '2020-01-01'}]->(bob)
                    CREATE (bob)-[:KNOWS {since: '2020-06-01'}]->(charlie)
                    CREATE (charlie)-[:KNOWS {since: '2021-01-01'}]->(diana)
                    CREATE (alice)-[:KNOWS {since: '2021-06-01'}]->(diana)
                """)
                
                # 创建公司和工作关系
                session.run("""
                    CREATE (techCorp:Company {name: 'TechCorp', industry: 'Technology'})
                    CREATE (dataCorp:Company {name: 'DataCorp', industry: 'Analytics'})
                    
                    MATCH (alice:Person {name: 'Alice'})
                    MATCH (bob:Person {name: 'Bob'})
                    MATCH (charlie:Person {name: 'Charlie'})
                    MATCH (diana:Person {name: 'Diana'})
                    MATCH (techCorp:Company {name: 'TechCorp'})
                    MATCH (dataCorp:Company {name: 'DataCorp'})
                    
                    CREATE (alice)-[:WORKS_FOR {position: 'Developer', since: '2019-01-01'}]->(techCorp)
                    CREATE (bob)-[:WORKS_FOR {position: 'Analyst', since: '2020-01-01'}]->(dataCorp)
                    CREATE (charlie)-[:WORKS_FOR {position: 'Manager', since: '2018-01-01'}]->(techCorp)
                    CREATE (diana)-[:WORKS_FOR {position: 'Scientist', since: '2021-01-01'}]->(dataCorp)
                """)
                
            print("示例数据创建成功!")
            return True
        except Exception as e:
            print(f"示例数据创建失败: {e}")
            return False
    
    def verify_installation(self):
        """验证安装"""
        if not self.driver:
            print("请先连接到数据库")
            return False
        
        try:
            with self.driver.session() as session:
                # 检查版本
                result = session.run("CALL dbms.components() YIELD name, versions, edition")
                for record in result:
                    print(f"组件: {record['name']}")
                    print(f"版本: {record['versions'][0]}")
                    print(f"版本: {record['edition']}")
                
                # 检查数据
                result = session.run("MATCH (n) RETURN count(n) as node_count")
                node_count = result.single()['node_count']
                
                result = session.run("MATCH ()-[r]->() RETURN count(r) as rel_count")
                rel_count = result.single()['rel_count']
                
                print(f"\n数据统计:")
                print(f"节点数量: {node_count}")
                print(f"关系数量: {rel_count}")
                
            return True
        except Exception as e:
            print(f"验证失败: {e}")
            return False
    
    def close(self):
        """关闭连接"""
        if self.driver:
            self.driver.close()

# 交互式初始化
def interactive_setup():
    print("Neo4j初始化向导")
    print("=" * 30)
    
    initializer = Neo4jInitializer()
    
    # 连接数据库
    print("\n1. 连接到Neo4j数据库")
    username = input("用户名 (默认: neo4j): ") or "neo4j"
    password = getpass.getpass("密码: ")
    
    if not initializer.connect(username, password):
        return
    
    # 修改密码（如果是默认密码）
    if password == "neo4j":
        print("\n2. 修改默认密码")
        new_password = getpass.getpass("新密码: ")
        confirm_password = getpass.getpass("确认密码: ")
        
        if new_password == confirm_password:
            if initializer.change_password(password, new_password):
                # 重新连接
                initializer.close()
                initializer = Neo4jInitializer()
                initializer.connect(username, new_password)
        else:
            print("密码不匹配，跳过密码修改")
    
    # 创建示例数据
    print("\n3. 创建示例数据")
    create_sample = input("是否创建示例数据? (y/N): ").lower() == 'y'
    
    if create_sample:
        initializer.create_sample_data()
    
    # 验证安装
    print("\n4. 验证安装")
    initializer.verify_installation()
    
    initializer.close()
    print("\n初始化完成!")
    print("您现在可以访问 http://localhost:7474 使用Neo4j Browser")

if __name__ == "__main__":
    interactive_setup()

2.7 环境验证

连接测试脚本

import sys
import time
from neo4j import GraphDatabase
import requests

def test_neo4j_installation():
    """全面测试Neo4j安装"""
    print("Neo4j安装验证")
    print("=" * 50)
    
    # 1. 测试HTTP连接
    print("\n1. 测试HTTP连接...")
    try:
        response = requests.get("http://localhost:7474", timeout=10)
        if response.status_code == 200:
            print("✓ HTTP连接正常")
        else:
            print(f"✗ HTTP连接异常: {response.status_code}")
            return False
    except Exception as e:
        print(f"✗ HTTP连接失败: {e}")
        return False
    
    # 2. 测试Bolt连接
    print("\n2. 测试Bolt连接...")
    try:
        driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
        with driver.session() as session:
            result = session.run("RETURN 'Hello Neo4j' as message")
            message = result.single()["message"]
            print(f"✓ Bolt连接正常: {message}")
        driver.close()
    except Exception as e:
        print(f"✗ Bolt连接失败: {e}")
        print("请检查用户名和密码是否正确")
        return False
    
    # 3. 测试基本操作
    print("\n3. 测试基本操作...")
    try:
        driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
        with driver.session() as session:
            # 创建测试节点
            session.run("CREATE (test:TestNode {name: 'verification', timestamp: $timestamp})", 
                       timestamp=int(time.time()))
            
            # 查询测试节点
            result = session.run("MATCH (test:TestNode {name: 'verification'}) RETURN test")
            if result.single():
                print("✓ 创建和查询操作正常")
            
            # 删除测试节点
            session.run("MATCH (test:TestNode {name: 'verification'}) DELETE test")
            print("✓ 删除操作正常")
        
        driver.close()
    except Exception as e:
        print(f"✗ 基本操作失败: {e}")
        return False
    
    # 4. 获取系统信息
    print("\n4. 系统信息...")
    try:
        driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
        with driver.session() as session:
            # 版本信息
            result = session.run("CALL dbms.components() YIELD name, versions, edition")
            for record in result:
                print(f"✓ {record['name']}: {record['versions'][0]} ({record['edition']})")
            
            # 数据库信息
            result = session.run("CALL db.info()")
            info = result.single()
            if info:
                print(f"✓ 数据库: {info.get('name', 'neo4j')}")
        
        driver.close()
    except Exception as e:
        print(f"✗ 获取系统信息失败: {e}")
    
    print("\n✓ Neo4j安装验证完成!")
    return True

if __name__ == "__main__":
    if test_neo4j_installation():
        sys.exit(0)
    else:
        sys.exit(1)

2.8 常见问题解决

内存不足

# 错误信息
# There is insufficient memory for the Java Runtime Environment to continue

# 解决方案：调整JVM内存设置
# 编辑 neo4j.conf
dbms.memory.heap.initial_size=512m
dbms.memory.heap.max_size=1G

端口冲突

# 错误信息
# Address already in use

# 查找占用端口的进程
sudo netstat -tlnp | grep :7474
sudo netstat -tlnp | grep :7687

# 修改端口配置
dbms.connector.http.listen_address=:8474
dbms.connector.bolt.listen_address=:8687

权限问题

# 设置正确的文件权限
sudo chown -R neo4j:neo4j /var/lib/neo4j
sudo chown -R neo4j:neo4j /var/log/neo4j
sudo chmod -R 755 /var/lib/neo4j

Python故障排除脚本

import os
import subprocess
import socket
import psutil

class Neo4jTroubleshooter:
    def __init__(self):
        self.issues = []
        self.solutions = []
    
    def check_java_installation(self):
        """检查Java安装"""
        try:
            result = subprocess.run(["java", "-version"], 
                                  capture_output=True, text=True)
            if result.returncode == 0:
                print("✓ Java已安装")
                return True
            else:
                self.issues.append("Java未安装")
                self.solutions.append("请安装Java 11或更高版本")
                return False
        except FileNotFoundError:
            self.issues.append("Java未找到")
            self.solutions.append("请安装Java并添加到PATH")
            return False
    
    def check_port_availability(self, port):
        """检查端口是否可用"""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            sock.bind(('localhost', port))
            sock.close()
            return True
        except OSError:
            return False
    
    def check_ports(self):
        """检查Neo4j端口"""
        ports = [7474, 7687]
        for port in ports:
            if not self.check_port_availability(port):
                self.issues.append(f"端口 {port} 被占用")
                self.solutions.append(f"释放端口 {port} 或修改Neo4j配置使用其他端口")
    
    def check_memory(self):
        """检查系统内存"""
        memory = psutil.virtual_memory()
        available_gb = memory.available / (1024**3)
        
        if available_gb < 1:
            self.issues.append(f"可用内存不足: {available_gb:.1f}GB")
            self.solutions.append("释放内存或调整Neo4j内存配置")
        else:
            print(f"✓ 可用内存: {available_gb:.1f}GB")
    
    def check_disk_space(self):
        """检查磁盘空间"""
        disk = psutil.disk_usage('/')
        free_gb = disk.free / (1024**3)
        
        if free_gb < 5:
            self.issues.append(f"磁盘空间不足: {free_gb:.1f}GB")
            self.solutions.append("清理磁盘空间")
        else:
            print(f"✓ 可用磁盘空间: {free_gb:.1f}GB")
    
    def check_neo4j_process(self):
        """检查Neo4j进程"""
        for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
            try:
                if 'java' in proc.info['name'].lower():
                    cmdline = ' '.join(proc.info['cmdline'])
                    if 'neo4j' in cmdline.lower():
                        print(f"✓ Neo4j进程运行中 (PID: {proc.info['pid']})")
                        return True
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        
        self.issues.append("Neo4j进程未运行")
        self.solutions.append("启动Neo4j服务")
        return False
    
    def run_diagnosis(self):
        """运行完整诊断"""
        print("Neo4j故障诊断")
        print("=" * 40)
        
        self.check_java_installation()
        self.check_memory()
        self.check_disk_space()
        self.check_ports()
        self.check_neo4j_process()
        
        if self.issues:
            print("\n发现问题:")
            print("-" * 20)
            for i, (issue, solution) in enumerate(zip(self.issues, self.solutions), 1):
                print(f"{i}. 问题: {issue}")
                print(f"   解决方案: {solution}")
                print()
        else:
            print("\n✓ 未发现问题")
        
        return len(self.issues) == 0

if __name__ == "__main__":
    troubleshooter = Neo4jTroubleshooter()
    troubleshooter.run_diagnosis()

2.9 章节总结

核心知识点

1. 系统要求：Java 11+，足够的内存和磁盘空间
2. 安装方式：官方包、包管理器、Docker
3. 配置文件：neo4j.conf的关键配置项
4. 服务管理：启动、停止、重启、状态检查
5. 初始化：密码修改、示例数据创建
6. 故障排除：常见问题和解决方案

最佳实践

1. 生产环境：使用包管理器安装，配置系统服务
2. 开发环境：Docker部署，便于版本管理
3. 内存配置：根据数据量和查询复杂度调整
4. 安全配置：修改默认密码，限制网络访问
5. 监控配置：启用查询日志，配置性能监控

练习题

1. 在不同操作系统上安装Neo4j并配置集群
2. 编写自动化部署脚本
3. 配置Neo4j的内存和性能参数
4. 实现Neo4j服务的健康检查
5. 创建Neo4j的备份和恢复策略

---

下一章预告：在下一章中，我们将学习Cypher查询语言的基础语法和常用操作。