Go语言数据库操作深入讲解

go操作MySQL

使用第三方开源的mysql库: github.com/go-sql-driver/mysql (mysql驱动)
github.com/jmoiron/sqlx (基于mysql驱动的封装)

命令行输入：

go get github.com/go-sql-driver/mysql
go get github.com/jmoiron/sqlx

Insert操作

// 连接Mysql
database, err := sqlx.Open("mysql", "root:XXXX@tcp(127.0.0.1:3306)/test")
//database, err := sqlx.Open("数据库类型", "用户名:密码@tcp(地址:端口)/数据库名") 
package main
import (
    "fmt"
    _ "github.com/go-sql-driver/mysql"
    "github.com/jmoiron/sqlx"
)

type Person struct {
    UserId   int    `db:"user_id"`
    Username string `db:"username"`
    Sex      string `db:"sex"`
    Email    string `db:"email"`
}

type Place struct {
    Country string `db:"country"`
    City    string `db:"city"`
    TelCode int    `db:"telcode"`
}

var Db *sqlx.DB

func init() {
    database, err := sqlx.Open("mysql", "root:root@tcp(127.0.0.1:3306)/test")
    if err != nil {
        fmt.Println("open mysql failed,", err)
        return
    }
    Db = database
    defer db.Close()  // 注意这行代码要写在上面err判断的下面
}

func main() {
    r, err := Db.Exec("insert into person(username, sex, email)values(?, ?, ?)", "stu001", "man", "stu01@qq.com")
    if err != nil {
        fmt.Println("exec failed, ", err)
        return
    }
    id, err := r.LastInsertId()
    if err != nil {
        fmt.Println("exec failed, ", err)
        return
    }

    fmt.Println("insert succ:", id)
}

Select操作

package main

import (
    "fmt"

    _ "github.com/go-sql-driver/mysql"
    "github.com/jmoiron/sqlx"
)

type Person struct {
    UserId   int    `db:"user_id"`
    Username string `db:"username"`
    Sex      string `db:"sex"`
    Email    string `db:"email"`
}

type Place struct {
    Country string `db:"country"`
    City    string `db:"city"`
    TelCode int    `db:"telcode"`
}

var Db *sqlx.DB

func init() {

    database, err := sqlx.Open("mysql", "root:root@tcp(127.0.0.1:3306)/test")
    if err != nil {
        fmt.Println("open mysql failed,", err)
        return
    }

    Db = database
    defer db.Close()  // 注意这行代码要写在上面err判断的下面
}

func main() {

    var person []Person
    err := Db.Select(&person, "select user_id, username, sex, email from person where user_id=?", 1)
    if err != nil {
        fmt.Println("exec failed, ", err)
        return
    }

    fmt.Println("select succ:", person)
}

Update操作

package main

import (
    "fmt"

    _ "github.com/go-sql-driver/mysql"
    "github.com/jmoiron/sqlx"
)

type Person struct {
    UserId   int    `db:"user_id"`
    Username string `db:"username"`
    Sex      string `db:"sex"`
    Email    string `db:"email"`
}

type Place struct {
    Country string `db:"country"`
    City    string `db:"city"`
    TelCode int    `db:"telcode"`
}

var Db *sqlx.DB

func init() {

    database, err := sqlx.Open("mysql", "root:root@tcp(127.0.0.1:3306)/test")
    if err != nil {
        fmt.Println("open mysql failed,", err)
        return
    }

    Db = database
    defer db.Close()  // 注意这行代码要写在上面err判断的下面
}

func main() {

    res, err := Db.Exec("update person set username=? where user_id=?", "stu0003", 1)
    if err != nil {
        fmt.Println("exec failed, ", err)
        return
    }
    row, err := res.RowsAffected()
    if err != nil {
        fmt.Println("rows failed, ",err)
    }
    fmt.Println("update succ:",row)

}

Delete操作

package main

import (
    "fmt"

    _ "github.com/go-sql-driver/mysql"
    "github.com/jmoiron/sqlx"
)

type Person struct {
    UserId   int    `db:"user_id"`
    Username string `db:"username"`
    Sex      string `db:"sex"`
    Email    string `db:"email"`
}

type Place struct {
    Country string `db:"country"`
    City    string `db:"city"`
    TelCode int    `db:"telcode"`
}

var Db *sqlx.DB

func init() {

    database, err := sqlx.Open("mysql", "root:root@tcp(127.0.0.1:3306)/test")
    if err != nil {
        fmt.Println("open mysql failed,", err)
        return
    }

    Db = database
    defer db.Close()  // 注意这行代码要写在上面err判断的下面
}

func main() {

    /*
    _, err := Db.Exec("delete from person where user_id=?", 1)
    if err != nil {
        fmt.Println("exec failed, ", err)
        return
    }
    */

    res, err := Db.Exec("delete from person where user_id=?", 1)
    if err != nil {
        fmt.Println("exec failed, ", err)
        return
    }

    row,err := res.RowsAffected()
    if err != nil {
        fmt.Println("rows failed, ",err)
    }

    fmt.Println("delete succ: ",row)
}

MySQL事务

