在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胜在简单、部署成本低。实际项目里怎么选，取决于你的业务场景——是追求可靠性、吞吐量，还是图个轻量快速。上面这些代码直接跑起来就能用，剩下的就看你的具体需求了。
