golang使用rabbitmq工作队列

Posted 2022-11-22 吴冬冬

在helloworld那一章讲了一个简单的一个生产者一个消费者的模型，但生产系统中很多时候是多个生产者或者多个消费者同时处理的。

生产者

	body := bodyFrom(os.Args)
	err = ch.Publish("", q.Name, false, false, amqp.Publishing
		DeliveryMode: amqp.Persistent,
		ContentType:  "text/plain",
		Body:         []byte(body),
	)

bodyFrom

func bodyFrom(args []string) string 
	if len(args) > 1 
		return strings.Join(args[1:], " ")
	
	return "default"

消费者

	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	failOnError(err, "Failed to register a consumer")


	go func() 
		for msg := range msgs 
			log.Printf("Received a message: %s", msg.Body)
			dotCount := bytes.Count(msg.Body, []byte("."))
			t := time.Duration(dotCount)
			time.Sleep(t * time.Second)
			log.Printf("Done")
		
	()

默认情况下消息会循环分发到消费者

消息ack

当消费者的autoack为true时，一旦收到消息就会直接把该消息设置为删除状态，如果消息的处理时间之内，消费者挂掉了那么这条消息就会丢失掉。
rabbitmq支持消息ack机制，将autoack设为false，当处理完毕再手动触发ack操作。如果处理消息的过程中挂掉了，那么这条消息就会分发给其他都消费者。

	msgs, err := ch.Consume(q.Name, "", false, false, false, false, nil)
	failOnError(err, "Failed to register a consumer")


	go func() 
		for msg := range msgs 
			log.Printf("Received a message: %s", msg.Body)
			dotCount := bytes.Count(msg.Body, []byte("."))
			t := time.Duration(dotCount)
			time.Sleep(t * time.Second)
			log.Printf("Done")
			msg.Ack(false)
		
	()

消息持久性

之前说的是消费者挂掉不会丢失消息的解决方法，那么如果mq挂掉呢，如果保证mq挂掉都不会丢失消息，那么就是消息持久性做到事情。

消息持久性分为两步

队列的持久性

将队列设为持久的，当mq重启的时候不会丢失队列

q, err := ch.QueueDeclare("workQueue", true, false, false, false, nil)

消息的持久性

发布消息时，设置消息为持久的

	err = ch.Publish("", q.Name, false, false, amqp.Publishing
		DeliveryMode: amqp.Persistent,
		ContentType:  "text/plain",
		Body:         []byte(body),
	)

公平分配

循环分配方式只能做到平均分配，每个消费者一个一次分配，但是平均不一定就是公平，比如有的消费者所在的服务器资源较差的情况下，平均肯定是不合理的，为了做到公平那么使用如下设置即可

err = ch.Qos(1, 0, false)

这段代码的意思是最多同时处理一个，多了先别发给我。