package main

    import (
        "fmt"

        _ "github.com/go-sql-driver/mysql"
        "github.com/jmoiron/sqlx"
    )

    type Person struct {
        UserId   int    `db:"user_id"`
        Username string `db:"username"`
        Sex      string `db:"sex"`
        Email    string `db:"email"`
    }

    type Place struct {
        Country string `db:"country"`
        City    string `db:"city"`
        TelCode int    `db:"telcode"`
    }

    var Db *sqlx.DB

    func init() {
        database, err := sqlx.Open("mysql", "root:root@tcp(127.0.0.1:3306)/test")
        if err != nil {
            fmt.Println("open mysql failed,", err)
            return
        }
        Db = database
    }

    func main() {
        conn, err := Db.Begin()
        if err != nil {
            fmt.Println("begin failed :", err)
            return
        }

        r, err := conn.Exec("insert into person(username, sex, email)values(?, ?, ?)", "stu001", "man", "stu01@qq.com")
        if err != nil {
            fmt.Println("exec failed, ", err)
            conn.Rollback()
            return
        }
        id, err := r.LastInsertId()
        if err != nil {
            fmt.Println("exec failed, ", err)
            conn.Rollback()
            return
        }
        fmt.Println("insert succ:", id)

        r, err = conn.Exec("insert into person(username, sex, email)values(?, ?, ?)", "stu001", "man", "stu01@qq.com")
        if err != nil {
            fmt.Println("exec failed, ", err)
            conn.Rollback()
            return
        }
        id, err = r.LastInsertId()
        if err != nil {
            fmt.Println("exec failed, ", err)
            conn.Rollback()
            return
        }
        fmt.Println("insert succ:", id)

        conn.Commit()
    }

go操作Redis

使用第三方开源的redis库: github.com/garyburd/redigo/redis: go get github.com/garyburd/redigo/redis

Redis连接

package main

import (
    "fmt"
    "github.com/garyburd/redigo/redis"
)

func main() {
    c, err := redis.Dial("tcp", "localhost:6379")
    if err != nil {
        fmt.Println("conn redis failed,", err)
        return
    } 

    fmt.Println("redis conn success")

    defer c.Close()
}

String类型Set、Get操作

package main

import (
    "fmt"
    "github.com/garyburd/redigo/redis"
)

func main() {
    c, err := redis.Dial("tcp", "localhost:6379")
    if err != nil {
        fmt.Println("conn redis failed,", err)
        return
    }

    defer c.Close()
    _, err = c.Do("Set", "abc", 100)
    if err != nil {
        fmt.Println(err)
        return
    }

    r, err := redis.Int(c.Do("Get", "abc"))
    if err != nil {
        fmt.Println("get abc failed,", err)
        return
    }

    fmt.Println(r)
}

String批量操作

package main

import (
    "fmt"
    "github.com/garyburd/redigo/redis"
)

func main() {
    c, err := redis.Dial("tcp", "localhost:6379")
    if err != nil {
        fmt.Println("conn redis failed,", err)
        return
    }

    defer c.Close()
    _, err = c.Do("MSet", "abc", 100, "efg", 300)
    if err != nil {
        fmt.Println(err)
        return
    }

    r, err := redis.Ints(c.Do("MGet", "abc", "efg"))
    if err != nil {
        fmt.Println("get abc failed,", err)
        return
    }

    for _, v := range r {
        fmt.Println(v)
    }
}

设置过期时间

package main

import (
    "fmt"
    "github.com/garyburd/redigo/redis"
)

func main() {
    c, err := redis.Dial("tcp", "localhost:6379")
    if err != nil {
        fmt.Println("conn redis failed,", err)
        return
    }

    defer c.Close()
    _, err = c.Do("expire", "abc", 10)
    if err != nil {
        fmt.Println(err)
        return
    }
}

List队列操作

package main

import (
    "fmt"
    "github.com/garyburd/redigo/redis"
)

func main() {
    c, err := redis.Dial("tcp", "localhost:6379")
    if err != nil {
        fmt.Println("conn redis failed,", err)
        return
    }

    defer c.Close()
    _, err = c.Do("lpush", "book_list", "abc", "ceg", 300)
    if err != nil {
        fmt.Println(err)
        return
    }

    r, err := redis.String(c.Do("lpop", "book_list"))
    if err != nil {
        fmt.Println("get abc failed,", err)
        return
    }

    fmt.Println(r)
}

Hash表

package main

import (
    "fmt"
    "github.com/garyburd/redigo/redis"
)

func main() {
    c, err := redis.Dial("tcp", "localhost:6379")
    if err != nil {
        fmt.Println("conn redis failed,", err)
        return
    }

    defer c.Close()
    _, err = c.Do("HSet", "books", "abc", 100)
    if err != nil {
        fmt.Println(err)
        return
    }

    r, err := redis.Int(c.Do("HGet", "books", "abc"))
    if err != nil {
        fmt.Println("get abc failed,", err)
        return
    }

    fmt.Println(r)
}

Redis连接池

package main
import(
    "fmt"
    "github.com/garyburd/redigo/redis"
)

var pool *redis.Pool  //创建redis连接池

func init(){
    pool = &redis.Pool{     //实例化一个连接池
        MaxIdle:16,    //最初的连接数量
        // MaxActive:1000000,    //最大连接数量
        MaxActive:0,    //连接池最大连接数量,不确定可以用0(0表示自动定义)，按需分配
        IdleTimeout:300,    //连接关闭时间 300秒 (300秒不使用自动关闭)    
        Dial: func() (redis.Conn ,error){     //要连接的redis数据库
            return redis.Dial("tcp","localhost:6379")
        },
    }
}

func main(){
  c := pool.Get() //从连接池，取一个链接
  defer c.Close() //函数运行结束 ，把连接放回连接池
    _,err := c.Do("Set","abc",200)
    if err != nil {
        fmt.Println(err)
        return
    }

    r,err := redis.Int(c.Do("Get","abc"))
    if err != nil {
        fmt.Println("get abc faild :",err)
        return
    }
    fmt.Println(r)
    pool.Close() //关闭连接池
}

go操作ETCD

etcd是使用Go语言开发的一个开源的、高可用的分布式key-value存储系统，可以用于配置共享和服务的注册和发现。

类似项目有zookeeper和consul。

etcd具有以下特点：

