Linux下Golang实现消息队列的完整指南

在Linux环境下用Golang实现消息队列，其实有不少选择。最常用的三种方案——RabbitMQ、Kafka、Redis——各有各的适用场景，下面一个个拆开来看。

1. 使用RabbitMQ

RabbitMQ是消息队列领域的“老牌选手”，支持多种协议，生态成熟，部署起来也相对简单。

安装RabbitMQ

在Linux上安装，直接走包管理器就行：

sudo apt-get update
sudo apt-get install rabbitmq-server

使用Golang客户端库

Go这边最常用的库是 streadway/amqp，安装命令：

go get github.com/streadway/amqp

示例代码

先看一个生产者，负责往队列里发消息：

package main

import (
    "fmt"
    "log"
    "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
        true,    // 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.Println(" [x] Sent %s", body)
}

消费者这边，逻辑也很清晰——监听队列，收到消息就处理：

package main

import (
    "fmt"
    "log"
    "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
        true,    // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "Failed to declare a queue")

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

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

2. 使用Kafka

如果你需要处理海量数据流，或者对高吞吐量有刚需，Kafka是更合适的选择。它本质上是一个分布式流处理平台，但做消息队列也完全够用。

安装Kafka

下载解压，然后启动Zookeeper和Kafka服务器：

wget https://downloads.apache.org/kafka/2.8.0/kafka_2.13-2.8.0.tgz
tar -xzf kafka_2.13-2.8.0.tgz
cd kafka_2.13-2.8.0
bin/zookeeper-server-start.sh config/zookeeper.properties &
bin/kafka-server-start.sh config/server.properties &

使用Golang客户端库

推荐用 confluent-kafka-go，安装：

go get github.com/confluentinc/confluent-kafka-go/kafka

示例代码

生产者示例：

package main

import (
    "fmt"
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
    p, err := kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092"})
    if err != nil {
        panic(err)
    }
    defer p.Close()

    go func() {
        for e := range p.Events() {
            switch ev := e.(type) {
            case *kafka.Message:
                if ev.TopicPartition.Error != nil {
                    fmt.Printf("Delivery failed: %v\n", ev.TopicPartition.Error)
                } else {
                    fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
                }
            }
        }
    }()

    topic := "test-topic"
    p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{
            Topic: &topic, Partition: kafka.PartitionAny},
        Value: []byte("Hello Kafka"),
    }, nil)

    p.Flush(15 * 1000)
}

消费者示例：

package main

import (
    "fmt"
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092",
        "group.id":          "test-consumer-group",
        "auto.offset.reset": "earliest",
    })
    if err != nil {
        panic(err)
    }
    defer c.Close()

    c.SubscribeTopics([]string{"test-topic"}, nil)

    for {
        msg, err := c.ReadMessage(-1)
        if err == nil {
            fmt.Printf("Received message: %s\n", string(msg.Value))
        } else {
            fmt.Printf("Consumer error: %v\n", err)
        }
    }
}

3. 使用Redis

Redis虽然以缓存闻名，但它的发布/订阅功能实现一个轻量级消息队列绰绰有余，特别适合对实时性要求不高、系统复杂度想控制得低一些的场景。

安装Redis

sudo apt-get update
sudo apt-get install redis-server

使用Golang客户端库

用 go-redis 库，安装：

go get github.com/go-redis/redis/v8

示例代码

生产者：

package main

import (
    "context"
    "fmt"
    "github.com/go-redis/redis/v8"
)

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "", // no password set
        DB:       0,  // use default DB
    })

    err := rdb.Publish(ctx, "channel", "Hello Redis").Err()
    if err != nil {
        panic(err)
    }
    fmt.Println("Message published")
}

消费者：

package main

import (
    "context"
    "fmt"
    "github.com/go-redis/redis/v8"
)

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "",
        DB:       0,
    })

    pubsub := rdb.Subscribe(ctx, "channel")
    defer pubsub.Close()

    ch := pubsub.Channel()
    for msg := range ch {
        fmt.Printf("Received message: %s\n", msg.Payload)
    }
}

这三种方案各有侧重：RabbitMQ胜在协议丰富、功能完善；Kafka强在吞吐量和持久化；Redis胜在简单、部署成本低。实际项目里怎么选，取决于你的业务场景——是追求可靠性、吞吐量，还是图个轻量快速。上面这些代码直接跑起来就能用，剩下的就看你的具体需求了。