64 lines
1.4 KiB
Go
64 lines
1.4 KiB
Go
package rabbitmq
|
|
|
|
import (
|
|
"fmt"
|
|
"github.com/nacos-group/nacos-sdk-go/v2/common/logger"
|
|
"github.com/streadway/amqp"
|
|
)
|
|
|
|
// RabbitMQConn is a struct that holds the connection and channel objects
|
|
type RabbitMQConn struct {
|
|
Connection *amqp.Connection
|
|
Channel *amqp.Channel
|
|
}
|
|
|
|
type ConnectProperty struct {
|
|
ExchangeName string
|
|
QueueName string
|
|
ExchangeType string
|
|
TopicKey string
|
|
}
|
|
|
|
// Send 向RabbitMQ中发送消息
|
|
func Send(conn *RabbitMQConn, connProp *ConnectProperty, message []byte) {
|
|
// 往哪里发
|
|
channel := conn.Channel
|
|
|
|
// 发送
|
|
err := channel.Publish(
|
|
connProp.ExchangeName,
|
|
connProp.TopicKey,
|
|
false,
|
|
false,
|
|
amqp.Publishing{
|
|
ContentType: "text/plain",
|
|
Body: message,
|
|
},
|
|
)
|
|
if err != nil {
|
|
logger.Error(fmt.Sprintf("Failed to publish a message: %v", err))
|
|
}
|
|
}
|
|
|
|
func Read(conn *RabbitMQConn, connProp *ConnectProperty, autoAck bool) <-chan amqp.Delivery {
|
|
|
|
// 拿到特定的Channel
|
|
channel := conn.Channel
|
|
|
|
// 开始读取队列中的全部消息
|
|
msgs, err := channel.Consume(
|
|
connProp.QueueName, // 队列名称
|
|
"", // 消费者名称
|
|
autoAck, // auto-ack
|
|
false, // exclusive
|
|
false, // no-local
|
|
false, // no-wait
|
|
nil, // arguments
|
|
)
|
|
if err != nil {
|
|
log.Error(fmt.Sprintf("Failed to register a consumer: %v", err))
|
|
}
|
|
|
|
return msgs
|
|
}
|