完全复制：集群中的每个节点都可以使用完整的存档
高可用性：Etcd可用于避免硬件的单点故障或网络问题
一致性：每次读取都会返回跨多主机的最新写入
简单：包括一个定义良好、面向用户的API(gRPC)
安全：实现了带有可选的客户端证书身份验证的自动化TLS
快速：每秒10000次写入的基准速度
可靠：使用Raft算法实现了强一致、高可用的服务存储目录

etcd应用场景

服务发现

服务发现要解决的也是分布式系统中最常见的问题之一，即在同一个分布式集群中的进程或服务，要如何才能找到对方并建立连接。本质上来说，服务发现就是想要了解集群中是否有进程在监听 udp 或 tcp 端口，并且通过名字就可以查找和连接。

配置中心

将一些配置信息放到 etcd 上进行集中管理。

这类场景的使用方式通常是这样：应用在启动的时候主动从 etcd 获取一次配置信息，同时，在 etcd 节点上注册一个 Watcher 并等待，以后每次配置有更新的时候，etcd 都会实时通知订阅者，以此达到获取最新配置信息的目的。

分布式锁

因为 etcd 使用 Raft 算法保持了数据的强一致性，某次操作存储到集群中的值必然是全局一致的，所以很容易实现分布式锁。锁服务有两种使用方式，一是保持独占，二是控制时序。

保持独占即所有获取锁的用户最终只有一个可以得到。etcd 为此提供了一套实现分布式锁原子操作 CAS(CompareAndSwap)的 API。通过设置prevExist值，可以保证在多个节点同时去创建某个目录时，只有一个成功。而创建成功的用户就可以认为是获得了锁。
控制时序，即所有想要获得锁的用户都会被安排执行，但是获得锁的顺序也是全局唯一的，同时决定了执行顺序。etcd 为此也提供了一套 API(自动创建有序键)，对一个目录建值时指定为POST动作，这样 etcd 会自动在目录下生成一个当前最大的值为键，存储这个新的值(客户端编号)。同时还可以使用 API 按顺序列出所有当前目录下的键值。此时这些键的值就是客户端的时序，而这些键中存储的值可以是代表客户端的编号。

为什么用 etcd 而不用ZooKeeper？

etcd 实现的这些功能，ZooKeeper都能实现。那么为什么要用 etcd 而非直接使用ZooKeeper呢？

为什么不选择ZooKeeper？

部署维护复杂，其使用的Paxos强一致性算法复杂难懂。官方只提供了Java和C两种语言的接口。
使用Java编写引入大量的依赖。运维人员维护起来比较麻烦。
最近几年发展缓慢，不如etcd和consul等后起之秀。

为什么选择etcd？

简单。使用 Go 语言编写部署简单；支持HTTP/JSON API,使用简单；使用 Raft 算法保证强一致性让用户易于理解。
etcd 默认数据一更新就进行持久化。
etcd 支持 SSL 客户端安全认证。

最后，etcd 作为一个年轻的项目，正在高速迭代和开发中，这既是一个优点，也是一个缺点。优点是它的未来具有无限的可能性，缺点是无法得到大项目长时间使用的检验。然而，目前 CoreOS、Kubernetes和CloudFoundry等知名项目均在生产环境中使用了etcd，所以总的来说，etcd值得你去尝试。

etcd集群

etcd 作为一个高可用键值存储系统，天生就是为集群化而设计的。由于 Raft 算法在做决策时需要多数节点的投票，所以 etcd 一般部署集群推荐奇数个节点，推荐的数量为 3、5 或者 7 个节点构成一个集群。

操作ETCD

put和get操作

// put命令用来设置键值对数据，get命令用来根据key获取值。
package main

import (
    "context"
    "fmt"
    "time"

    "go.etcd.io/etcd/clientv3"
)

// etcd client put/get demo
// use etcd/clientv3

func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        // handle error!
        fmt.Printf("connect to etcd failed, err:%v\n", err)
        return
    }
    fmt.Println("connect to etcd success")
    defer cli.Close()
    // put
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    _, err = cli.Put(ctx, "lmh", "lmh")
    cancel()
    if err != nil {
        fmt.Printf("put to etcd failed, err:%v\n", err)
        return
    }
    // get
    ctx, cancel = context.WithTimeout(context.Background(), time.Second)
    resp, err := cli.Get(ctx, "lmh")
    cancel()
    if err != nil {
        fmt.Printf("get from etcd failed, err:%v\n", err)
        return
    }
    for _, ev := range resp.Kvs {
        fmt.Printf("%s:%s\n", ev.Key, ev.Value)
    }
}

watch操作

// watch用来获取未来更改的通知。
func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        fmt.Printf("connect to etcd failed, err:%v\n", err)
        return
    }
    fmt.Println("connect to etcd success")
    defer cli.Close()
    // watch key:lmh change
    rch := cli.Watch(context.Background(), "lmh") // <-chan WatchResponse
    for wresp := range rch {
        for _, ev := range wresp.Events {
            fmt.Printf("Type: %s Key:%s Value:%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
        }
    }
}

lease租约

package main

import (
    "fmt"
    "time"
)

// etcd lease

import (
    "context"
    "log"

    "go.etcd.io/etcd/clientv3"
)

func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: time.Second * 5,
    })
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("connect to etcd success.")
    defer cli.Close()

    // 创建一个5秒的租约
    resp, err := cli.Grant(context.TODO(), 5)
    if err != nil {
        log.Fatal(err)
    }

    // 5秒钟之后, /lmh/ 这个key就会被移除
    _, err = cli.Put(context.TODO(), "/lmh/", "lmh", clientv3.WithLease(resp.ID))
    if err != nil {
        log.Fatal(err)
    }
}

keepAlive

