Introduction to how golang operates Redis&Mysql&RabbitMQ
golangOperation Redis&Mysql&RabbitMQ:
Reids
Installation import
go get github.com/garyburd/redigo/redis import "github.com/garyburd/redigo/redis"
Use
Connection
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
Batch setting
_, 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 operation
_, 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
Set expiration time
_, err = c.Do("expire", "names", 5)
if err != nil {
fmt.Println("expire error: ", err)
return
}lpush & lpop & llen
Queue
// 队列
_, 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
}Connection pool
Each The explanation of the parameters is as follows:
MaxIdle: The maximum number of idle connections, which means that even if there is no redis connection, N idle connections can still be maintained without being cleared and in a standby state at any time.
MaxActive: The maximum number of active connections, indicating that there are at most N connections at the same time
IdleTimeout: The maximum idle connection waiting time, after this time, the idle connection will be closed
pool := &redis.Pool{
MaxIdle: 16,
MaxActive: 1024,
IdleTimeout: 300,
Dial: func() (redis.Conn, error) {
return redis.Dial("tcp", "localhost:6379")
},
}Connection pool example:
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)
}Pipeline operation
The request/response service can continuously process new requests. The client can send multiple commands to the server without waiting for a response, and finally in one Read multiple responses.
Use the Send(), Flush(), and Receive() methods to support pipelined operations
Send writes commands to the connection's output buffer.
Flush clears the connection's output buffer and writes it to the server.
Recevie reads the server's response sequentially in FIFO order.
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
Installation import
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" )
Connection
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
}Build table
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")
}Transaction
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
Installation
go get "github.com/streadway/amqp"
Normal mode
Producer:
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
}Consumer:
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
}Subscription mode
Subscription producer:
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
}Subscription consumer:
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 mode
RPC Responder:
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 Requester:
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
} Recommended: golang tutorial
The above is the detailed content of Introduction to how golang operates Redis&Mysql&RabbitMQ. For more information, please follow other related articles on the PHP Chinese website!
Hot AI Tools
Undresser.AI Undress
AI-powered app for creating realistic nude photos
AI Clothes Remover
Online AI tool for removing clothes from photos.
Undress AI Tool
Undress images for free
Clothoff.io
AI clothes remover
AI Hentai Generator
Generate AI Hentai for free.
Hot Article
Hot Tools
Notepad++7.3.1
Easy-to-use and free code editor
SublimeText3 Chinese version
Chinese version, very easy to use
Zend Studio 13.0.1
Powerful PHP integrated development environment
Dreamweaver CS6
Visual web development tools
SublimeText3 Mac version
God-level code editing software (SublimeText3)
Hot Topics
How to safely read and write files using Golang?
Jun 06, 2024 pm 05:14 PM
Reading and writing files safely in Go is crucial. Guidelines include: Checking file permissions Closing files using defer Validating file paths Using context timeouts Following these guidelines ensures the security of your data and the robustness of your application.
How to configure connection pool for Golang database connection?
Jun 06, 2024 am 11:21 AM
How to configure connection pooling for Go database connections? Use the DB type in the database/sql package to create a database connection; set MaxOpenConns to control the maximum number of concurrent connections; set MaxIdleConns to set the maximum number of idle connections; set ConnMaxLifetime to control the maximum life cycle of the connection.
Comparison of advantages and disadvantages of golang framework
Jun 05, 2024 pm 09:32 PM
The Go framework stands out due to its high performance and concurrency advantages, but it also has some disadvantages, such as being relatively new, having a small developer ecosystem, and lacking some features. Additionally, rapid changes and learning curves can vary from framework to framework. The Gin framework is a popular choice for building RESTful APIs due to its efficient routing, built-in JSON support, and powerful error handling.
What are the best practices for error handling in Golang framework?
Jun 05, 2024 pm 10:39 PM
Best practices: Create custom errors using well-defined error types (errors package) Provide more details Log errors appropriately Propagate errors correctly and avoid hiding or suppressing Wrap errors as needed to add context
How to save JSON data to database in Golang?
Jun 06, 2024 am 11:24 AM
JSON data can be saved into a MySQL database by using the gjson library or the json.Unmarshal function. The gjson library provides convenience methods to parse JSON fields, and the json.Unmarshal function requires a target type pointer to unmarshal JSON data. Both methods require preparing SQL statements and performing insert operations to persist the data into the database.
Golang framework vs. Go framework: Comparison of internal architecture and external features
Jun 06, 2024 pm 12:37 PM
The difference between the GoLang framework and the Go framework is reflected in the internal architecture and external features. The GoLang framework is based on the Go standard library and extends its functionality, while the Go framework consists of independent libraries to achieve specific purposes. The GoLang framework is more flexible and the Go framework is easier to use. The GoLang framework has a slight advantage in performance, and the Go framework is more scalable. Case: gin-gonic (Go framework) is used to build REST API, while Echo (GoLang framework) is used to build web applications.
What are the common dependency management issues in the Golang framework?
Jun 05, 2024 pm 07:27 PM
Common problems and solutions in Go framework dependency management: Dependency conflicts: Use dependency management tools, specify the accepted version range, and check for dependency conflicts. Vendor lock-in: Resolved by code duplication, GoModulesV2 file locking, or regular cleaning of the vendor directory. Security vulnerabilities: Use security auditing tools, choose reputable providers, monitor security bulletins and keep dependencies updated.
Detailed practical explanation of golang framework development: Questions and Answers
Jun 06, 2024 am 10:57 AM
In Go framework development, common challenges and their solutions are: Error handling: Use the errors package for management, and use middleware to centrally handle errors. Authentication and authorization: Integrate third-party libraries and create custom middleware to check credentials. Concurrency processing: Use goroutines, mutexes, and channels to control resource access. Unit testing: Use gotest packages, mocks, and stubs for isolation, and code coverage tools to ensure sufficiency. Deployment and monitoring: Use Docker containers to package deployments, set up data backups, and track performance and errors with logging and monitoring tools.
