使用Go语言在RabbitMQ中实现发布/订阅消息中间件模型

2023-09-08 13:56:40

简介
消息中间件是一种允许应用程序之间交换消息的软件。它提供了可靠、可扩展和异步的消息传递。发布/订阅模型是一种消息中间件模型，其中发布者将消息发送到交换机，然后交换机将消息路由到订阅者。

基本概念

在发布/订阅模型中，有三个主要组件：

发布者 ：发布者是发送消息的应用程序。
交换机 ：交换机是接收消息并将其路由到订阅者的组件。
订阅者 ：订阅者是接收消息的应用程序。

代码示例

以下是一个使用Go语言在RabbitMQ中实现发布/订阅模型的代码示例：

package main

import (
	"fmt"
	"log"
	"os"

	"github.com/streadway/amqp"
)

func main() {
	// 创建一个AMQP连接
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()

	// 创建一个信道
	ch, err := conn.Channel()
	if err != nil {
		log.Fatal(err)
	}
	defer ch.Close()

	// 创建一个交换机
	err = ch.ExchangeDeclare(
		"logs", // 交换机名称
		"fanout", // 交换机类型
		true,     // 是否持久化
		false,    // 是否自动删除
		false,    // 是否内部
		false,    // 是否没有等待者时自动删除
		nil,      // 其他属性
	)
	if err != nil {
		log.Fatal(err)
	}

	// 创建一个队列
	q, err := ch.QueueDeclare(
		"", // 队列名称（由RabbitMQ自动生成）
		false, // 是否持久化
		false, // 是否自动删除
		true,  // 是否独占
		false, // 是否没有消费者时自动删除
		nil,   // 其他属性
	)
	if err != nil {
		log.Fatal(err)
	}

	// 将队列绑定到交换机
	err = ch.QueueBind(
		q.Name, // 队列名称
		"", // 路由键（由于是广播，因此留空）
		"logs", // 交换机名称
		false,  // 是否没有匹配的路由键时将消息转发给下一个交换机
		nil,    // 其他属性
	)
	if err != nil {
		log.Fatal(err)
	}

	// 创建一个发布者
	messages := []string{"Hello, world!", "How are you?"}
	for _, message := range messages {
		err = ch.Publish(
			"logs", // 交换机名称
			"",     // 路由键（由于是广播，因此留空）
			false,  // 是否强制
			false,  // 是否立即
			amqp.Publishing{
				ContentType: "text/plain",
				Body:        []byte(message),
			},
		)
		if err != nil {
			log.Fatal(err)
		}

		fmt.Println("Sent message:", message)
	}

	// 创建一个订阅者
	msgs, err := ch.Consume(
		q.Name, // 队列名称
		"", // 消费者名称（由RabbitMQ自动生成）
		true,  // 是否自动确认
		false, // 是否独占
		false, // 是否没有等待者时自动删除
		false, // 是否内部
		nil,   // 其他属性
	)
	if err != nil {
		log.Fatal(err)
	}

	// 接收消息
	forever := make(chan bool)
	go func() {
		for d := range msgs {
			fmt.Println("Received message:", string(d.Body))
		}
	}()

	fmt.Println("Waiting for messages. Press Ctrl+C to exit.")
	<-forever

	// 关闭信道
	err = ch.Close()
	if err != nil {
		log.Fatal(err)
	}

	// 关闭连接
	err = conn.Close()
	if err != nil {
		log.Fatal(err)
	}
}