Neo4j 在 Go 生态中的基本使用

_neo4j_go.md

使用 Neo4j 官方包，处理查询

go.mod

module Neo4j

go 1.21.12

require (
    github.com/neo4j/neo4j-go-driver/v5 v5.28.0
)

main.go

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/neo4j/neo4j-go-driver/v5/neo4j"
)

// Neo4jPool 连接池结构
type Neo4jPool struct {
	driver neo4j.DriverWithContext
	pool   chan neo4j.SessionWithContext
	size   int
}

// Neo4jConfig 配置结构
type Neo4jConfig struct {
	URI      string // "bolt://localhost:7687"
	Username string
	Password string
	MaxConn  int
}

// NewNeo4jPool 创建新的连接池
func NewNeo4jPool(config *Neo4jConfig) (*Neo4jPool, error) {
	// auth := neo4j.BasicAuth(config.Username, config.Password, "")
	driver, err := neo4j.NewDriverWithContext(config.URI, neo4j.NoAuth())
	if err != nil {
		return nil, fmt.Errorf("failed to create driver: %v", err)
	}

	pool := &Neo4jPool{
		driver: driver,
		pool:   make(chan neo4j.SessionWithContext, config.MaxConn),
		size:   config.MaxConn,
	}

	// 初始化连接池
	for i := 0; i < pool.size; i++ {
		session := driver.NewSession(context.Background(), neo4j.SessionConfig{})
		pool.pool <- session
	}

	return pool, nil
}

// GetSession 从连接池获取会话
func (p *Neo4jPool) GetSession() neo4j.SessionWithContext {
	return <-p.pool
}

// ReleaseSession 释放会话回连接池
func (p *Neo4jPool) ReleaseSession(session neo4j.SessionWithContext) {
	p.pool <- session
}

// Close 关闭连接池
func (p *Neo4jPool) Close() {
	ctx := context.Background()
	close(p.pool)
	for session := range p.pool {
		session.Close(ctx)
	}
	p.driver.Close(ctx)
}

// ExecuteInTransaction 在事务中执行查询
func (p *Neo4jPool) ExecuteInTransaction(ctx context.Context, queries []string, params []map[string]interface{}) error {
	session := p.GetSession()
	defer p.ReleaseSession(session)

	_, err := session.ExecuteWrite(ctx, func(tx neo4j.ManagedTransaction) (interface{}, error) {
		for i, query := range queries {
			result, err := tx.Run(ctx, query, params[i])
			if err != nil {
				return nil, fmt.Errorf("failed to execute query %d: %v", i, err)
			}
			_, err = result.Consume(ctx)
			if err != nil {
				return nil, fmt.Errorf("failed to consume result %d: %v", i, err)
			}
		}
		return nil, nil
	})

	return err
}

// QueryWithRetry 带重试的查询执行
func (p *Neo4jPool) QueryWithRetry(ctx context.Context, queries []string, params []map[string]interface{}, maxRetries int) error {
	var err error
	for i := 0; i < maxRetries; i++ {
		err = p.ExecuteInTransaction(ctx, queries, params)
		if err == nil {
			return nil
		}

		// 检查是否需要重试
		if !isRetryableError(err) {
			return err
		}

		log.Printf("Retry %d: %v", i+1, err)
		time.Sleep(time.Second * time.Duration(i+1))
	}

	return fmt.Errorf("max retries reached: %v", err)
}

// isRetryableError 判断错误是否需要重试
func isRetryableError(err error) bool {
	// 这里可以根据实际需求添加更多的错误类型判断
	return err != nil
}

func main() {
	// 配置
	config := &Neo4jConfig{
		URI:      "bolt://localhost:7687",
		Username: "neo4j",
		Password: "password",
		MaxConn:  10,
	}

	// 创建连接池
	pool, err := NewNeo4jPool(config)
	if err != nil {
		log.Fatalf("Failed to create pool: %v", err)
	}
	defer pool.Close()

	// 准备查询和参数
	teamUUID := "Team1"
	queries := []string{
		"CREATE (u:" + teamUUID + ":User {name: $name, age: $age})",
		"MATCH (u:" + teamUUID + ":User {name: $name}) CREATE (u)-[:FOLLOWS]->(f:User {name: $friendName})",
	}
	params := []map[string]interface{}{
		{
			"name": "Alice",
			"age":  25,
		},
		{
			"name":       "Alice",
			"friendName": "Bob",
		},
	}

	// 执行查询
	ctx := context.Background()
	err = pool.QueryWithRetry(ctx, queries, params, 3)
	if err != nil {
		log.Printf("Failed to execute queries: %v", err)
		return
	}

	log.Println("Queries executed successfully")
}

// 使用示例：批量创建用户和关系
func createUsersAndRelationships(pool *Neo4jPool, users []map[string]interface{}) error {
	queries := make([]string, 0, len(users)*2)
	params := make([]map[string]interface{}, 0, len(users)*2)

	for _, user := range users {
		// 创建用户节点
		queries = append(queries, "CREATE (u:User {name: $name, age: $age})")
		params = append(params, map[string]interface{}{
			"name": user["name"],
			"age":  user["age"],
		})

		// 创建关系
		if friendName, ok := user["friend"].(string); ok {
			queries = append(queries, "MATCH (u:User {name: $name}) CREATE (u)-[:FOLLOWS]->(f:User {name: $friendName})")
			params = append(params, map[string]interface{}{
				"name":       user["name"],
				"friendName": friendName,
			})
		}
	}

	return pool.QueryWithRetry(context.Background(), queries, params, 3)
}

run.bash

go get Neo4j
go run main.go