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

news/2024/10/18 18:24:35/

在这里插入图片描述

在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 mainimport ("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()// 创建一个channelch, err := conn.Channel()if err != nil {log.Fatalf("Failed to open a channel: %s", err)}defer ch.Close()// 声明一个名为"hello"的queueq, 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 mainimport ("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()// 创建一个channelch, err := conn.Channel()if err != nil {log.Fatalf("Failed to open a channel: %s", err)}defer ch.Close()// 声明一个名为"hello"的queueq, 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 mainimport ("context""fmt""log""time""github.com/segmentio/kafka-go"
)func main() {topic := "my-topic"partition := 0conn, 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 mainimport ("context""fmt""log""os""os/signal""syscall""github.com/segmentio/kafka-go"
)func main() {topic := "my-topic"partition := 0offset := kafka.LastOffsetr := kafka.NewReader(kafka.ReaderConfig{Brokers: []string{"localhost:9092"},Topic:   topic,Partition: partition,MinBytes: 10e3, // 10KBMaxBytes: 10e6, // 10MBMaxWait:  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...")returndefault: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 mainimport ("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 mainimport ("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 mainimport ("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 mainimport ("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 mainimport ("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 mainimport ("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 mainimport ("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 mainimport ("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 mainimport ("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.Cfmt.Printf("Received message: %s\n", string(msg.Body))}
}

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

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


http://www.ppmy.cn/news/1279070.html

相关文章

(JAVA)-线程中的通信(生产者消费者模型)

(JAVA)-线程中的通信(生产者消费者模型)

在Java线程通信中&#xff0c;等待通知机制是最传统的方式&#xff0c;就是在一个线程进行了规定操作后&#xff0c;该线程就进入等待状态&#xff08;wait&#xff09;&#xff0c; 等待其它线程执行完它们的指定代码过后&#xff0c;再将之前等待的线程唤醒&#xff08;notif…
阅读更多...
Openwrt AP 发射 WiFi 信号

Openwrt AP 发射 WiFi 信号

问题 想一次把 OpenWrt 路由器 wifi 问题给解决&#xff0c;完全取代路由器。 使用 倍控的 N5105 设备&#xff0c;有 mPCIe 接口&#xff0c;使用了 intel AX200 无线网卡&#xff0c;支持 2.4G 与 5G。 设置步骤 OpenWrt 镜像 第一次使用的镜像不支持 wifi&#xff0c;在…
阅读更多...
AI-数学-初高中-因式分解

AI-数学-初高中-因式分解

原教学视频链接:初高衔接】1等式变形方法回顾_哔哩哔哩_bilibili 1.乘法公式&#xff1a; 推导&#xff1a;ab(ab)-2ab 2.十字相乘法因式分解&#xff1a; 3.增添项法因式分解&#xff1a;
阅读更多...
国际站店铺装修外贸平台alibaba鼠标经过渐变的效果功能代码代码生成器店铺装修旺铺旺铺装修生成器阿里巴巴国际站店铺怎么装修首页进入装修后台自定义内容模块

国际站店铺装修外贸平台alibaba鼠标经过渐变的效果功能代码代码生成器店铺装修旺铺旺铺装修生成器阿里巴巴国际站店铺怎么装修首页进入装修后台自定义内容模块

国际站店铺装修外贸平台alibaba鼠标经过渐变的效果功能代码代码生成器店铺装修旺铺旺铺装修生成器阿里巴巴国际站店铺怎么装修首页进入装修后台自定义内容模块 鼠标经过有个渐变的效果功能
阅读更多...
安捷伦DSOX2024A示波器

安捷伦DSOX2024A示波器

参考波形 示波器的非易失参考波形存储器可以存储两个波形。比较这些参考波形与实时波形&#xff0c;并对已存储数据进行后分析和测量。您也可将波形数据存储到移动USB 存储器设备。这些数据还能调用到示波器的两个参考存储器的其中一个&#xff0c;进行全面的波形测量和分析。为…
阅读更多...
Xcode15 iOS 17 Simulator 离线安装，模拟器安装

Xcode15 iOS 17 Simulator 离线安装，模拟器安装

Xcode 15 安装包的大小相比之前更小&#xff0c;因为除了 macOS 的 Components&#xff0c;其他都需要动态下载安装&#xff0c;否则提示 iOS 17 Simulator Not Installed。 如果不安装对应的运行模拟库 无法真机和模拟器运行&#xff0c;更无法新建项目。但是由于模拟器安装包…
阅读更多...
【Skynet 入门实战练习】事件模块 | 批处理模块 | GM 指令 | 模糊搜索

【Skynet 入门实战练习】事件模块 | 批处理模块 | GM 指令 | 模糊搜索

文章目录 前言事件模块批处理模块GM 指令模块模糊搜索最后 前言 本节完善了项目&#xff0c;实现了事件、批处理、模糊搜索模块、GM 指令模块。 事件模块 什么是事件模块&#xff1f;事件模块是用来在各系统之间传递事件消息的。 为什么需要事件模块&#xff1f;主要目的是…
阅读更多...
关于游戏性能优化的技巧

关于游戏性能优化的技巧

关于游戏性能优化的技巧 游戏性能优化对象池Jobs、Burst、多线程间隔处理定时更新全局广播缓存组件缓存常用数据2D残影优化2D骨骼转GPU动画定时器优化DrawCall合批处理优化碰撞层优化粒子特效 游戏性能优化 好久没有在CSDN上面写文章了&#xff0c;今天突然看到鬼谷工作室技术…
阅读更多...
最新文章

Copyright @ 2022~2023