package main

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

    "go.etcd.io/etcd/clientv3"
)

// etcd keepAlive

func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: time.Second * 5,
    })
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("connect to etcd success.")
    defer cli.Close()

    resp, err := cli.Grant(context.TODO(), 5)
    if err != nil {
        log.Fatal(err)
    }

    _, err = cli.Put(context.TODO(), "/lmh/", "lmh", clientv3.WithLease(resp.ID))
    if err != nil {
        log.Fatal(err)
    }

    // the key 'foo' will be kept forever
    ch, kaerr := cli.KeepAlive(context.TODO(), resp.ID)
    if kaerr != nil {
        log.Fatal(kaerr)
    }
    for {
        ka := <-ch
        fmt.Println("ttl:", ka.TTL)
    }
}

基于etcd实现分布式锁

cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
if err != nil {
    log.Fatal(err)
}
defer cli.Close()

// 创建两个单独的会话用来演示锁竞争
s1, err := concurrency.NewSession(cli)
if err != nil {
    log.Fatal(err)
}
defer s1.Close()
m1 := concurrency.NewMutex(s1, "/my-lock/")

s2, err := concurrency.NewSession(cli)
if err != nil {
    log.Fatal(err)
}
defer s2.Close()
m2 := concurrency.NewMutex(s2, "/my-lock/")

// 会话s1获取锁
if err := m1.Lock(context.TODO()); err != nil {
    log.Fatal(err)
}
fmt.Println("acquired lock for s1")

m2Locked := make(chan struct{})
go func() {
    defer close(m2Locked)
    // 等待直到会话s1释放了/my-lock/的锁
    if err := m2.Lock(context.TODO()); err != nil {
        log.Fatal(err)
    }
}()

if err := m1.Unlock(context.TODO()); err != nil {
    log.Fatal(err)
}
fmt.Println("released lock for s1")

<-m2Locked
fmt.Println("acquired lock for s2")

zookeeper

简单的分布式server

目前分布式系统已经很流行了，一些开源框架也被广泛应用，如dubbo、Motan等。对于一个分布式服务，最基本的一项功能就是服务的注册和发现，而利用zk的EPHEMERAL节点则可以很方便的实现该功能。EPHEMERAL节点正如其名，是临时性的，其生命周期是和客户端会话绑定的，当会话连接断开时，节点也会被删除。下边我们就来实现一个简单的分布式server：

server：

服务启动时，创建zk连接，并在go_servers节点下创建一个新节点，节点名为”ip:port”，完成服务注册
服务结束时，由于连接断开，创建的节点会被删除，这样client就不会连到该节点

client：

先从zk获取go_servers节点下所有子节点，这样就拿到了所有注册的server
从server列表中选中一个节点(这里只是随机选取，实际服务一般会提供多种策略)，创建连接进行通信
这里为了演示，我们每次client连接server，获取server发送的时间后就断开。主要代码如下：

// server.go
package main

import (
    "fmt"
    "net"
    "os"
    "time"

    "github.com/samuel/go-zookeeper/zk"
)

func main() {
    go starServer("127.0.0.1:8897")
    go starServer("127.0.0.1:8898")
    go starServer("127.0.0.1:8899")

    a := make(chan bool, 1)
    <-a
}

func checkError(err error) {
    if err != nil {
        fmt.Println(err)
    }
}

func starServer(port string) {
    tcpAddr, err := net.ResolveTCPAddr("tcp4", port)
    fmt.Println(tcpAddr)
    checkError(err)

    listener, err := net.ListenTCP("tcp", tcpAddr)
    checkError(err)

    //注册zk节点q
    // 链接zk
    conn, err := GetConnect()
    if err != nil {
        fmt.Printf(" connect zk error: %s ", err)
    }
    defer conn.Close()
    // zk节点注册
    err = RegistServer(conn, port)
    if err != nil {
        fmt.Printf(" regist node error: %s ", err)
    }

    for {
        conn, err := listener.Accept()
        if err != nil {
            fmt.Fprintf(os.Stderr, "Error: %s", err)
            continue
        }
        go handleCient(conn, port)
    }

    fmt.Println("aaaaaa")
}

func handleCient(conn net.Conn, port string) {
    defer conn.Close()

    daytime := time.Now().String()
    conn.Write([]byte(port + ": " + daytime))
}
func GetConnect() (conn *zk.Conn, err error) {
    zkList := []string{"localhost:2181"}
    conn, _, err = zk.Connect(zkList, 10*time.Second)
    if err != nil {
        fmt.Println(err)
    }
    return
}

func RegistServer(conn *zk.Conn, host string) (err error) {
    _, err = conn.Create("/go_servers/"+host, nil, zk.FlagEphemeral, zk.WorldACL(zk.PermAll))
    return
}

func GetServerList(conn *zk.Conn) (list []string, err error) {
    list, _, err = conn.Children("/go_servers")
    return
}

// client.go
package main

import (
    "errors"
    "fmt"
    "io/ioutil"
    "math/rand"
    "net"
    "time"

    "github.com/samuel/go-zookeeper/zk"
)

func checkError(err error) {
    if err != nil {
        fmt.Println(err)
    }
}
func main() {
    for i := 0; i < 100; i++ {
        startClient()

        time.Sleep(1 * time.Second)
    }
}

func startClient() {
    // service := "127.0.0.1:8899"
    //获取地址
    serverHost, err := getServerHost()
    if err != nil {
        fmt.Printf("get server host fail: %s \n", err)
        return
    }

    fmt.Println("connect host: " + serverHost)
    tcpAddr, err := net.ResolveTCPAddr("tcp4", serverHost)
    checkError(err)
    conn, err := net.DialTCP("tcp", nil, tcpAddr)
    checkError(err)
    defer conn.Close()

    _, err = conn.Write([]byte("timestamp"))
    checkError(err)

    result, err := ioutil.ReadAll(conn)
    checkError(err)
    fmt.Println(string(result))

    return
}

