运行并发事务

package main

import (
    "fmt"
    "context"
    "time"
    "sync"
    "github.com/neo4j/neo4j-go-driver/v6/neo4j"
)

type item struct {
    record *neo4j.Record
    err    error
}

func main() {
    ctx := context.Background()

    // Connection to database
    dbUri := "<database-uri>"
    dbUser := "<username>"
    dbPassword := "<password>"
    driver, err := neo4j.NewDriver(
        dbUri,
        neo4j.BasicAuth(dbUser, dbPassword, ""))
    if err != nil {
        panic(err)
    }
    defer driver.Close(ctx)
    err = driver.VerifyConnectivity(ctx)
    if err != nil {
        panic(err)
    }

    // Run a query and get results in a channel
    recordsC := queryToChannel(ctx, driver)  (1)

    // Spawn some consumers that will process records
    // They communicate back on the log channel
    // WaitGroup allows to keep track of progress and close channel when all are done
    log := make(chan string)  (4)
    wg := &sync.WaitGroup{}  (5)
    for i := 1; i < 10; i++ {  // i starts from 1 because 0th receiver would process too fast
        wg.Add(1)
        go consumer(wg, recordsC, log, i)  (6)
    }
    // When all consumers are done, close log channel
    go func() {
        wg.Wait()
        close(log)
    }()
    // Print log as it comes
    for v := range log {
        fmt.Println(v)
    }
}

func queryToChannel(ctx context.Context, driver neo4j.DriverWithContext) chan item {
    recordsC := make(chan item, 10)  (2)
    go func() {
        defer close(recordsC)
        session := driver.NewSession(ctx, neo4j.SessionConfig{DatabaseName: "<database-name>"})
        defer session.Close(ctx)
        // Cypher query to create and retrieve some nodes
        result, err := session.Run(ctx, `
            UNWIND range(1,25) AS id
            MERGE (p:Person {id: id})
            RETURN p
            `, nil)
        if err != nil {
            var it item
            it.record, it.err = nil, err
            recordsC <- it
        } else {
            // Stream results to channel as they come from the server
            for result.Next(ctx) { (3)
                record := result.Record()
                var it item
                it.record, it.err = record, nil
                recordsC <- it
            }
            // If the stream interrupts, send the error through the channel
            if result.Err() != nil {
                var it item
                it.record, it.err = nil, err
                recordsC <- it
            }
        }
    }()
    return recordsC
}

func consumer(wg *sync.WaitGroup, records <-chan item, log chan string, n int) {
    defer wg.Done()  // will communicate that routine is done
    for it := range records {
        if it.err != nil {
            fmt.Println("ERROR:", it.err)
        } else {
            log <- fmt.Sprintf("Receiver %v processed %v", n, it.record)
            time.Sleep(time.Duration(n) * time.Second)  // proxy for a time-consuming processing
        }
    }
}

package main

import (
    "fmt"
    "context"
    "sync"
    "github.com/neo4j/neo4j-go-driver/v6/neo4j"
)

func main() {
    ctx := context.Background()

    // Connection to database
    dbUri := "<database-uri>"
    dbUser := "<username>"
    dbPassword := "<password>"
    driver, err := neo4j.NewDriver(
        dbUri,
        neo4j.BasicAuth(dbUser, dbPassword, ""))
    if err != nil {
        panic(err)
    }
    defer driver.Close(ctx)
    err = driver.VerifyConnectivity(ctx)
    if err != nil {
        panic(err)
    }

    log := make(chan string)  (1)
    wg := &sync.WaitGroup{}  (2)
    // Spawn 10 concurrent queries
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go runQuery(wg, ctx, driver, log)  (3)
    }
    // Wait for all runner routines to be done before closing log
    go func() {
        wg.Wait()
        close(log)
    }()
    // Print log
    for msg := range log {
        fmt.Println(msg)
    }
}

// Run Neo4j query with random sleep time, returning the sleep time in ms
func runQuery(wg *sync.WaitGroup, ctx context.Context, driver neo4j.Driver, log chan string) {
    defer wg.Done()  // will communicate that routine is done
    result, err := neo4j.ExecuteQuery(ctx, driver, `
        WITH round(rand()*2000) AS waitTime
        CALL apoc.util.sleep(toInteger(waitTime)) RETURN waitTime AS time
        `, nil, neo4j.EagerResultTransformer,
        neo4j.ExecuteQueryWithDatabase("<database-name>"))
    if err != nil {
        log <- fmt.Sprintf("ERROR: %v", err)
    } else {
        neo, _ := result.Records[0].Get("time")
        log <- fmt.Sprintf("Query returned %v", neo)
    }
}