亲宝软件园·资讯

展开

GoLang RabbitMQ实现六种工作模式示例

Onemorelight95 人气:0

六种工作模式介绍

1.简单(Simple)模式

P:生产者,也就是要发送消息的程序。

C:消费者:消息的接收者,会一直等待消息到来。

queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。

简单模式就是单发单收,消息的消费者监听消息队列,如果队列中有消息,就消费掉,消息被拿走后,自动从队列中删除。

2.工作队列(Work Queue)模式

这种模式就是多个消费者消费同一个队列中的消息,既然消费者多了那么就出现了消息分配的问题,所以对应着两种分配策略:

3.发布/订阅(Pub/Sub)模式

在这种模型中,多了一个 Exchange 角色,而且过程略有变化:

P:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给X (交换机)。

C:消费者,消息的接收者,会一直等待消息到来。

Queue:消息队列,接收消息、缓存消息。

Exchange:交换机(X) ,一方面,接收生产者发送的消息。另一方面,如何处理消息,递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。 Exchange有以下4种类型:

注意:Exchange负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失。

4.路由(Routing)模式

路由模式其实就是上述发布/订阅模式的交换机转发类型变成了Direct类型。在这种模式下:

Exchange 不再把消息交给每一个绑定的队列,而是根据消息的routing key进行判断,只有队列的

routing key与消息的routing key完全一致,才会接收到消息。

P:生产者,向 Exchange 发送消息,发送消息时,会指定一个routing key

X:Exchange(交换机),接收生产者的消息,然后把消息递交给与routing key完全匹配的队列。

C1:消费者,其所在队列指定了需要routing key为error的消息。

C2:消费者,其所在队列指定了需要routing key为 info、error、warning 的消息。

5.通配符(Tpoic)模式

路由模式其实就是上述发布/订阅模式的交换机转发类型变成了Topic类型。在这种模式下:

队列的routing key与消息的routing key符合匹配规则,就可以接收到消息,有两种规则:

*:可以(只能)匹配一个单词。

#:可以匹配多个单词(或者零个)。

所以图中,routing keya.orange.b的消息就会被转发到Q1,而routing keyLazy.a.b.c的消息就会被发送到Q2。

Go语言的实现

安装操作库

安装API库

Go可以使用streadway/amqp库来操作rabbit,使用以下命令来安装:

go get github.com/streadway/amqp

封装rabbitmq

接下来我们对streadway/amqp库的内容进行一个二次封装,封装为一个rabbitmq.go文件:

package rabbitmq
import (
	"encoding/json"
	"github.com/streadway/amqp"
	"log"
)
// RabbitMQ RabbitMQ结构
type RabbitMQ struct {
	channel  *amqp.Channel
	Name     string
	exchange string
}
// Connect 连接服务器
func Connect(s string) *RabbitMQ {
	//连接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "连接Rabbitmq服务器失败!")
	ch, e := conn.Channel()
	failOnError(e, "无法打开频道!")
	mq := new(RabbitMQ)
	mq.channel = ch
	return mq
}
// New 初始化消息队列
//第一个参数:rabbitmq服务器的链接,第二个参数:队列名字
func New(s string, name string) *RabbitMQ {
	//连接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "连接Rabbitmq服务器失败!")
	ch, e := conn.Channel()
	failOnError(e, "无法打开频道!")
	q, e := ch.QueueDeclare(
		name,  //队列名
		false, //是否开启持久化
		true,  //不使用时删除
		false, //排他
		false, //不等待
		nil,   //参数
	)
	failOnError(e, "初始化消息队列失败!")
	mq := new(RabbitMQ)
	mq.channel = ch
	mq.Name = q.Name
	return mq
}
// QueueDeclare 声明queue
func (q *RabbitMQ) QueueDeclare(queue string) {
	_, e := q.channel.QueueDeclare(queue, false, true, false, false, nil)
	failOnError(e, "声明queue失败!")
}
// QueueDelete 删除queue
func (q *RabbitMQ) QueueDelete(queue string) {
	_, e := q.channel.QueueDelete(queue, false, true, false)
	failOnError(e, "删除queue失败!")
}
// Qos 配置queue参数
func (q *RabbitMQ) Qos() {
	e := q.channel.Qos(1, 0, false)
	failOnError(e, "无法设置QoS")
}
// NewExchange 初始化交换机
//第一个参数:rabbitmq服务器的链接,第二个参数:交换机名字,第三个参数:交换机类型
func NewExchange(s string, name string, typename string) {
	//连接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "连接Rabbitmq服务器失败!")
	ch, e := conn.Channel()
	failOnError(e, "无法打开频道!")
	e = ch.ExchangeDeclare(
		name,     // name
		typename, // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(e, "初始化交换机失败!")
}
// ExchangeDelete 删除交换机
func (q *RabbitMQ) ExchangeDelete(exchange string) {
	e := q.channel.ExchangeDelete(exchange, false, true)
	failOnError(e, "删除交换机失败!")
}
// Bind 绑定消息队列到exchange
func (q *RabbitMQ) Bind(exchange string, key string) {
	e := q.channel.QueueBind(
		q.Name,
		key,
		exchange,
		false,
		nil,
	)
	failOnError(e, "绑定队列失败!")
	q.exchange = exchange
}
// Send 向消息队列发送消息
//Send方法可以往某个消息队列发送消息
func (q *RabbitMQ) Send(queue string, body interface{}) {
	str, e := json.Marshal(body)
	failOnError(e, "消息序列化失败!")
	e = q.channel.Publish(
		"",    //交换
		queue, //路由键
		false, //必填
		false, //立即
		amqp.Publishing{
			ReplyTo: q.Name,
			Body:    []byte(str),
		})
	msg := "向队列:" + q.Name + "发送消息失败!"
	failOnError(e, msg)
}
// Publish 向exchange发送消息
//Publish方法可以往某个exchange发送消息
func (q *RabbitMQ) Publish(exchange string, body interface{}, key string) {
	str, e := json.Marshal(body)
	failOnError(e, "消息序列化失败!")
	e = q.channel.Publish(
		exchange,
		key,
		false,
		false,
		amqp.Publishing{ReplyTo: q.Name,
			Body: []byte(str)},
	)
	failOnError(e, "向交换机发送消息失败!")
}
// Consume 接收某个消息队列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
	c, e := q.channel.Consume(
		q.Name, //指定从哪个队列中接收消息
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	failOnError(e, "接收消息失败!")
	return c
}
// Close 关闭队列连接
func (q *RabbitMQ) Close() {
	q.channel.Close()
}
//错误处理函数
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

简单(Simple)模式

生产者

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "simple")
	i := 0
	for {
		// 每隔2s发送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("simple", " simple message: "+strconv.Itoa(i))
		i = i + 1
	}
}

消费者

func main() {
	consumer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "simple")
	//接收消息时,指定
	messages := consumer.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

运行结果:

2022/11/05 18:54:47 Received a message: " simple message: 0"
2022/11/05 18:54:52 Received a message: " simple message: 1"
2022/11/05 18:54:57 Received a message: " simple message: 2"

工作队列(Work Queue)模式

公平分发模式:

公平分发模式采用的是轮询机制,它会将数个任务按顺序平均分发给消费者。

生产者

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	i := 0
	for {
		// 每隔2s发送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("worker", " worker message: "+strconv.Itoa(i))
		i = i + 1
	}
}

消费者1

