golang操作Redis&Mysql&RabbitMQ:
Reids
安装导入
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
批量设置
_, 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
设置过期时间
_, err = c.Do("expire", "names", 5) if err != nil { fmt.Println("expire error: ", err) return }
lpush & lpop & llen
队列
// 队列 _, 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 }
连接池
各参数的解释如下:
MaxIdle:最大的空闲连接数,表示即使没有redis连接时依然可以保持N个空闲的连接,而不被清除,随时处于待命状态。
MaxActive:最大的激活连接数,表示同时最多有N个连接
IdleTimeout:最大的空闲连接等待时间,超过此时间后,空闲连接将被关闭
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) }
管道操作
请求/响应服务可以实现持续处理新请求,客户端可以发送多个命令到服务器而无需等待响应,最后在一次读取多个响应。
使用Send(),Flush(),Receive()方法支持管道化操作
Send向连接的输出缓冲中写入命令。
Flush将连接的输出缓冲清空并写入服务器端。
Recevie按照FIFO顺序依次读取服务器的响应。
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
安装导入
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") }
事务
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"
普通模式
生产者:
package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" "time")/*默认点对点模式*/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() // 申明一个队列 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") body := bodyFrom(os.Args) // 发布 err = ch.Publish( "", // exchange 默认模式,exchange为空 q.Name, // routing key 默认模式路由到同名队列,即是task_queue false, // mandatory false, amqp.Publishing{ // 持久性的发布,因为队列被声明为持久的,发布消息必须加上这个(可能不用),但消息还是可能会丢,如消息到缓存但MQ挂了来不及持久化。 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 }
消费者:
package main import ( "bytes" "fmt" "github.com/streadway/amqp" "log" "time" ) /* 默认点对点模式 工作方,多个,拿发布方的消息 */ 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() // 指定队列! 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 预取,每个工作方每次拿一个消息,确认后才拿下一次,缓解压力 err = ch.Qos( 1, // prefetch count 0, // prefetch size false, // global ) failOnError(err, "Failed to set QoS") // 消费根据队列名 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() { 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") // 确认消息被收到!!如果为真的,那么同在一个channel,在该消息之前未确认的消息都会确认,适合批量处理 // 真时场景:每十条消息确认一次,类似 d.Ack(false) } }() log.Printf(" [*] Waiting for messages. To exit press CTRL+C") <-forever }
订阅模式
订阅 生产者:
package main import ( "fmt" "github.com/streadway/amqp" "log" "os" "strings" "time" ) /* 广播模式 发布方 */ 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() // 默认模式有默认交换机,广播自己定义一个交换机,交换机可与队列进行绑定 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) // 发布 err = ch.Publish( "logs", // exchange 消息发送到交换机,这个时候没队列绑定交换机,消息会丢弃 "", // routing key 广播模式不需要这个,它会把所有消息路由到绑定的所有队列 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 }
订阅 消费者:
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() // 同样要申明交换机 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") // 新建队列,这个队列没名字,随机生成一个名字 q, err := ch.QueueDeclare( "", // name false, // durable false, // delete when usused true, // exclusive 表示连接一断开,这个队列自动删除 false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare a queue") // 队列和交换机绑定,即是队列订阅了发到这个交换机的消息 err = ch.QueueBind( q.Name, // queue name 队列的名字 "", // routing key 广播模式不需要这个 "logs", // exchange 交换机名字 false, nil) failOnError(err, "Failed to bind 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") 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 应答方:
package main import ( "fmt" "log" "strconv" "github.com/streadway/amqp" ) /* RPC模式 应答方 */ 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") // 公平分发 没有这个则round-robbin err = ch.Qos( 1, // prefetch count 0, // prefetch size false, // global ) failOnError(err, "Failed to set QoS") // 消费,等待请求 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() { //请求来了 for d := range msgs { n, err := strconv.Atoi(string(d.Body)) failOnError(err, "Failed to convert body to integer") log.Printf(" [.] fib(%d)", n) // 计算 response := fib(n) // 回答 err = ch.Publish( "", // exchange d.ReplyTo, // routing key false, // mandatory false, // immediate amqp.Publishing{ ContentType: "text/plain", CorrelationId: d.CorrelationId, //序列号 Body: []byte(strconv.Itoa(response)), }) failOnError(err, "Failed to publish a message") // 确认回答完毕 d.Ack(false) } }() log.Printf(" [*] Awaiting RPC requests") <-forever }
RPC 请求方:
package main import ( "fmt" "log" "math/rand" "os" "strconv" "strings" "time" "github.com/streadway/amqp" ) /* RPC模式 请求方 */ 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() // 队列声明 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 这个为真,服务器会认为这是该队列唯一的消费者 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 }
推荐:golang教程
以上是golang操作Redis&Mysql&RabbitMQ的方法介绍的详细内容。更多信息请关注PHP中文网其他相关文章!

