[每周一更]-(第38期)：Go常见的操作消息队列

在消费者示例中，我们通过 nc.Subscribe 方法订阅 my-topic 主题，并使用一个回调函数处理接收到的消息。在消费者示例中，我们通过 nsq.NewConsumer 方法创建一个 NSQ 消费者，在实际生产环境中，我们需要在程序中添加正确的退出逻辑。在选择消息队列时，需要根据具体的业务需求和性能要求进行选择，并且需要考虑安全性、可靠性和扩展性等因素，确保消息传递的可靠性和性能。我们有

ifanatic

955人浏览 · 2023-12-22 15:43:19

ifanatic · 2023-12-22 15:43:19 发布

在这里插入图片描述

在Go语言中，常见的消息队列有以下几种：

RabbitMQ：RabbitMQ是一个开源的AMQP（高级消息队列协议）消息代理软件，用于支持多种编程语言，包括Go语言。RabbitMQ提供了可靠的消息传递机制和灵活的路由规则，可以用于处理大量的消息和任务。
Apache Kafka：Apache Kafka是一个开源的分布式流处理平台，也可以用作消息队列，用于处理高容量的消息流和实时数据。Kafka提供了高吞吐量、低延迟的消息传递机制，并且具有良好的可伸缩性和可靠性。
NSQ：NSQ是一个开源的实时分布式消息平台，用于处理大规模的消息和数据流。NSQ提供了高可靠性和低延迟的消息传递机制，并且具有良好的可扩展性和可伸缩性。
NATS：NATS是一个轻量级、高性能的消息系统，用于支持分布式应用程序和微服务。NATS提供了简单易用的API和协议，具有高可靠性、低延迟和高吞吐量的消息传递机制。
Redis：Redis是一个开源的内存数据库，也可以用作消息队列。Redis提供了支持发布订阅模式、阻塞队列等特性，可以用于处理实时数据和大量的消息。
ActiveMQ：ActiveMQ是一个开源的消息代理软件，用于支持多种消息传递协议和编程语言。ActiveMQ提供了高可靠性、可伸缩性和可扩展性的消息传递机制，可以用于处理大规模的消息和任务。

除了以上常见的消息队列，还有一些其他的开源消息系统和组件，例如RocketMQ、ZeroMQ等，也可以用于处理消息和任务。

在选择消息队列时，需要根据具体的业务需求和性能要求进行选择，并且需要考虑安全性、可靠性和扩展性等因素，确保消息传递的可靠性和性能。

消息队列的使用场景有哪些？

不同的消息队列适用于不同的场景，以下是常见的使用场景：

RabbitMQ：RabbitMQ适用于需要可靠的消息传递和灵活的路由规则的场景，例如电商网站的订单处理、银行的支付处理等。
Apache Kafka：Kafka适用于处理大规模的消息和数据流，例如社交媒体的实时消息、大型网站的日志处理等。
NSQ：NSQ适用于需要高可靠性和低延迟的场景，例如在线游戏的实时消息、金融交易的实时处理等。
NATS：NATS适用于需要高可靠性、低延迟和高吞吐量的场景，例如移动互联网应用的实时通信、物联网设备的数据传输等。
Redis：Redis适用于需要快速处理大量消息的场景，例如在线聊天、实时数据分析等。
ActiveMQ：ActiveMQ适用于需要支持多种消息传递协议和编程语言的场景，例如企业应用集成、分布式系统的消息传递等。

当然，这些场景只是一些常见的示例，具体的使用场景还需要根据业务需求和性能要求来选择。需要根据消息队列的特性、优缺点和性能指标进行评估和比较，选择最适合自己业务需求的消息队列。

使用示例

操作RabbitMQ

我们有一个需求，需要向多个客户端发送消息，可以使用RabbitMQ作为消息队列，Go作为开发语言。

1、安装RabbitMQ并启动服务。

2、安装amqp库：


go get github.com/streadway/amqp

3、生产者向消息队列中发送消息：


package main

import (
	"fmt"
	"log"

	"github.com/streadway/amqp"
)

func main() {
	// 连接RabbitMQ服务器
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %s", err)
	}
	defer conn.Close()

	// 创建一个channel
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Failed to open a channel: %s", err)
	}
	defer ch.Close()

	// 声明一个名为"hello"的queue
	q, err := ch.QueueDeclare(
		"hello", // 队列名
		false,   // 是否持久化
		false,   // 是否自动删除
		false,   // 是否具有排他性
		false,   // 是否阻塞
		nil,     // 额外的参数
	)
	if err != nil {
		log.Fatalf("Failed to declare a queue: %s", err)
	}

	// 发送消息到队列
	body := "Hello, World!"
	err = ch.Publish(
		"",     // exchange名
		q.Name, // queue名
		false,  // 是否强制发送
		false,  // 是否立即发送
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		},
	)
	if err != nil {
		log.Fatalf("Failed to publish a message: %s", err)
	}

	fmt.Println("Message sent successfully")
}

4、消费者从消息队列中获取消息并处理：


