1,在生产者端初始化mq连接
package rabbitmq
import (
"fmt"
"log"
"github.com/streadway/amqp"
)
var (
conn *amqp.Config
)
func InitRabbitMq() {
// 连接RabbitMQ服务器
conn, err := amqp.Dial("amqp://guest:guest@你的mq服务器地址:5672/")
if err != nil {
log.Fatalf("Failed to connect to RabbitMQ: %v", err)
}
defer conn.Close()
// 创建一个通道
ch, err := conn.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %v", err)
}
defer ch.Close()
// 声明一个交换机
err = ch.ExchangeDeclare(
"my_exchange", // 交换机名称
"direct", // 交换机类型
true, // 是否持久化
false, // 是否自动删除
false, // 是否内部使用
false, // 是否等待确认
nil, // 其他属性
)
if err != nil {
log.Fatalf("Failed to declare an exchange: %v", err)
}
// 声明一个队列
q, err := ch.QueueDeclare(
"my_queue", // 队列名称
true, // 是否持久化
false, // 是否自动删除
false, // 是否排他
false, // 是否等待确认
nil, // 其他属性
)
if err != nil {
log.Fatalf("Failed to declare a queue: %v", err)
}
// 绑定队列到交换机
err = ch.QueueBind(
q.Name, // 队列名称
"my_routing_key", // 路由键
"my_exchange", // 交换机名称
false, // 是否等待确认
nil, // 其他属性
)
if err != nil {
log.Fatalf("Failed to bind a queue: %v", err)
}
fmt.Println("Exchange and queue created and bound successfully!")
}
2,创建生产者
package api
import (
"fmt"
"github.com/gin-gonic/gin"
"github.com/streadway/amqp"
"log"
)
func RabbitMqPublish(c *gin.Context) {
name := c.Param("name")
fmt.Println(">>>>>>>", name)
// 建立与RabbitMQ服务器的连接
conn, err := amqp.Dial("amqp://guest:guest@你的mq服务器地址:5672/")
if err != nil {
log.Fatalf("Failed to connect to RabbitMQ: %v", err)
}
defer conn.Close()
// 创建一个通道
ch, err := conn.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %v", err)
}
defer ch.Close()
// 声明一个队列
queue, err := ch.QueueDeclare(
"my_queue", // 队列名称
true, // 是否持久化
false, // 是否自动删除
false, // 是否具有排他性
false, // 是否阻塞
nil, // 额外参数
)
if err != nil {
log.Fatalf("Failed to declare a queue: %v", err)
fmt.Println(queue.Name)
}
//绑定队列
//_ = ch.QueueBind("my_queue", "my_routing_key", "my_exchange", false, nil)
// 发布消息到队列
//message := "Hello, RabbitMQ!"
if len(name) < 1 {
name = "空消息"
}
err = ch.Publish(
"my_exchange", // 交换机名称
"my_routing_key", // 路由key
false, // 是否强制
false, // 是否立即
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(name),
},
)
if err != nil {
log.Fatalf("Failed to publish a message: %v", err)
}
log.Println("消息发送成功!!!!!!!!!!")
}
3,另起一个go服务进行消费者消费
文章来源：https://www.toymoban.com/news/detail-823533.html
package mq
import (
"fmt"
"github.com/streadway/amqp"
"log"
)
func MqSubContext() {
// 连接RabbitMQ服务器
conn, err := amqp.Dial("amqp://guest:guest@你的mq服务器地址:5672/")
if err != nil {
log.Fatalf("Failed to connect to RabbitMQ: %v", err)
}
defer conn.Close()
// 创建一个通道
ch, err := conn.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %v", err)
}
defer ch.Close()
// 声明一个队列
queue, err := ch.QueueDeclare(
"my_queue", // 队列名称
true, // 是否持久化
false, // 是否自动删除
false, // 是否具有排他性
false, // 是否阻塞
nil, // 额外参数
)
if err != nil {
log.Fatalf("Failed to declare a queue: %v", err)
}
// 绑定队列到交换机
err = ch.QueueBind(
queue.Name, // 队列名称
"my_routing_key", // routing key
"my_exchange", // 交换机名称
false, // 是否阻塞
nil, // 额外参数
)
if err != nil {
log.Fatalf("Failed to bind a queue: %v", err)
}
// 消费消息
msgs, err := ch.Consume(
queue.Name, // 队列名称
"", // 消费者标识符
true, // 是否自动应答
false, // 是否具有排他性
false, // 是否阻塞
false, // 是否等待
nil, // 额外参数
)
if err != nil {
log.Fatalf("Failed to consume messages: %v", err)
}
// 处理接收到的消息
for msg := range msgs {
fmt.Printf("消费的消息是: %s\n", msg.Body)
}
}
后面将会发布golang整合es操作的文章文章来源地址https://www.toymoban.com/news/detail-823533.html
到了这里，关于golang整合rabbitmq-实现创建生产者绑定交换机-创建消费者消费完整代码的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
