目录go操作MySQLgo操作NoSQLgo操作PgSQLgo操作Redisgo操作ETCDzookeepergo操作kafkago操作RabbitMQgo操作ElasticSearchNSQgo操作MySQL使用第三方开源的mysql库:github.com/go
命令行输入 :
// 连接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)
}
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)
}
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)
}
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)
}
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()
}
使用第三方开源的redis库: github.com/garyburd/redigo/redis: go get github.com/garyburd/redigo/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()
}
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)
}
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
}
}
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)
}
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)
}
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() //关闭连接池
}
etcd是使用Go语言开发的一个开源的、高可用的分布式key-value存储系统,可以用于配置共享和服务的注册和发现。
类似项目有zookeeper和consul。
etcd具有以下特点:
服务发现要解决的也是分布式系统中最常见的问题之一,即在同一个分布式集群中的进程或服务,要如何才能找到对方并建立连接。本质上来说,服务发现就是想要了解集群中是否有进程在监听 udp 或 tcp 端口,并且通过名字就可以查找和连接。
将一些配置信息放到 etcd 上进行集中管理。
这类场景的使用方式通常是这样:应用在启动的时候主动从 etcd 获取一次配置信息,同时,在 etcd 节点上注册一个 Watcher 并等待,以后每次配置有更新的时候,etcd 都会实时通知订阅者,以此达到获取最新配置信息的目的。
因为 etcd 使用 Raft 算法保持了数据的强一致性,某次操作存储到集群中的值必然是全局一致的,所以很容易实现分布式锁。锁服务有两种使用方式,一是保持独占,二是控制时序。
etcd 实现的这些功能,ZooKeeper都能实现。那么为什么要用 etcd 而非直接使用ZooKeeper呢?
部署维护复杂,其使用的Paxos强一致性算法复杂难懂。官方只提供了Java和C两种语言的接口。 使用Java编写引入大量的依赖。运维人员维护起来比较麻烦。 最近几年发展缓慢,不如etcd和consul等后起之秀。
最后,etcd 作为一个年轻的项目,正在高速迭代和开发中,这既是一个优点,也是一个缺点。优点是它的未来具有无限的可能性,缺点是无法得到大项目长时间使用的检验。然而,目前 CoreOS、Kubernetes和CloudFoundry等知名项目均在生产环境中使用了etcd,所以总的来说,etcd值得你去尝试。
etcd 作为一个高可用键值存储系统,天生就是为集群化而设计的。由于 Raft 算法在做决策时需要多数节点的投票,所以 etcd 一般部署集群推荐奇数个节点,推荐的数量为 3、5 或者 7 个节点构成一个集群。
// 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用来获取未来更改的通知。
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)
}
}
}
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)
}
}
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)
}
}
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")
目前分布式系统已经很流行了,一些开源框架也被广泛应用,如dubbo、Motan等。对于一个分布式服务,最基本的一项功能就是服务的注册和发现,而利用zk的EPHEMERAL节点则可以很方便的实现该功能。EPHEMERAL节点正如其名,是临时性的,其生命周期是和客户端会话绑定的,当会话连接断开时,节点也会被删除。下边我们就来实现一个简单的分布式server:
服务启动时,创建zk连接,并在go_servers节点下创建一个新节点,节点名为”ip:port”,完成服务注册 服务结束时,由于连接断开,创建的节点会被删除,这样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
}
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项目。
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 访问 Go官网 下载并安装最新版本的Go。
安装RabbitMQ 访问 RabbitMQ官网 安装RabbitMQ。安装完成后启动服务,并确保可以在浏览器中通过 http://localhost:15672 访问管理界面。
安装RabbitMQ客户端库 打开终端或命令提示符,运行以下命令来安装Go语言的RabbitMQ客户端库:
go get -u github.com/streadway/amqp
发送消息
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模式 通过队列传递请求和响应,实现远程过程调用。 示例代码:
// 发送请求
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
示例代码:
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)
}
创建客户端
使用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"])
}
首先确保已经安装了Go环境,并且可以通过go version命令验证Go是否正确安装。
接着,安装Go语言的NSQ客户端库go-nsq:
go get github.com/nsqio/go-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集群的消息。下面是一个简单的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的主题。当接收到消息时,消费者会打印出消息的内容。
如果觉得我的文章对您有用,请随意打赏。你的支持将鼓励我继续创作!