這篇文章將為大家詳細(xì)講解有關(guān)golang實(shí)現(xiàn)連接池的方法,小編覺得挺實(shí)用的,因此分享給大家做個(gè)參考,希望大家閱讀完這篇文章后可以有所收獲。
成都創(chuàng)新互聯(lián)公司是專業(yè)的南通網(wǎng)站建設(shè)公司,南通接單;提供網(wǎng)站建設(shè)、網(wǎng)站設(shè)計(jì),網(wǎng)頁(yè)設(shè)計(jì),網(wǎng)站設(shè)計(jì),建網(wǎng)站,PHP網(wǎng)站建設(shè)等專業(yè)做網(wǎng)站服務(wù);采用PHP框架,可快速的進(jìn)行南通網(wǎng)站開發(fā)網(wǎng)頁(yè)制作和功能擴(kuò)展;專業(yè)做搜索引擎喜愛的網(wǎng)站,專業(yè)的做網(wǎng)站團(tuán)隊(duì),希望更多企業(yè)前來(lái)合作!
Reids
安裝導(dǎo)入
go get github.com/garyburd/redigo/redis import "github.com/garyburd/redigo/redis"
使用
連接
import "github.com/garyburd/redigo/redis" func main() { c, err := redis.Dial("tcp", "localhost:6379") if err != nil { fmt.Println("conn redis failed, err:", err) return } defer c.Close() }
set & get
_, err = c.Do("Set", "name", "nick") if err != nil { fmt.Println(err) return } r, err := redis.String(c.Do("Get", "name")) if err != nil { fmt.Println(err) return } fmt.Println(r)
mset & mget
批量設(shè)置
_, err = c.Do("MSet", "name", "nick", "age", "18") if err != nil { fmt.Println("MSet error: ", err) return } r2, err := redis.Strings(c.Do("MGet", "name", "age")) if err != nil { fmt.Println("MGet error: ", err) return } fmt.Println(r2)
hset & hget
hash操作
_, err = c.Do("HSet", "names", "nick", "suoning") if err != nil { fmt.Println("hset error: ", err) return } r, err = redis.String(c.Do("HGet", "names", "nick")) if err != nil { fmt.Println("hget error: ", err) return } fmt.Println(r)
expire
設(shè)置過期時(shí)間
_, err = c.Do("expire", "names", 5) if err != nil { fmt.Println("expire error: ", err) return }
lpush & lpop & llen
隊(duì)列
// 隊(duì)列 _, err = c.Do("lpush", "Queue", "nick", "dawn", 9) if err != nil { fmt.Println("lpush error: ", err) return } for { r, err = redis.String(c.Do("lpop", "Queue")) if err != nil { fmt.Println("lpop error: ", err) break } fmt.Println(r) } r3, err := redis.Int(c.Do("llen", "Queue")) if err != nil { fmt.Println("llen error: ", err) return }
連接池
各參數(shù)的解釋如下:
MaxIdle:最大的空閑連接數(shù),表示即使沒有redis連接時(shí)依然可以保持N個(gè)空閑的連接,而不被清除,隨時(shí)處于待命狀態(tài)。
MaxActive:最大的激活連接數(shù),表示同時(shí)最多有N個(gè)連接
IdleTimeout:最大的空閑連接等待時(shí)間,超過此時(shí)間后,空閑連接將被關(guān)閉
pool := &redis.Pool{ MaxIdle: 16, MaxActive: 1024, IdleTimeout: 300, Dial: func() (redis.Conn, error) { return redis.Dial("tcp", "localhost:6379") }, }
連接池例子:
package main import ( "fmt" "github.com/garyburd/redigo/redis" ) var pool *redis.Pool func init() { pool = &redis.Pool{ MaxIdle: 16, MaxActive: 1024, IdleTimeout: 300, Dial: func() (redis.Conn, error) { return redis.Dial("tcp", "localhost:6379") }, } } func main() { c := pool.Get() defer c.Close() _, err := c.Do("Set", "name", "nick") if err != nil { fmt.Println(err) return } r, err := redis.String(c.Do("Get", "name")) if err != nil { fmt.Println(err) return } fmt.Println(r) }
管道操作
請(qǐng)求/響應(yīng)服務(wù)可以實(shí)現(xiàn)持續(xù)處理新請(qǐng)求,客戶端可以發(fā)送多個(gè)命令到服務(wù)器而無(wú)需等待響應(yīng),最后在一次讀取多個(gè)響應(yīng)。
使用Send(),F(xiàn)lush(),Receive()方法支持管道化操作
Send向連接的輸出緩沖中寫入命令。
Flush將連接的輸出緩沖清空并寫入服務(wù)器端。
Recevie按照FIFO順序依次讀取服務(wù)器的響應(yīng)。
func main() { c, err := redis.Dial("tcp", "localhost:6379") if err != nil { fmt.Println("conn redis failed, err:", err) return } defer c.Close() c.Send("SET", "name1", "sss1") c.Send("SET", "name2", "sss2") c.Flush() v, err := c.Receive() fmt.Printf("v:%v,err:%v\n", v, err) v, err = c.Receive() fmt.Printf("v:%v,err:%v\n", v, err) v, err = c.Receive() // 夯住,一直等待 fmt.Printf("v:%v,err:%v\n", v, err) }
MySQL
安裝導(dǎo)入
go get "github.com/go-sql-driver/mysql" go get "github.com/jmoiron/sqlx" import ( _ "github.com/go-sql-driver/mysql" "github.com/jmoiron/sqlx" )
連接
import ( _ "github.com/go-sql-driver/mysql" "github.com/jmoiron/sqlx" ) var Db *sqlx.DB func init() { database, err := sqlx.Open("mysql", "root:@tcp(127.0.0.1:3306)/test") if err != nil { fmt.Println("open mysql failed,", err) return } Db = database }
建表
CREATE TABLE `person` ( `user_id` int(128) DEFAULT NULL, `username` varchar(255) DEFAULT NULL, `sex` varchar(16) DEFAULT NULL, `email` varchar(128) DEFAULT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8
(insert)
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"` } var Db *sqlx.DB func init() { database, err := sqlx.Open("mysql", "root:@tcp(127.0.0.1:3306)/test") if err != nil { fmt.Println("open mysql failed,", err) return } Db = database } func main() { r, err := Db.Exec("insert into person(username, sex, email)values(?, ?, ?)", "suoning", "man", "suoning@net263.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) }
(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"` } var Db *sqlx.DB func init() { database, err := sqlx.Open("mysql", "root:@tcp(127.0.0.1:3306)/test") if err != nil { fmt.Println("open mysql failed,", err) return } Db = database } func main() { _, err := Db.Exec("update person set user_id=? where username=?", 20170808, "suoning") if err != nil { fmt.Println("exec failed, ", err) return } }
(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:@tcp(127.0.0.1:3306)/test") if err != nil { fmt.Println("open mysql failed,", err) return } Db = database } 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) people := []Person{} Db.Select(&people, "SELECT * FROM person ORDER BY user_id ASC") fmt.Println(people) jason, john := people[0], people[1] fmt.Printf("%#v\n%#v", jason, john) }
(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"` } var Db *sqlx.DB func init() { database, err := sqlx.Open("mysql", "root:@tcp(127.0.0.1:3306)/test") if err != nil { fmt.Println("open mysql failed,", err) return } Db = database } func main() { _, err := Db.Exec("delete from person where username=? limit 1", "suoning") if err != nil { fmt.Println("exec failed, ", err) return } fmt.Println("delete succ") }
事務(wù)
package main import ( "github.com/astaxie/beego/logs" _ "github.com/go-sql-driver/mysql" "github.com/jmoiron/sqlx" ) var Db *sqlx.DB func init() { database, err := sqlx.Open("mysql", "root:@tcp(127.0.0.1:3306)/test") if err != nil { logs.Error("open mysql failed,", err) return } Db = database } func main() { conn, err := Db.Begin() if err != nil { logs.Warn("DB.Begin failed, err:%v", err) return } defer func() { if err != nil { conn.Rollback() return } conn.Commit() }() // do something }
RabbitMQ
安裝
go get "github.com/streadway/amqp"
普通模式
生產(chǎn)者:
package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" "time")/*默認(rèn)點(diǎn)對(duì)點(diǎn)模式*/func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) 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() // 打開一個(gè)并發(fā)服務(wù)器通道來(lái)處理消息 ch, err := conn.Channel() failOnError(err, "Failed to open a channel") defer ch.Close() // 申明一個(gè)隊(duì)列 q, err := ch.QueueDeclare( "task_queue", // name true, // durable 持久性的,如果事前已經(jīng)聲明了該隊(duì)列,不能重復(fù)聲明 false, // delete when unused false, // exclusive 如果是真,連接一斷開,隊(duì)列刪除 false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare a queue") body := bodyFrom(os.Args) // 發(fā)布 err = ch.Publish( "", // exchange 默認(rèn)模式,exchange為空 q.Name, // routing key 默認(rèn)模式路由到同名隊(duì)列,即是task_queue false, // mandatory false, amqp.Publishing{ // 持久性的發(fā)布,因?yàn)殛?duì)列被聲明為持久的,發(fā)布消息必須加上這個(gè)(可能不用),但消息還是可能會(huì)丟,如消息到緩存但MQ掛了來(lái)不及持久化。 DeliveryMode: amqp.Persistent, ContentType: "text/plain", Body: []byte(body), }) failOnError(err, "Failed to publish a message") log.Printf(" [x] Sent %s", body) } func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" { s = fmt.Sprintf("%s-%v","hello", time.Now()) } else { s = strings.Join(args[1:], " ") } return s }
消費(fèi)者:
package main import ( "bytes" "fmt" "github.com/streadway/amqp" "log" "time" ) /* 默認(rèn)點(diǎn)對(duì)點(diǎn)模式 工作方,多個(gè),拿發(fā)布方的消息 */ func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) 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() // 指定隊(duì)列! q, err := ch.QueueDeclare( "task_queue", // name true, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare a queue") // Fair dispatch 預(yù)取,每個(gè)工作方每次拿一個(gè)消息,確認(rèn)后才拿下一次,緩解壓力 err = ch.Qos( 1, // prefetch count 0, // prefetch size false, // global ) failOnError(err, "Failed to set QoS") // 消費(fèi)根據(jù)隊(duì)列名 msgs, err := ch.Consume( q.Name, // queue "", // consumer false, // auto-ack 設(shè)置為真自動(dòng)確認(rèn)消息 false, // exclusive false, // no-local false, // no-wait nil, // args ) failOnError(err, "Failed to register a consumer") forever := make(chan bool) go func() { for d := range msgs { log.Printf("Received a message: %s", d.Body) dot_count := bytes.Count(d.Body, []byte(".")) t := time.Duration(dot_count) time.Sleep(t * time.Second) log.Printf("Done") // 確認(rèn)消息被收到!!如果為真的,那么同在一個(gè)channel,在該消息之前未確認(rèn)的消息都會(huì)確認(rèn),適合批量處理 // 真時(shí)場(chǎng)景:每十條消息確認(rèn)一次,類似 d.Ack(false) } }() log.Printf(" [*] Waiting for messages. To exit press CTRL+C") <-forever }
訂閱模式
訂閱 生產(chǎn)者:
package main import ( "fmt" "github.com/streadway/amqp" "log" "os" "strings" "time" ) /* 廣播模式 發(fā)布方 */ func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) 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() // 默認(rèn)模式有默認(rèn)交換機(jī),廣播自己定義一個(gè)交換機(jī),交換機(jī)可與隊(duì)列進(jìn)行綁定 err = ch.ExchangeDeclare( "logs", // name "fanout", // type 廣播模式 true, // durable false, // auto-deleted false, // internal false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare an exchange") body := bodyFrom(os.Args) // 發(fā)布 err = ch.Publish( "logs", // exchange 消息發(fā)送到交換機(jī),這個(gè)時(shí)候沒隊(duì)列綁定交換機(jī),消息會(huì)丟棄 "", // routing key 廣播模式不需要這個(gè),它會(huì)把所有消息路由到綁定的所有隊(duì)列 false, // mandatory false, // immediate amqp.Publishing{ ContentType: "text/plain", Body: []byte(body), }) failOnError(err, "Failed to publish a message") log.Printf(" [x] Sent %s", body) } func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" { s = fmt.Sprintf("%s-%v","hello", time.Now()) } else { s = strings.Join(args[1:], " ") } return s }
訂閱 消費(fèi)者:
package main import ( "fmt" "github.com/streadway/amqp" "log" ) /* 廣播模式 訂閱方 */ func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) 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() // 同樣要申明交換機(jī) err = ch.ExchangeDeclare( "logs", // name "fanout", // type true, // durable false, // auto-deleted false, // internal false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare an exchange") // 新建隊(duì)列,這個(gè)隊(duì)列沒名字,隨機(jī)生成一個(gè)名字 q, err := ch.QueueDeclare( "", // name false, // durable false, // delete when usused true, // exclusive 表示連接一斷開,這個(gè)隊(duì)列自動(dòng)刪除 false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare a queue") // 隊(duì)列和交換機(jī)綁定,即是隊(duì)列訂閱了發(fā)到這個(gè)交換機(jī)的消息 err = ch.QueueBind( q.Name, // queue name 隊(duì)列的名字 "", // routing key 廣播模式不需要這個(gè) "logs", // exchange 交換機(jī)名字 false, nil) failOnError(err, "Failed to bind a queue") // 開始消費(fèi)消息,可開多個(gè)訂閱方,因?yàn)殛?duì)列是臨時(shí)生成的,所有每個(gè)訂閱方都能收到同樣的消息 msgs, err := ch.Consume( q.Name, // queue 隊(duì)列名字 "", // consumer true, // auto-ack 自動(dòng)確認(rèn) false, // exclusive false, // no-local false, // no-wait nil, // args ) failOnError(err, "Failed to register a consumer") forever := make(chan bool) go func() { for d := range msgs { log.Printf(" [x] %s", d.Body) } }() log.Printf(" [*] Waiting for logs. To exit press CTRL+C") <-forever }
RPC模式
RPC 應(yīng)答方:
package main import ( "fmt" "log" "strconv" "github.com/streadway/amqp" ) /* RPC模式 應(yīng)答方 */ func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err)) } } func fib(n int) int { if n == 0 { return 0 } else if n == 1 { return 1 } else { return fib(n-1) + fib(n-2) } } 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( "rpc_queue", // name false, // durable false, // delete when usused false, // exclusive false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare a queue") // 公平分發(fā) 沒有這個(gè)則round-robbin err = ch.Qos( 1, // prefetch count 0, // prefetch size false, // global ) failOnError(err, "Failed to set QoS") // 消費(fèi),等待請(qǐng)求 msgs, err := ch.Consume( q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args ) failOnError(err, "Failed to register a consumer") forever := make(chan bool) go func() { //請(qǐng)求來(lái)了 for d := range msgs { n, err := strconv.Atoi(string(d.Body)) failOnError(err, "Failed to convert body to integer") log.Printf(" [.] fib(%d)", n) // 計(jì)算 response := fib(n) // 回答 err = ch.Publish( "", // exchange d.ReplyTo, // routing key false, // mandatory false, // immediate amqp.Publishing{ ContentType: "text/plain", CorrelationId: d.CorrelationId, //序列號(hào) Body: []byte(strconv.Itoa(response)), }) failOnError(err, "Failed to publish a message") // 確認(rèn)回答完畢 d.Ack(false) } }() log.Printf(" [*] Awaiting RPC requests") <-forever }
RPC 請(qǐng)求方:
package main import ( "fmt" "log" "math/rand" "os" "strconv" "strings" "time" "github.com/streadway/amqp" ) /* RPC模式 請(qǐng)求方 */ func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err)) } } func randomString(l int) string { bytes := make([]byte, l) for i := 0; i < l; i++ { bytes[i] = byte(randInt(65, 90)) } return string(bytes) } func randInt(min int, max int) int { return min + rand.Intn(max-min) } func fibonacciRPC(n int) (res int, err error) { 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() // 隊(duì)列聲明 q, err := ch.QueueDeclare( "", // name false, // durable false, // delete when usused true, // exclusive 為真即連接斷開就刪除 false, // noWait nil, // arguments ) failOnError(err, "Failed to declare a queue") msgs, err := ch.Consume( q.Name, // queue "", // consumer true, // auto-ack false, // exclusive 這個(gè)為真,服務(wù)器會(huì)認(rèn)為這是該隊(duì)列唯一的消費(fèi)者 false, // no-local false, // no-wait nil, // args ) failOnError(err, "Failed to register a consumer") corrId := randomString(32) err = ch.Publish( "", // exchange "rpc_queue", // routing key false, // mandatory false, // immediate amqp.Publishing{ ContentType: "text/plain", CorrelationId: corrId, ReplyTo: q.Name, Body: []byte(strconv.Itoa(n)), }) failOnError(err, "Failed to publish a message") for d := range msgs { if corrId == d.CorrelationId { res, err = strconv.Atoi(string(d.Body)) failOnError(err, "Failed to convert body to integer") break } } return } func main() { rand.Seed(time.Now().UTC().UnixNano()) n := bodyFrom(os.Args) log.Printf(" [x] Requesting fib(%d)", n) res, err := fibonacciRPC(n) failOnError(err, "Failed to handle RPC request") log.Printf(" [.] Got %d", res) } func bodyFrom(args []string) int { var s string if (len(args) < 2) || os.Args[1] == "" { s = "30" } else { s = strings.Join(args[1:], " ") } n, err := strconv.Atoi(s) failOnError(err, "Failed to convert arg to integer") return n }
關(guān)于golang實(shí)現(xiàn)連接池的方法就分享到這里了,希望以上內(nèi)容可以對(duì)大家有一定的幫助,可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò),可以把它分享出去讓更多的人看到。
網(wǎng)站欄目:golang實(shí)現(xiàn)連接池的方法
標(biāo)題路徑:http://www.chinadenli.net/article46/iigeeg.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供網(wǎng)站營(yíng)銷、網(wǎng)站設(shè)計(jì)公司、手機(jī)網(wǎng)站建設(shè)、網(wǎng)站內(nèi)鏈、網(wǎng)站維護(hù)、企業(yè)網(wǎng)站制作
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請(qǐng)盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來(lái)源: 創(chuàng)新互聯(lián)