Golang实现RabbitMQ五种模式
使用的依赖包
github.com/streadway/amqp
创建RabbitMQ实例
package RabbitMQ
import (
"fmt"
"github.com/streadway/amqp"
"log"
)
// 用户名 密码 ip:端口/虚拟机
const MQURL = "amqp://admin:123456@127.0.0.1:5672/test"
type RabbitMQ struct {
conn *amqp.Connection
channel *amqp.Channel
//队列名称
QueueName string
//交换机
Exchange string
//key
key string
//连接信息
MqUrl string
}
//创建RabbitMQ结构体实例
func NewRabbitMQ(queueName, exchange, key string) *RabbitMQ {
rabbitmq := &RabbitMQ{QueueName: queueName, Exchange: exchange, key: key, MqUrl: MQURL}
var err error
//创建RabbitMQ连接
rabbitmq.conn, err = amqp.Dial(rabbitmq.MqUrl)
rabbitmq.failOnErr(err, "创建连接错误!")
rabbitmq.channel, err = rabbitmq.conn.Channel()
rabbitmq.failOnErr(err, "获取channel失败!")
return rabbitmq
}
//断开channel和connection
func (r *RabbitMQ) Destroy() {
r.channel.Close()
r.conn.Close()
}
//错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
if err != nil {
log.Fatalf("%s:%s", message, err)
panic(fmt.Sprintf("%s,%s", message, err))
}
}
Simple模式
Simple模式工作流程:
//创建简单模式下RabbitMQ实例
func NewRabbitMQSimple(queueName string) *RabbitMQ {
return NewRabbitMQ(queueName, "", "")
}
//简单模式下生产代码
func (r *RabbitMQ) PublishSimple(message string) {
//1.申请队列,如果队列不存在会自动创建,如果存在则跳过创建
//保证队列存在,消息队列能发送到队列中
_, err := r.channel.QueueDeclare(
r.QueueName,
//是否持久化
false,
//是否为自动删除
false,
//是否具有排他性
false,
//是否阻塞
false,
//额外属性
nil,
)
if err != nil {
fmt.Println("QueueDeclare:", err)
}
//2.发送消息到队列中
err = r.channel.Publish(
r.Exchange,
r.QueueName,
//如果为true,根据exchange类型和routekey规则,如果无法找到符合条件的队列那么会把发送的消息返回给发送者
false,
//如果为true,当exchange发送消息队列到队列后发现队列上没有绑定消费者,则会把消息发还给发送者
false,
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
if err != nil {
fmt.Println("Publish:", err)
}
}
//简单模式下消费代码
func (r *RabbitMQ) ConsumeSimple() {
//1.申请队列,如果队列不存在会自动创建,如果存在则跳过创建
//保证队列存在,消息队列能发送到队列中
_, err := r.channel.QueueDeclare(
r.QueueName,
//是否持久化
false,
//是否为自动删除
false,
//是否具有排他性
false,
//是否阻塞
false,
//额外属性
nil)
if err != nil {
fmt.Println("QueueDeclare:", err)
}
//2.接受消息
msgs, err := r.channel.Consume(
r.QueueName,
//用来区分多个消费者
"",
//是否自动应答
true,
//是否具有排他性
false,
//如果设置为true,表示不能将同一个connection中发送消息传递给这个connection中的消费者
false,
//队列消费是否阻塞
false,
nil)
if err != nil {
fmt.Println("Consume:", err)
}
forever := make(chan bool)
//3.启用协程处理消息
go func() {
for d := range msgs {
//实现我们要处理的逻辑函数
log.Printf("Received a message:%s", d.Body)
}
}()
log.Printf("[*] Waiting for messages,To exit press CTRL+C\n")
<-forever
}
简单模式publish
package main
import (
"fmt"
"go-rabbitmq/RabbitMQ"
)
func main() {
rabbitmq := RabbitMQ.NewRabbitMQSimple("test")
rabbitmq.PublishSimple("myxy99.cn msg")
fmt.Println("发送成功!")
}
简单模式recevie
package main
import "rabbitmq/RabbitMQ"
func main() {
rabbitmq := RabbitMQ.NewRabbitMQSimple("test")
rabbitmq.ConsumeSimple()
}
Work模式
Work模式工作流程:
simple模式和work模式其实用的是一套逻辑代码,只是work模式是可以有多个消费者的,work模式起到一个负载均衡的作用。
work模式publish
package main
import (
"fmt"
"rabbitmq/RabbitMQ"
"strconv"
"time"
)
func main() {
rabbitmq := RabbitMQ.NewRabbitMQSimple("test")
for i := 0; i <= 100; i++ {
rabbitmq.PublishSimple("Hello test!" + strconv.Itoa(i))
time.Sleep(1 * time.Second)
fmt.Println(i)
}
}
work模式receive1
package main
import "rabbitmq/RabbitMQ"
func main() {
rabbitmq := RabbitMQ.NewRabbitMQSimple("test")
rabbitmq.ConsumeSimple()
}
work模式receive2
package main
import "rabbitmq/RabbitMQ"
func main() {
rabbitmq := RabbitMQ.NewRabbitMQSimple("test")
rabbitmq.ConsumeSimple()
}
订阅模式
订阅模式工作流程:
订阅模式的特别是:一个消息被投递到多个队列,一个消息能被多个消费者获取。过程是由生产者将消息发送到exchange(交换机)里,然后exchange通过一系列的规则发送到队列上,然后由绑定对应的消费者进行消息。
//订阅模式创建RabbitMQ实例
func NewRabbitMQPubSub(exchangeName string) *RabbitMQ {
//创建RabbitMQ实例
rabbitmq := NewRabbitMQ("",exchangeName,"")
var err error
//获取connection
rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
rabbitmq.failOnErr(err,"failed to connect rabbitmq!")
//获取channel
rabbitmq.channel, err = rabbitmq.conn.Channel()
rabbitmq.failOnErr(err, "failed to open a channel")
return rabbitmq
}
//订阅模式生产
func (r *RabbitMQ) PublishPub(message string) {
//1.尝试创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
"fanout",
true,
false,
//true表示这个exchange不可以被client用来推送消息,仅用来进行exchange和exchange之间的绑定
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an excha"+
"nge")
//2.发送消息
err = r.channel.Publish(
r.Exchange,
"",
false,
false,
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
}
//订阅模式消费端代码
func (r *RabbitMQ) RecieveSub() {
//1.试探性创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//交换机类型
"fanout",
true,
false,
//YES表示这个exchange不可以被client用来推送消息,仅用来进行exchange和exchange之间的绑定
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an exch"+
"ange")
//2.试探性创建队列,这里注意队列名称不要写
q, err := r.channel.QueueDeclare(
"", //随机生产队列名称
false,
false,
true,
false,
nil,
)
r.failOnErr(err, "Failed to declare a queue")
//绑定队列到 exchange 中
err = r.channel.QueueBind(
q.Name,
//在pub/sub模式下,这里的key要为空
"",
r.Exchange,
false,
nil)
//消费消息
messges, err := r.channel.Consume(
q.Name,
"",
true,
false,
false,
false,
nil,
)
forever := make(chan bool)
go func() {
for d := range messges {
log.Printf("Received a message: %s", d.Body)
}
}()
fmt.Println("退出请按 CTRL+C\n")
<-forever
}
订阅模式publish
package main
import (
"rabbitmq/RabbitMQ"
"strconv"
"time"
"fmt"
)
func main() {
rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")
for i := 0; i < 100; i++ {
rabbitmq.PublishPub("订阅模式生产第" +
strconv.Itoa(i) + "条" + "数据")
fmt.Println("订阅模式生产第" +
strconv.Itoa(i) + "条" + "数据")
time.Sleep(1 * time.Second)
}
}
订阅模式receive1
package main
import "rabbitmq/RabbitMQ"
func main() {
rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")
rabbitmq.RecieveSub()
}
订阅模式receive2
package main
import "rabbitmq/RabbitMQ"
func main() {
rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")
rabbitmq.RecieveSub()
}
路由模式
路由模式工作流程:
路由模式:一个消息由多个消费者消费的基础上指定由哪些消息者来消费。
func NewRabbitMQRouting(exchangeName string,routingKey string) *RabbitMQ {
//创建RabbitMQ实例
rabbitmq := NewRabbitMQ("",exchangeName,routingKey)
var err error
//获取connection
rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
rabbitmq.failOnErr(err,"failed to connect rabbitmq!")
//获取channel
rabbitmq.channel, err = rabbitmq.conn.Channel()
rabbitmq.failOnErr(err, "failed to open a channel")
return rabbitmq
}
//路由模式发送消息
func (r *RabbitMQ) PublishRouting(message string ) {
//1.尝试创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//要改成direct
"direct",
true,
false,
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an exchange")
//2.发送消息
err = r.channel.Publish(
r.Exchange,
//要设置
r.Key,
false,
false,
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
}
//路由模式接受消息
func (r *RabbitMQ) RecieveRouting() {
//1.试探性创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//交换机类型
"direct",
true,
false,
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an exch"+
"ange")
//2.试探性创建队列,这里注意队列名称不要写
q, err := r.channel.QueueDeclare(
"", //随机生产队列名称
false,
false,
true,
false,
nil,
)
r.failOnErr(err, "Failed to declare a queue")
//绑定队列到 exchange 中
err = r.channel.QueueBind(
q.Name,
//需要绑定key
r.Key,
r.Exchange,
false,
nil)
//消费消息
messges, err := r.channel.Consume(
q.Name,
"",
true,
false,
false,
false,
nil,
)
forever := make(chan bool)
go func() {
for d := range messges {
log.Printf("Received a message: %s", d.Body)
}
}()
fmt.Println("退出请按 CTRL+C\n")
<-forever
}
路由模式publish
package main
import (
"rabbitmq/RabbitMQ"
"strconv"
"time"
"fmt"
)
func main() {
mqOne:=RabbitMQ.NewRabbitMQRouting("myxy99","myxy99_one")
mqTwo:=RabbitMQ.NewRabbitMQRouting("myxy99","myxy99_two")
for i := 0; i <= 10; i++ {
mqOne.PublishRouting("Hello myxy99 one!" + strconv.Itoa(i))
mqTwo.PublishRouting("Hello myxy99 Two!" + strconv.Itoa(i))
time.Sleep(1 * time.Second)
fmt.Println(i)
}
}
路由模式receive-one
package main
import "rabbitmq/RabbitMQ"
func main() {
mq:=RabbitMQ.NewRabbitMQRouting("myxy99","myxy99_one")
mq.RecieveRouting()
}
路由模式receive-two
package main
import "rabbitmq/RabbitMQ"
func main() {
mq:=RabbitMQ.NewRabbitMQRouting("myxy99","myxy99_two")
mq.RecieveRouting()
}
话题模式
话题模式工作流程:
话题模式:话题模式是在路由模式上演化而来。不同的是我们以通配符的方式来指定我们的消费者。
//话题模式
//创建RabbitMQ实例
func NewRabbitMQTopic(exchangeName string,routingKey string) *RabbitMQ {
//创建RabbitMQ实例
rabbitmq := NewRabbitMQ("",exchangeName,routingKey)
var err error
//获取connection
rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
rabbitmq.failOnErr(err,"failed to connect rabbitmq!")
//获取channel
rabbitmq.channel, err = rabbitmq.conn.Channel()
rabbitmq.failOnErr(err, "failed to open a channel")
return rabbitmq
}
//话题模式发送消息
func (r *RabbitMQ) PublishTopic(message string ) {
//1.尝试创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//要改成topic
"topic",
true,
false,
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an excha"+
"nge")
//2.发送消息
err = r.channel.Publish(
r.Exchange,
//要设置
r.Key,
false,
false,
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
}
//话题模式接受消息
//要注意key,规则
//其中“*”用于匹配一个单词,“#”用于匹配多个单词(可以是零个)
//匹配 myxy99.* 表示匹配 myxy99.hello, 但是myxy99.hello.one需要用myxy99.#才能匹配到
func (r *RabbitMQ) RecieveTopic() {
//1.试探性创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//交换机类型
"topic",
true,
false,
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an exch"+
"ange")
//2.试探性创建队列,这里注意队列名称不要写
q, err := r.channel.QueueDeclare(
"", //随机生产队列名称
false,
false,
true,
false,
nil,
)
r.failOnErr(err, "Failed to declare a queue")
//绑定队列到 exchange 中
err = r.channel.QueueBind(
q.Name,
//在pub/sub模式下,这里的key要为空
r.Key,
r.Exchange,
false,
nil)
//消费消息
messges, err := r.channel.Consume(
q.Name,
"",
true,
false,
false,
false,
nil,
)
forever := make(chan bool)
go func() {
for d := range messges {
log.Printf("Received a message: %s", d.Body)
}
}()
fmt.Println("退出请按 CTRL+C\n")
<-forever
}
话题模式publish
package main
import (
"rabbitmq/RabbitMQ"
"strconv"
"time"
"fmt"
)
func main() {
mqOne:=RabbitMQ.NewRabbitMQTopic("myxy99Topic","myxy99.topic.one")
mqTwo:=RabbitMQ.NewRabbitMQTopic("myxy99Topic","myxy99.topic.two")
for i := 0; i <= 10; i++ {
mqOne.PublishTopic("Hello myxy99 topic one!" + strconv.Itoa(i))
mqTwo.PublishTopic("Hello myxy99 topic Two!" + strconv.Itoa(i))
time.Sleep(1 * time.Second)
fmt.Println(i)
}
}
话题模式receive-all
package main
import "rabbitmq/RabbitMQ"
func main() {
mq:=RabbitMQ.NewRabbitMQTopic("myxy99Topic","#")
mq.RecieveTopic()
}
话题模式receive-two
package main
import "rabbitmq/RabbitMQ"
func main() {
mq:=RabbitMQ.NewRabbitMQTopic("myxy99Topic","myxy99.*.two")
mq.RecieveTopic()
}
GitHub地址:https://github.com/myxy99/Go-RabbitMQ
本文使用CC BY-NC-SA 3.0 中国大陆协议许可
本文链接:https://myxy99.cn/posts/golang/4b90efe7.html