func getServerHost() (host string, err error) {
    conn, err := GetConnect()
    if err != nil {
        fmt.Printf(" connect zk error: %s \n ", err)
        return
    }
    defer conn.Close()
    serverList, err := GetServerList(conn)
    if err != nil {
        fmt.Printf(" get server list error: %s \n", err)
        return
    }

    count := len(serverList)
    if count == 0 {
        err = errors.New("server list is empty \n")
        return
    }

    //随机选中一个返回
    r := rand.New(rand.NewSource(time.Now().UnixNano()))
    host = serverList[r.Intn(3)]
    return
}
func GetConnect() (conn *zk.Conn, err error) {
    zkList := []string{"localhost:2181"}
    conn, _, err = zk.Connect(zkList, 10*time.Second)
    if err != nil {
        fmt.Println(err)
    }
    return
}
func GetServerList(conn *zk.Conn) (list []string, err error) {
    list, _, err = conn.Children("/go_servers")
    return
}

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.
149.
150.
151.
152.
153.
154.
155.
156.
157.
158.
159.
160.
161.
162.
163.
164.
165.
166.
167.
168.
169.
170.
171.
172.
173.
174.

go操作kafka

Kafka介绍

kafka使用scala开发，支持多语言客户端(c++、java、python、go等)
Kafka最先由LinkedIn公司开发，之后成为Apache的顶级项目。
Kafka是一个分布式的、分区化、可复制提交的日志服务
LinkedIn使用Kafka实现了公司不同应用程序之间的松耦和，那么作为一个可扩展、高可靠的消息系统
支持高Throughput的应用
scale out：无需停机即可扩展机器
持久化：通过将数据持久化到硬盘以及replication防止数据丢失
支持online和offline的场景

Kafka的特点

Kafka是分布式的，其所有的构件borker(服务端集群)、producer(消息生产)、consumer(消息消费者)都可以是分布式的。

在消息的生产时可以使用一个标识topic来区分，且可以进行分区；每一个分区都是一个顺序的、不可变的消息队列，并且可以持续的添加。

同时为发布和订阅提供高吞吐量。据了解，Kafka每秒可以生产约25万消息(50 MB)，每秒处理55万消息(110 MB)。

消息被处理的状态是在consumer端维护，而不是由server端维护。当失败时能自动平衡

常用的场景

**监控：**主机通过Kafka发送与系统和应用程序健康相关的指标，然后这些信息会被收集和处理从而创建监控仪表盘并发送警告。

消息队列： 应用程度使用Kafka作为传统的消息系统实现标准的队列和消息的发布—订阅，例如搜索和内容提要(Content Feed)。比起大多数的消息系统来说，Kafka有更好的吞吐量，内置的分区，冗余及容错性，这让Kafka成为了一个很好的大规模消息处理应用的解决方案。消息系统一般吞吐量相对较低，但是需要更小的端到端延时，并尝尝依赖于Kafka提供的强大的持久性保障。在这个领域，Kafka足以媲美传统消息系统，如ActiveMR或RabbitMQ

站点的用户活动追踪: 为了更好地理解用户行为，改善用户体验，将用户查看了哪个页面、点击了哪些内容等信息发送到每个数据中心的Kafka集群上，并通过Hadoop进行分析、生成日常报告。

流处理： 保存收集流数据，以提供之后对接的Storm或其他流式计算框架进行处理。很多用户会将那些从原始topic来的数据进行阶段性处理，汇总，扩充或者以其他的方式转换到新的topic下再继续后面的处理。例如一个文章推荐的处理流程，可能是先从RSS数据源中抓取文章的内容，然后将其丢入一个叫做“文章”的topic中；后续操作可能是需要对这个内容进行清理，比如回复正常数据或者删除重复数据，最后再将内容匹配的结果返还给用户。这就在一个独立的topic之外，产生了一系列的实时数据处理的流程。

日志聚合:使用Kafka代替日志聚合(log aggregation)。日志聚合一般来说是从服务器上收集日志文件，然后放到一个集中的位置(文件服务器或HDFS)进行处理。然而Kafka忽略掉文件的细节，将其更清晰地抽象成一个个日志或事件的消息流。这就让Kafka处理过程延迟更低，更容易支持多数据源和分布式数据处理。比起以日志为中心的系统比如Scribe或者Flume来说，Kafka提供同样高效的性能和因为复制导致的更高的耐用性保证，以及更低的端到端延迟

持久性日志： Kafka可以为一种外部的持久性日志的分布式系统提供服务。这种日志可以在节点间备份数据，并为故障节点数据回复提供一种重新同步的机制。Kafka中日志压缩功能为这种用法提供了条件。在这种用法中，Kafka类似于Apache BookKeeper项目。

Kafka中包含以下基础概念

Topic(话题)：Kafka中用于区分不同类别信息的类别名称。由producer指定
Producer(生产者)：将消息发布到Kafka特定的Topic的对象(过程)
Consumers(消费者)：订阅并处理特定的Topic中的消息的对象(过程)
Broker(Kafka服务集群)：已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理(Broker). 消费者可以订阅一个或多个话题，并从Broker拉数据，从而消费这些已发布的消息。
Partition(分区)：Topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id(offset)
Message：消息，是通信的基本单位，每个producer可以向一个topic(主题)发布一些消息。

package main

import (
    "fmt"

    "github.com/Shopify/sarama"
)