package main

import (
	"log"

	"github.com/streadway/amqp"
)

func main() {
	// 连接RabbitMQ服务器
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %s", err)
	}
	defer conn.Close()

	// 创建一个channel
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Failed to open a channel: %s", err)
	}
	defer ch.Close()

	// 声明一个名为"hello"的queue
	q, err := ch.QueueDeclare(
		"hello", // 队列名
		false,   // 是否持久化
		false,   // 是否自动删除
		false,   // 是否具有排他性
		false,   // 是否阻塞
		nil,     // 额外的参数
	)
	if err != nil {
		log.Fatalf("Failed to declare a queue: %s", err)
	}

	// 消费队列中的消息
	msgs, err := ch.Consume(
		q.Name, // queue名
		"",     // 消费者名
		true,   // 是否自动应答

操作Kafka

我们使用 github.com/segmentio/kafka-go 库作为 Kafka Go 客户端。在生产者示例中，我们通过 kafka.DialLeader 方法连接到 Kafka 集群，
然后使用 conn.WriteMessages 方法向 Kafka 集群发送消息。在消费者示例中，我们通过 kafka.NewReader 方法创建一个 Kafka 消费者，
然后通过 r.ReadMessage 方法从 Kafka 集群读取消息。

在示例中，我们还使用了一个 sigchan 信号通道来监听操作系统的信号并退出程序。

安装 Kafka Go 客户端：

go get -u github.com/segmentio/kafka-go

生产者示例：


package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/segmentio/kafka-go"
)

func main() {
    topic := "my-topic"
    partition := 0
    conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
    if err != nil {
        log.Fatal("failed to dial leader:", err)
    }
    defer conn.Close()

    // 发送消息
    msg := kafka.Message{
        Value: []byte("Hello, Kafka!"),
    }
    _, err = conn.WriteMessages(msg)
    if err != nil {
        log.Fatal("failed to write message:", err)
    }

    fmt.Println("message sent")
}

消费者示例：

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"

    "github.com/segmentio/kafka-go"
)

func main() {
    topic := "my-topic"
    partition := 0
    offset := kafka.LastOffset

    r := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{"localhost:9092"},
        Topic:   topic,
        Partition: partition,
        MinBytes: 10e3, // 10KB
        MaxBytes: 10e6, // 10MB
        MaxWait:  10 * time.Second,
    })

    // 接收消息
    sigchan := make(chan os.Signal, 1)
    signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

    for {
        select {
        case <-sigchan:
            log.Println("received signal, exiting...")
            return
        default:
            msg, err := r.ReadMessage(context.Background())
            if err != nil {
                log.Fatal("failed to read message:", err)
            }

            fmt.Println(string(msg.Value))
        }
    }
}

操作NSQ

使用 github.com/nsqio/go-nsq 库作为 NSQ Go 客户端。在生产者示例中，我们通过 nsq.NewProducer 方法创建一个 NSQ 生产者，

然后使用 producer.Publish 方法向 NSQ 集群发送消息。在消费者示例中，我们通过 nsq.NewConsumer 方法创建一个 NSQ 消费者，

然后使用 consumer.AddHandler 方法设置消息处理函数。最后，我们通过 consumer.ConnectToNSQLookupd 方法连接到 NSQ 集群，

并使用 select {} 语句保持消费者程序不退出。

需要注意的是，上述示例中我们使用了一个空的 select {} 语句来保持消费者程序不退出。在实际生产环境中，我们需要在程序中添加正确的退出逻辑。

例如，使用一个 sigchan 信号通道来监听操作系统的信号并退出程序。

安装 NSQ Go 客户端：


安装 NSQ Go 客户端：

生产者示例：

package main

import (
    "fmt"

    "github.com/nsqio/go-nsq"
)

func main() {
    producer, err := nsq.NewProducer("localhost:4150", nsq.NewConfig())
    if err != nil {
        panic(err)
    }

    // 发送消息
    err = producer.Publish("my-topic", []byte("Hello, NSQ!"))
    if err != nil {
        panic(err)
    }

    producer.Stop()

    fmt.Println("message sent")
}

消费者示例：

package main

import (
    "fmt"
    "log"

    "github.com/nsqio/go-nsq"
)

type MyHandler struct{}

func (*MyHandler) HandleMessage(msg *nsq.Message) error {
    fmt.Println(string(msg.Body))
    return nil
}

func main() {
    consumer, err := nsq.NewConsumer("my-topic", "my-channel", nsq.NewConfig())
    if err != nil {
        panic(err)
    }

    consumer.AddHandler(&MyHandler{})

    err = consumer.ConnectToNSQLookupd("localhost:4161")
    if err != nil {
        panic(err)
    }

    fmt.Println("consumer started")

    select {}
}

操作NATS

使用 github.com/nats-io/nats.go 库作为 NATS Go 客户端。在生产者示例中，我们通过 nats.Connect 方法创建一个 NATS 连接，

