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

2022-12-18 10:12 开发作者： Onemorelight95

六种工作模式介绍

1.简单（Simple）模式

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

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

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

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

2.工作队列（Work Queue）模式

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

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

公平分发：每个消费者接收消息的概率是相等的，消息队列会循环依次给每个消费者发送消息，这种是默认的策略。
公平派遣：保证消费者在消费完某个消息，并发送确认信息后，消息队列才会向它推送新的消息，在此之间若是有新的消息，将会被推送到其它消费者，若所有的消费者都在消费消息，那么就会等待。

3.发布/订阅（Pub/Sub）模式

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

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

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

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

Queue：消息队列，接收消息、缓存消息。

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

Fanout：广播，将消息交给所有绑定到交换机的队列。
Direct：全值匹配，把消息交给符合指定routing key的队列。
Topic：通配符，与Direct类型类似，但Direct类型要求routing key完全相等，而Topic类型是对routing key进行模糊匹配，比Direct灵活。
Headers：根据Message的一些头部信息来分发过滤Message，用的比较少。

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

4.路由（Routing）模式

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

路由模式其实就是上述发布/订阅模式的交换机转发类型变成了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）模式

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

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

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

*：可以（只能）匹配一个单词。

#：可以匹配多个单词（或者零个）。

所以图中，routing key为a.orange.b的消息就会被转发到Q1，而routing key为Lazy.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")www.devze.com
	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 := ran编程客栈ge msgs {
			log.Printf("Cojavascriptnsumer1 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/android11/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"

到此这篇关于golang RabbitMQ实现六种工作模式示例的文章就介绍到这了,更多相关GoLang RabbitMQ工作模式内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！

继续阅读：GoLangRabbitMQ工作模式 RabbitMQ工作模式

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

目录

六种工作模式介绍

1.简单（Simple）模式

2.工作队列（Work Queue）模式

3.发布/订阅（Pub/Sub）模式

4.路由（Routing）模式

5.通配符（Tpoic）模式

Go语言的实现

安装操作库

简单（Simple）模式

工作队列（Work Queue）模式

发布/订阅（Pub/Sub）模式

路由（Routing）模式

通配符（Tpoic）模式

更多精彩内容

精彩评论

最新开发

SpringBoot整合Liquibase实现对数据库管理和迁移

java8升级java21方法步骤(完整pom文件升级及代码修改)

javacv ffmpeg使用原生ffmpeg命令方式

golang进行xml文件解析的操作方法

如何使用go实现创建WebSocket服务器

开发排行榜

springboot后端存储富文本内容的思路与步骤(含图片内容)

PyCharm运行python测试,报错“没有发现测试”/“空套件”的解决

return base64.b64encode(b).decode(

基于C语言实现钻石棋游戏的示例代码

Sublime Text 3解决中文乱码问题（实测可用）