func main() {
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	//接收消息
	messages := consumer1.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

消费者2

func main() {
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	//接收消息
	messages := consumer2.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

运行结果:

# 消费者1
2022/11/05 19:45:03 Received a message: " worker message: 0"
2022/11/05 19:45:07 Received a message: " worker message: 2"
2022/11/05 19:45:11 Received a message: " worker message: 4"

# 消费者2
2022/11/05 19:45:05 Received a message: " worker message: 1"
2022/11/05 19:45:09 Received a message: " worker message: 3"
2022/11/05 19:45:13 Received a message: " worker message: 5"

可以发现,公平模式下,偶数消息都被发送给了消费者1,而奇数消息都被发送给了消费者2。

公平派遣模式:

有时候,如果消息之间的复杂度不同,那么不同消费者消费消息所用的时间会不同。这个时候如果使用公平派发模式,可能会造成某一个消费者需要消费的消息积压过多。可以采用公平派遣模式:

公平派遣模式下发送端与公平分发相同,消费者端只需要加一段配置代码,我们可以将预取计数设置为1。这告诉RabbitMQ一次不要给消费者一个以上的消息。换句话说,在处理并确认上一条消息之前,不要将新消息发送给消费者。而是将其分派给不忙的下一个消费者。

关于消息的确认:

为了确保消息永不丢失,RabbitMQ支持 消息确认。消费者发送回一个确认(acknowledgement),以告知RabbitMQ已经接收,处理了特定的消息,并且RabbitMQ可以自由删除它。

我们之前的代码中,RabbitMQ一旦向消费者传递了一条消息,便立即将其标记为删除(调用Consumer的第三个参数是autoAck,表示是否自动回复)。在这种情况下,如果你终止一个消费者那么你就可能会丢失这个任务,我们还将丢失所有已经交付给这个消费者的尚未消费的消息。如果一个消费者意外宕机了,那么我们希望将任务交付给其他消费者来消费者。

所以一旦向消费者传递了一条消息,就不能马上将其标记为删除,而是要手动确认。我们需要在创建消费者的时候将autoAck参数标记为false:

// Consume 接收某个消息队列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
	c, e := q.channel.Consume(
		q.Name, //指定从哪个队列中接收消息
		"",
		false, // 不自动确认消息
		false,
		false,
		false,
		nil,
	)
	failOnError(e, "接收消息失败!")
	return c
}

然后每消费完一条消息需要调用Ack(false)函数手动回复。

生产者

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	i := 0
	for {
		// 每隔2s发送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("worker", " worker message: "+strconv.Itoa(i))
		i = i + 1
	}
}

消费端限流:

实现公平派遣模式我们需要设置消费者端一次只能消费一条消息,之前我们已经进行了封装,直接在消费者端调用即可:

// Qos 配置queue参数
func (q *RabbitMQ) Qos() {
	e := q.channel.Qos(1, 0, false)
	failOnError(e, "无法设置QoS")
}

消费者1

func main() {
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	// 指定一次只消费一条消息,直到消费完才重新接收
	consumer1.Qos()
	//接收消息
	messages := consumer1.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用10s
			time.Sleep(time.Second * 10)
			// 手动回复
			ch.Ack(false)
		}
	}()
	select {}
}

消费者2

func main() {
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	// 指定一次只消费一条消息,直到消费完才重新接收
	consumer2.Qos()
	//接收消息
	messages := consumer2.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用2s
			time.Sleep(time.Second * 2)
			// 手动回复
			ch.Ack(false)
		}
	}()
	select {}
}

运行结果:

# 消费者1
2022/11/05 20:31:26 Received a message: " worker message: 0"
2022/11/05 20:31:36 Received a message: " worker message: 5"

# 消费者2
2022/11/05 20:31:28 Received a message: " worker message: 1"
2022/11/05 20:31:30 Received a message: " worker message: 2"
2022/11/05 20:31:32 Received a message: " worker message: 3"
2022/11/05 20:31:34 Received a message: " worker message: 4"
2022/11/05 20:31:38 Received a message: " worker message: 6"
2022/11/05 20:31:40 Received a message: " worker message: 7"

发布/订阅(Pub/Sub)模式

生产者

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange1", "fanout")
	i := 0
	for {
		time.Sleep(time.Second)
		// fanout模式下不用routing key
		producer.Publish("exchange1", "pubsub message: "+strconv.Itoa(i), "")
		i = i + 1
	}
}

消费者1

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 队列绑定到exchange
	consumer1.Bind("exchange1", "")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

消费者2

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 队列绑定到exchange
	consumer2.Bind("exchange1", "")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

运行结果:

# 消费者1
2022/11/05 22:32:19 Consumer1 received a message: "pubsub message: 0"
2022/11/05 22:32:20 Consumer1 received a message: "pubsub message: 1"
2022/11/05 22:32:21 Consumer1 received a message: "pubsub message: 2"
2022/11/05 22:32:22 Consumer1 received a message: "pubsub message: 3"
2022/11/05 22:32:23 Consumer1 received a message: "pubsub message: 4"
2022/11/05 22:32:24 Consumer1 received a message: "pubsub message: 5"

# 消费者2
2022/11/05 22:32:19 Consumer2 received a message: "pubsub message: 0"
2022/11/05 22:32:20 Consumer2 received a message: "pubsub message: 1"
2022/11/05 22:32:21 Consumer2 received a message: "pubsub message: 2"
2022/11/05 22:32:22 Consumer2 received a message: "pubsub message: 3"
2022/11/05 22:32:23 Consumer2 received a message: "pubsub message: 4"
2022/11/05 22:32:24 Consumer2 received a message: "pubsub message: 5"

路由(Routing)模式

生产者

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	// 指定为direct类型
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange", "direct")
	i := 0
	for {
		time.Sleep(time.Second)
		// 如果是奇数,就发key1
		// 如果是偶数,就发key2
		if i%2 != 0 {
			producer.Publish("exchange", "routing message: "+strconv.Itoa(i), "key1")
		} else {
			producer.Publish("exchange", "routing message: "+strconv.Itoa(i), "key2")
		}
		i = i + 1
	}
}

消费者1

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 队列绑定到exchange
	consumer1.Bind("exchange", "key1")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

消费者2

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 队列绑定到exchange
	consumer2.Bind("exchange", "key2")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

运行结果:

# 消费者1
2022/11/05 22:51:10 Consumer1 received a message: "routing message: 1"
2022/11/05 22:51:12 Consumer1 received a message: "routing message: 3"
2022/11/05 22:51:14 Consumer1 received a message: "routing message: 5"
2022/11/05 22:51:16 Consumer1 received a message: "routing message: 7"

# 消费者2
2022/11/05 22:51:11 Consumer2 received a message: "routing message: 0"
2022/11/05 22:51:13 Consumer2 received a message: "routing message: 2"
2022/11/05 22:51:15 Consumer2 received a message: "routing message: 4"
2022/11/05 22:51:17 Consumer2 received a message: "routing message: 6"

通配符(Tpoic)模式

生产者

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	// 指定为topic类型
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange2", "topic")
	var i int
	for {
		time.Sleep(time.Second)
		if i%2 != 0 {
			producer.Publish("exchange2", "topic message: "+strconv.Itoa(i), "a.test.b.c")
		} else {
			producer.Publish("exchange2", "topic message: "+strconv.Itoa(i), "a.test.b")
		}
		i++
	}
}

消费者1

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 队列绑定到exchange
	consumer1.Bind("exchange2", "*.test.*")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

消费者2

func main() {
	//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 队列绑定到exchange
	consumer2.Bind("exchange2", "#.test.#")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

运行结果:

# 消费者1
2022/11/05 23:09:53 Consumer1 received a message: "topic message: 0"
2022/11/05 23:09:55 Consumer1 received a message: "topic message: 2"
2022/11/05 23:09:57 Consumer1 received a message: "topic message: 4"
2022/11/05 23:09:59 Consumer1 received a message: "topic message: 6"

# 消费者2
2022/11/05 23:09:53 Consumer2 received a message: "topic message: 0"
2022/11/05 23:09:54 Consumer2 received a message: "topic message: 1"
2022/11/05 23:09:55 Consumer2 received a message: "topic message: 2"
2022/11/05 23:09:56 Consumer2 received a message: "topic message: 3"
2022/11/05 23:09:57 Consumer2 received a message: "topic message: 4"
2022/11/05 23:09:58 Consumer2 received a message: "topic message: 5"
2022/11/05 23:09:59 Consumer2 received a message: "topic message: 6"

加载全部内容

相关教程
猜你喜欢
用户评论