然后使用 nc.Publish 方法向 NATS 集群发送消息。在消费者示例中，我们通过 nc.Subscribe 方法订阅 my-topic 主题，并使用一个回调函数处理接收到的消息。

最后，我们使用一个空的 select {} 语句保持消费者程序不退出。

需要注意的是，上述示例中我们使用了一个空的 select {} 语句来保持消费者程序不退出。在实际生产环境中，

我们需要在程序中添加正确的退出逻辑。例如，使用一个 sigchan 信号通道来监听操作系统的信号并退出程序。

安装 NATS Go 客户端：

go get github.com/nats-io/nats.go

生产者示例：

package main

import (
    "fmt"
    "time"

    "github.com/nats-io/nats.go"
)

func main() {
    nc, err := nats.Connect("nats://localhost:4222")
    if err != nil {
        panic(err)
    }
    defer nc.Close()

    // 发送消息
    err = nc.Publish("my-topic", []byte("Hello, NATS!"))
    if err != nil {
        panic(err)
    }

    fmt.Println("message sent")
}

消费者示例：

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/nats-io/nats.go"
)

func main() {
    nc, err := nats.Connect("nats://localhost:4222")
    if err != nil {
        panic(err)
    }
    defer nc.Close()

    // 订阅消息
    _, err = nc.Subscribe("my-topic", func(msg *nats.Msg) {
        fmt.Println(string(msg.Data))
    })
    if err != nil {
        panic(err)
    }

    fmt.Println("consumer started")

    select {}
}

操作Redis

安装 Redis Go 客户端：

go get github.com/go-redis/redis

连接 Redis：

package main

import (
    "fmt"

    "github.com/go-redis/redis"
)

func main() {
    // 创建 Redis 客户端
    client := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "", // 没有设置密码
        DB:       0,  // 使用默认数据库
    })

    // 检查 Redis 是否正常连接
    pong, err := client.Ping().Result()
    fmt.Println(pong, err)
}

Redis String 类型操作：

package main

import (
    "fmt"

    "github.com/go-redis/redis"
)

func main() {
    // 创建 Redis 客户端
    client := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "", // 没有设置密码
        DB:       0,  // 使用默认数据库
    })

    // 设置字符串
    err := client.Set("key", "value", 0).Err()
    if err != nil {
        panic(err)
    }

    // 获取字符串
    val, err := client.Get("key").Result()
    if err != nil {
        panic(err)
    }
    fmt.Println("key", val)

    // 删除字符串
    err = client.Del("key").Err()
    if err != nil {
        panic(err)
    }
}

Redis List 类型操作：

package main

import (
    "fmt"

    "github.com/go-redis/redis"
)

func main() {
    // 创建 Redis 客户端
    client := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "", // 没有设置密码
        DB:       0,  // 使用默认数据库
    })

    // 将元素添加到列表
    err := client.RPush("list", "a", "b", "c").Err()
    if err != nil {
        panic(err)
    }

    // 获取列表长度
    length, err := client.LLen("list").Result()
    if err != nil {
        panic(err)
    }
    fmt.Println("list length:", length)

    // 获取列表中的元素
    val, err := client.LRange("list", 0, -1).Result()
    if err != nil {
        panic(err)
    }
    fmt.Println("list elements:", val)

    // 弹出列表左侧元素
    elem, err := client.LPop("list").Result()
    if err != nil {
        panic(err)
    }
    fmt.Println("popped element:", elem)
}

操作ActiveMQ

在Go中操作ActiveMQ，可以使用go-stomp库。以下是一个简单的示例代码，用于连接到ActiveMQ，并向队列发送消息：

package main

import (
	"fmt"
	"github.com/go-stomp/stomp"
)

func main() {
	conn, err := stomp.Dial("tcp", "localhost:61613")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer conn.Disconnect()

	msg := "hello, activemq"
	err = conn.Send("/queue/test", "text/plain", []byte(msg), nil)
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Printf("Message sent: %s\n", msg)
}

在该示例中，我们通过stomp.Dial()方法连接到ActiveMQ的默认端口61613。然后，我们使用conn.Send()方法向队列“/queue/test”发送消息。

要从队列中接收消息，请使用conn.Subscribe()方法。以下是一个示例代码：

package main

import (
	"fmt"
	"github.com/go-stomp/stomp"
)

func main() {
	conn, err := stomp.Dial("tcp", "localhost:61613")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer conn.Disconnect()

	sub, err := conn.Subscribe("/queue/test", stomp.AckAuto)
	if err != nil {
		fmt.Println(err)
		return
	}
	defer sub.Unsubscribe()

	for {
		msg := <-sub.C
		fmt.Printf("Received message: %s\n", string(msg.Body))
	}
}

在该示例中，我们使用conn.Subscribe()方法订阅队列“/queue/test”。然后，我们通过使用sub.C通道来接收来自队列的消息。收到消息后，我们使用string(msg.Body)将其转换为字符串，并打印到控制台上。

需要注意的是，这只是一个简单的示例代码。在实际应用中，您需要考虑诸如异常处理、连接丢失、重连等问题。