Golang和C 在性能竞赛中的表现各有优势:1)Golang适合高并发和快速开发,2)C 提供更高性能和细粒度控制。选择应基于项目需求和团队技术栈。

Golang适合快速开发和并发编程,而C 更适合需要极致性能和底层控制的项目。1)Golang的并发模型通过goroutine和channel简化并发编程。2)C 的模板编程提供泛型代码和性能优化。3)Golang的垃圾回收方便但可能影响性能,C 的内存管理复杂但控制精细。

GoimpactsdevelopmentPositationalityThroughSpeed,效率和模拟性。1)速度:gocompilesquicklyandrunseff,ifealforlargeprojects.2)效率:效率:ITScomprehenSevestAndArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdArdEcceSteral Depentencies,增强开发的简单性:3)SimpleflovelmentIcties:3)简单性。

C 更适合需要直接控制硬件资源和高性能优化的场景,而Golang更适合需要快速开发和高并发处理的场景。1.C 的优势在于其接近硬件的特性和高度的优化能力,适合游戏开发等高性能需求。2.Golang的优势在于其简洁的语法和天然的并发支持,适合高并发服务开发。

Golang在实际应用中表现出色,以简洁、高效和并发性着称。 1)通过Goroutines和Channels实现并发编程,2)利用接口和多态编写灵活代码,3)使用net/http包简化网络编程,4)构建高效并发爬虫,5)通过工具和最佳实践进行调试和优化。

Go语言的核心特性包括垃圾回收、静态链接和并发支持。1.Go语言的并发模型通过goroutine和channel实现高效并发编程。2.接口和多态性通过实现接口方法,使得不同类型可以统一处理。3.基本用法展示了函数定义和调用的高效性。4.高级用法中,切片提供了动态调整大小的强大功能。5.常见错误如竞态条件可以通过gotest-race检测并解决。6.性能优化通过sync.Pool重用对象,减少垃圾回收压力。

Go语言在构建高效且可扩展的系统中表现出色,其优势包括:1.高性能:编译成机器码,运行速度快;2.并发编程:通过goroutines和channels简化多任务处理;3.简洁性:语法简洁,降低学习和维护成本;4.跨平台:支持跨平台编译,方便部署。

关于SQL查询结果排序的疑惑学习SQL的过程中,常常会遇到一些令人困惑的问题。最近,笔者在阅读《MICK-SQL基础�...


热AI工具

Undresser.AI Undress
人工智能驱动的应用程序,用于创建逼真的裸体照片

AI Clothes Remover
用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool
免费脱衣服图片

Clothoff.io
AI脱衣机

AI Hentai Generator
免费生成ai无尽的。

热门文章

热工具

mPDF
mPDF是一个PHP库,可以从UTF-8编码的HTML生成PDF文件。原作者Ian Back编写mPDF以从他的网站上“即时”输出PDF文件,并处理不同的语言。与原始脚本如HTML2FPDF相比,它的速度较慢,并且在使用Unicode字体时生成的文件较大,但支持CSS样式等,并进行了大量增强。支持几乎所有语言,包括RTL(阿拉伯语和希伯来语)和CJK(中日韩)。支持嵌套的块级元素(如P、DIV),

Atom编辑器mac版下载
最流行的的开源编辑器

EditPlus 中文破解版
体积小,语法高亮,不支持代码提示功能

PhpStorm Mac 版本
最新(2018.2.1 )专业的PHP集成开发工具

WebStorm Mac版
好用的JavaScript开发工具