// 基于sarama第三方库开发的kafka client
func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
    config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
    config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

    // 构造一个消息
    msg := &sarama.ProducerMessage{}
    msg.Topic = "web_log"
    msg.Value = sarama.StringEncoder("this is a test log")
    // 连接kafka
    client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
    if err != nil {
        fmt.Println("producer closed, err:", err)
        return
    }
    defer client.Close()
    // 发送消息
    pid, offset, err := client.SendMessage(msg)
    if err != nil {
        fmt.Println("send msg failed, err:", err)
        return
    }
    fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

// kafka consumer
func main() {
  consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
  if err != nil {
      fmt.Printf("fail to start consumer, err:%v\n", err)
      return
  }
  partitionList, err := consumer.Partitions("web_log") // 根据topic取到所有的分区
  if err != nil {
      fmt.Printf("fail to get list of partition:err%v\n", err)
      return
  }
  fmt.Println(partitionList)
  for partition := range partitionList { // 遍历所有的分区
    // 针对每个分区创建一个对应的分区消费者
    pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
    if err != nil {
        fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
        return
    }
    defer pc.AsyncClose()
    // 异步从每个分区消费信息
    go func(sarama.PartitionConsumer) {
        for msg := range pc.Messages() {
            fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
        }
    }(pc)
  }
}

go操作RabbitMQ

环境准备

安装Go
访问 Go官网下载并安装最新版本的Go。

安装RabbitMQ
访问 RabbitMQ官网安装RabbitMQ。安装完成后启动服务，并确保可以在浏览器中通过 http://localhost:15672 访问管理界面。

安装RabbitMQ客户端库
打开终端或命令提示符，运行以下命令来安装Go语言的RabbitMQ客户端库：

基础概念

AMQP: Advanced Message Queuing Protocol，一种为应用程序之间提供通用协议的标准。
Exchange: 交换器，用来接收生产者发送的消息然后根据key决定如何路由消息。
Queue: 消息队列，用来保存消息直到消费者获取它们。
Binding: 绑定，用于将队列与交换器连接起来。
Consumer: 消费者，接收消息的应用程序。
Producer: 生产者，发送消息的应用程序。

基本示例

发送消息

package main

import (
    "fmt"
    amqp "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        panic(fmt.Sprintf("%s: %s", msg, err))
    }
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "hello", // name
        false,   // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "Failed to declare a queue")

    body := "Hello World!"
    err = ch.Publish(
        "",     // exchange
        q.Name, // routing key
        false,  // mandatory
        false,  // immediate
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(body),
        })
    failOnError(err, "Failed to publish a message")

    fmt.Printf(" [x] Sent %s\n", body)
}

接收消息

package main

import (
    "fmt"
    amqp "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        panic(fmt.Sprintf("%s: %s", msg, err))
    }
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "hello", // name
        false,   // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "Failed to declare a queue")

    msgs, err := ch.Consume(
        q.Name, // queue
        "",     // consumer
        true,   // auto-ack
        false,  // exclusive
        false,  // no-local
        false,  // no-wait
        nil,    // args
    )
    failOnError(err, "Failed to register a consumer")

    fmt.Println(" [*] Waiting for messages. To exit press CTRL+C")
    forever := make(chan bool)

    go func() {
        for d := range msgs {
            fmt.Printf(" [x] Received %s\n", d.Body)
        }
    }()

    <-forever
}

持久化消息

持久化队列

设置队列为持久化队列，即使RabbitMQ重启后队列仍然存在。
示例代码：

q, err := ch.QueueDeclare(
    "hello", // name
    true,    // durable
    false,   // delete when unused
    false,   // exclusive
    false,   // no-wait
    nil,     // arguments
)

持久化消息

发布消息时设置Delivery.Persistent为true。
示例代码：

err = ch.Publish(
    "",     // exchange
    q.Name, // routing key
    false,  // mandatory
    false,  // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
        DeliveryMode: amqp.Persistent, // 持久化消息
    })

发布确认

开启发布确认

在通道上启用发布确认功能。
示例代码：

if err := ch.Confirm(false); err != nil {
    log.Fatalf("Failed to enable publisher confirms: %v", err)
}

等待确认

发布消息后等待确认结果。
示例代码：

err = ch.Publish(
    "",     // exchange
    q.Name, // routing key
    false,  // mandatory
    false,  // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
    })
if err != nil {
    log.Printf("Failed to publish a message: %s", err)
    return
}

// 等待确认
if ok, n, err := ch.WaitForConfirmsOrDie(); err != nil {
    log.Fatalf("Error waiting for confirms: %v", err)
} else {
    log.Printf("Published %d messages", n)
}

事务

开启事务
开启事务后，可以确保消息处理的原子性。
示例代码：

if err := ch.Tx(); err != nil {
    log.Fatalf("Failed to start transaction: %v", err)
}

// 发布消息
err = ch.Publish(
    "",     // exchange
    q.Name, // routing key
    false,  // mandatory
    false,  // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
    })
if err != nil {
    log.Printf("Failed to publish a message: %s", err)
    return
}

// 提交事务
if err := ch.Commit(); err != nil {
    log.Fatalf("Failed to commit transaction: %v", err)
}

RPC模式

实现RPC模式
通过队列传递请求和响应，实现远程过程调用。
示例代码：

// 发送请求
corrId := generateCorrelationId()
replyQueueName, _ := ch.QueueDeclarePassive(replyQueueName)
err = ch.Publish(
    "",       // exchange
    requestQ, // routing key
    false,    // mandatory
    false,    // immediate
    amqp.Publishing{
        ContentType:      "text/plain",
        CorrelationId:    corrId,
        ReplyTo:          replyQueueName,
        Body:             []byte(request),
    })
if err != nil {
    log.Fatalf("Failed to publish a RPC request: %v", err)
}

