Go 使用第三方库实现消息队列功能

56 min read

在Golang中，有很多第三方库可以实现消息队列功能。其中比较流行的有：

  1. RabbitMQ：RabbitMQ是一个开源的、高度可伸缩的消息队列系统，支持多种消息传输协议，如AMQP、AMQPS、MQTT等。可以通过Go语言的第三方库github.com/streadway/amqp来与RabbitMQ进行交互。

  2. NSQ：NSQ是一个分布式实时消息平台，可以稳定地处理高并发的消息流。可以通过Go语言的第三方库github.com/nsqio/go-nsq来与NSQ进行交互。

  3. Kafka：Kafka是一个分布式的、支持发布-订阅和处理流式数据的消息系统。可以通过Go语言的第三方库github.com/Shopify/sarama来与Kafka进行交互。

下面是一个使用RabbitMQ实现消息队列的例子：

package main

import (
	"fmt"
	"log"
	"os"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%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(
		"hello", // name
		false,   // durable
		false,   // delete when unused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)
	failOnError(err, "Failed to declare a queue")

	body := "Hello World!"
	err = ch.Publish(
		"",     // exchange
		q.Name, // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "Failed to publish a message")
	fmt.Printf(" [x] Sent %s\n", body)

	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("Received a message: %s", d.Body)
		}
	}()

	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}

在这个例子中，我们使用github.com/streadway/amqp库连接到RabbitMQ，并声明了一个名称为"hello"的队列。然后，我们通过ch.Publish函数将一条字符串消息发布到该队列中。最后，我们通过ch.Consume函数注册一个消费者来消费该队列中的消息，并打印出所有接收到的消息。

需要注意的是，在实际使用中，我们一般会将这些操作封装成一个消息队列客户端，供其他代码调用。