// 接收响应
var response string
select {
case delivery := <-msgs:
    if delivery.CorrelationId == corrId {
        response = string(delivery.Body)
        delivery.Ack(false)
    }
case <-time.After(time.Second * 10):
    log.Fatalf("RPC request timed out")
}

Fanout Exchange

Fanout Exchange可以将消息广播到所有绑定的队列。

创建Fanout Exchange

示例代码：

err = ch.ExchangeDeclare(
    "fanout_exchange", // name
    "fanout",          // type
    true,              // durable
    false,             // auto-deleted
    false,             // internal
    false,             // no-wait
    nil,               // arguments
)
if err != nil {
    log.Fatalf("Failed to declare an exchange: %v", err)
}

绑定队列到Fanout Exchange

示例代码：

q1, err := ch.QueueDeclare(
    "queue1", // name
    true,     // durable
    false,    // delete when unused
    false,    // exclusive
    false,    // no-wait
    nil,      // arguments
)
if err != nil {
    log.Fatalf("Failed to declare a queue: %v", err)
}

err = ch.QueueBind(
    q1.Name, // queue name
    "",      // routing key
    "fanout_exchange", // exchange
    false,
    nil,
)
if err != nil {
    log.Fatalf("Failed to bind a queue: %v", err)
}

q2, err := ch.QueueDeclare(
    "queue2", // name
    true,     // durable
    false,    // delete when unused
    false,    // exclusive
    false,    // no-wait
    nil,      // arguments
)
if err != nil {
    log.Fatalf("Failed to declare a queue: %v", err)
}

err = ch.QueueBind(
    q2.Name, // queue name
    "",      // routing key
    "fanout_exchange", // exchange
    false,
    nil,
)
if err != nil {
    log.Fatalf("Failed to bind a queue: %v", err)
}

发布消息到Fanout Exchange

示例代码：

body := "Hello World!"
err = ch.Publish(
    "fanout_exchange", // exchange
    "",                // routing key
    false,             // mandatory
    false,             // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
    })
if err != nil {
    log.Fatalf("Failed to publish a message: %v", err)
}

go操作ElasticSearch

连接Elasticsearch

创建客户端

使用elasticsearch.NewDefaultClient创建客户端。
示例代码：

import (
    "context"
    "github.com/elastic/go-elasticsearch/v8"
    "github.com/elastic/go-elasticsearch/v8/esapi"
)

func main() {
    client, err := elasticsearch.NewDefaultClient()
    if err != nil {
        log.Fatalf("Error creating the client: %s", err)
    }

    ctx := context.Background()

    // 测试连接
    resp, err := client.Info(ctx)
    if err != nil {
        log.Fatalf("Error getting response: %s", err)
    }
    defer resp.Body.Close()

    if resp.IsError() {
        log.Fatalf("Error response: %s", resp.Status())
    }
    fmt.Println("Connected to Elasticsearch!")
}

创建索引

使用esapi.PutMappingRequest创建索引。
示例代码：

indexName := "my_index"

req := esapi.IndicesCreateRequest{
    Index: indexName,
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error creating index: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}
fmt.Println("Index created successfully!")

映射定义

使用esapi.PutMappingRequest定义映射。
示例代码：

mapping := `
{
    "properties": {
        "title": {
            "type": "text"
        },
        "author": {
            "type": "keyword"
        },
        "content": {
            "type": "text"
        }
    }
}`

req := esapi.IndicesPutMappingRequest{
    Index: []string{indexName},
    Body:  strings.NewReader(mapping),
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error putting mapping: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}
fmt.Println("Mapping defined successfully!")

插入文档

使用esapi.IndexRequest插入文档。
示例代码：

doc := `
{
    "title": "First Document",
    "author": "John Doe",
    "content": "This is the first document."
}`

req := esapi.IndexRequest{
    Index: indexName,
    Body:  strings.NewReader(doc),
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error indexing document: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}
fmt.Println("Document indexed successfully!")

查询文档

使用esapi.SearchRequest进行简单查询。
示例代码：

query := `
{
    "query": {
        "match": {
            "title": "First Document"
        }
    }
}`

req := esapi.SearchRequest{
    Index: []string{indexName},
    Body:  strings.NewReader(query),
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error searching documents: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}

var result map[string]interface{}
json.NewDecoder(resp.Body).Decode(&result)

hits := result["hits"].(map[string]interface{})["hits"].([]interface{})
for _, hit := range hits {
    fmt.Println(hit.(map[string]interface{})["_source"])
}

批量操作

批量插入
使用esapi.BulkRequest进行批量插入。
示例代码：

bulkBody := `
[
    { "index": { "_index": "my_index" } },
    { "title": "Second Document", "author": "Jane Doe", "content": "This is the second document." },
    { "index": { "_index": "my_index" } },
    { "title": "Third Document", "author": "Alice Smith", "content": "This is the third document." }
]`

req := esapi.BulkRequest{
    Body: strings.NewReader(bulkBody),
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error performing bulk operation: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}
fmt.Println("Bulk operation completed successfully!")

更新文档

使用esapi.UpdateRequest更新文档。
示例代码：

updateDoc := `
{
    "doc": {
        "content": "Updated content"
    }
}`

req := esapi.UpdateRequest{
    Index: indexName,
    Id:    "1",
    Body:  strings.NewReader(updateDoc),
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error updating document: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}
fmt.Println("Document updated successfully!")

删除文档

使用esapi.DeleteRequest删除文档。
示例代码：

req := esapi.DeleteRequest{
    Index: indexName,
    Id:    "1",
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error deleting document: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}
fmt.Println("Document deleted successfully!")

聚合查询

使用esapi.SearchRequest进行聚合查询。
示例代码：

aggQuery := `
{
    "size": 0,
    "aggs": {
        "authors": {
            "terms": {
                "field": "author.keyword"
            }
        }
    }
}`

req := esapi.SearchRequest{
    Index: []string{indexName},
    Body:  strings.NewReader(aggQuery),
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error performing aggregation query: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}

var result map[string]interface{}
json.NewDecoder(resp.Body).Decode(&result)

buckets := result["aggregations"].(map[string]interface{})["authors"].(map[string]interface{})["buckets"].([]interface{})
for _, bucket := range buckets {
    fmt.Println(bucket.(map[string]interface{})["key"], bucket.(map[string]interface{})["doc_count"])
}

复合查询

使用esapi.SearchRequest进行复合查询。
示例代码：

complexQuery := `
{
    "query": {
        "bool": {
            "must": [
                { "match": { "title": "Document" } }
            ],
            "filter": [
                { "term": { "author": "John Doe" } }
            ]
        }
    }
}`

req := esapi.SearchRequest{
    Index: []string{indexName},
    Body:  strings.NewReader(complexQuery),
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error performing complex query: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}

var result map[string]interface{}
json.NewDecoder(resp.Body).Decode(&result)

hits := result["hits"].(map[string]interface{})["hits"].([]interface{})
for _, hit := range hits {
    fmt.Println(hit.(map[string]interface{})["_source"])
}

排序查询

使用esapi.SearchRequest进行排序查询。
示例代码：

sortQuery := `
{
    "query": {
        "match_all": {}
    },
    "sort": [
        { "title": { "order": "desc" } }
    ]
}`

req := esapi.SearchRequest{
    Index: []string{indexName},
    Body:  strings.NewReader(sortQuery),
}

resp, err := req.Do(ctx, client)
if err != nil {
    log.Fatalf("Error performing sort query: %s", err)
}
defer resp.Body.Close()

if resp.IsError() {
    log.Fatalf("Error response: %s", resp.Status())
}

var result map[string]interface{}
json.NewDecoder(resp.Body).Decode(&result)

hits := result["hits"].(map[string]interface{})["hits"].([]interface{})
for _, hit := range hits {
    fmt.Println(hit.(map[string]interface{})["_source"])
}

NSQ

安装NSQ客户端

首先确保已经安装了Go环境，并且可以通过go version命令验证Go是否正确安装。

接着，安装Go语言的NSQ客户端库go-nsq：

NSQ生产者

生产者负责向NSQ集群发送消息。下面是一个简单的Go程序示例，展示如何使用go-nsq库来创建一个NSQ生产者：

package main

import (
"fmt"
"os"
"time"

"github.com/nsqio/go-nsq"
)

func main() {
config := nsq.NewConfig()
producer, err := nsq.NewProducer(nsqlookupdHTTPAddrs, config)
if err != nil {
fmt.Println(err)
os.Exit(1)
}

defer producer.Stop()

msgBody := []byte("Hello, NSQ!")
topic := "test_topic"

err = producer.Publish(topic, msgBody)
if err != nil {
fmt.Println(err)
os.Exit(1)
}

fmt.Printf("Published message: %s\n", msgBody)
time.Sleep(time.Second * 2) // 等待一段时间以确保消息被发送
}

在这个例子中，我们创建了一个NSQ生产者，并向名为test_topic的主题发送了一条消息。

NSQ消费者

消费者负责接收来自NSQ集群的消息。下面是一个简单的Go程序示例，展示如何使用go-nsq库来创建一个NSQ消费者：

package main

import (
"fmt"
"os"
"os/signal"
"syscall"
"time"

"github.com/nsqio/go-nsq"
)

type MyHandler struct{}

func (h *MyHandler) HandleMessage(message *nsq.Message) error {
fmt.Printf("Received message: %s\n", message.Body)
return nil
}

func main() {
config := nsq.NewConfig()
consumer, err := nsq.NewConsumer("test_topic", "test_channel", config)
if err != nil {
fmt.Println(err)
os.Exit(1)
}

consumer.AddHandler(&MyHandler{})

err = consumer.ConnectToNSQLookupd("localhost:4161")
if err != nil {
fmt.Println(err)
os.Exit(1)
}

// 设置信号处理器以优雅地关闭消费者
sigTerm := make(chan os.Signal, 1)
signal.Notify(sigTerm, syscall.SIGINT, syscall.SIGTERM)
<-sigTerm

consumer.Stop()
fmt.Println("Shutting down...")
time.Sleep(time.Second * 2)
}

在这个例子中，我们创建了一个NSQ消费者，并订阅了名为test_topic的主题。当接收到消息时，消费者会打印出消息的内容， https://t.me/gtokentool 。

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

Go语言数据库操作深入讲解

go操作MySQL

Insert操作

Select操作

Update操作

Delete操作

MySQL事务

go操作Redis

Redis连接

String类型Set、Get操作

String批量操作

设置过期时间

List队列操作

Hash表

Redis连接池

go操作ETCD

etcd应用场景

服务发现

配置中心

分布式锁

为什么用 etcd 而不用ZooKeeper？

为什么不选择ZooKeeper？

为什么选择etcd？

etcd集群

操作ETCD

put和get操作

watch操作

lease租约

keepAlive

基于etcd实现分布式锁

zookeeper

简单的分布式server

server：

client：

go操作kafka

Kafka介绍

Kafka的特点

常用的场景

Kafka中包含以下基础概念

go操作RabbitMQ

环境准备

基础概念

基本示例

持久化消息

发布确认

事务

RPC模式

Fanout Exchange

go操作ElasticSearch

连接Elasticsearch

创建索引

映射定义

插入文档

查询文档

批量操作

更新文档

删除文档

聚合查询

复合查询

排序查询

NSQ

安装NSQ客户端

NSQ生产者

NSQ消费者

相关文章