[agent-go] 调整代码结构
This commit is contained in:
zeaslity
@@ -1,63 +0,0 @@
|
||||
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
|
||||
}
|
||||
72
agent-go/rabbitmq/OMsgConnector.go
Normal file
72
agent-go/rabbitmq/OMsgConnector.go
Normal file
@@ -0,0 +1,72 @@
|
||||
package rabbitmq
|
||||
|
||||
import (
|
||||
"agent-go/g"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
var OctopusToServerQueue = &RabbitQueue{}
|
||||
|
||||
func BuildOMsgRuntimeConnectorQueue(agentTopicName string) {
|
||||
|
||||
// 建立 业务消息 接收队列
|
||||
// agentTopicName为名称的队列
|
||||
nacosConfig := g.G.NacosConfig
|
||||
|
||||
octopusExchangeName := nacosConfig.GetString("octopus.message.octopus_exchange")
|
||||
|
||||
octopusConnectProp := &ConnectProperty{
|
||||
ExchangeName: octopusExchangeName,
|
||||
QueueName: agentTopicName,
|
||||
ExchangeType: g.QueueTopic,
|
||||
TopicKey: agentTopicName + "*",
|
||||
}
|
||||
|
||||
octopusMsgQueue := &RabbitQueue{
|
||||
RabbitProp: octopusConnectProp,
|
||||
}
|
||||
octopusMsgQueue.Connect()
|
||||
|
||||
deliveries := octopusMsgQueue.Read(true)
|
||||
|
||||
// 死循环，处理Octopus Message
|
||||
P.Submit(
|
||||
func() {
|
||||
for delivery := range deliveries {
|
||||
|
||||
var om *OctopusMessage
|
||||
err := json.Unmarshal(delivery.Body, &om)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("octopus message convert to json is wrong! msg is => %s", delivery.Body))
|
||||
// 保存到某处
|
||||
continue
|
||||
}
|
||||
|
||||
// 策略模式 处理消息
|
||||
om.Handle()
|
||||
}
|
||||
})
|
||||
|
||||
// 建立 业务消息 返回队列
|
||||
// 统一为 OctopusToServer
|
||||
|
||||
octopusToServerQueueName := nacosConfig.GetString("octopus.message.octopus_to_server")
|
||||
|
||||
octopusToServerProp := &ConnectProperty{
|
||||
ExchangeName: octopusExchangeName,
|
||||
QueueName: octopusToServerQueueName,
|
||||
ExchangeType: g.QueueTopic,
|
||||
TopicKey: octopusToServerQueueName,
|
||||
}
|
||||
|
||||
OctopusToServerQueue = &RabbitQueue{
|
||||
RabbitProp: octopusToServerProp,
|
||||
}
|
||||
|
||||
// 开启运行时消息返回队列
|
||||
OctopusToServerQueue.Connect()
|
||||
|
||||
log.InfoF("Octopus Message Replay Queue is established ! => %v", OctopusToServerQueue)
|
||||
|
||||
}
|
||||
113
agent-go/rabbitmq/OctopusMessage.go
Normal file
113
agent-go/rabbitmq/OctopusMessage.go
Normal file
@@ -0,0 +1,113 @@
|
||||
package rabbitmq
|
||||
|
||||
import (
|
||||
"agent-go/executor"
|
||||
"agent-go/g"
|
||||
"agent-go/utils"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
var P = g.G.P
|
||||
|
||||
type IOctopusMessage interface {
|
||||
OctopusMsgHandler
|
||||
OctopusMsgSender
|
||||
OctopusMsgBuilder
|
||||
}
|
||||
|
||||
type OctopusMsgHandler interface {
|
||||
Handle(octopusMessage *OctopusMessage)
|
||||
}
|
||||
|
||||
type OctopusMsgSender interface {
|
||||
Send(rabbitQueue *RabbitQueue, msg []byte)
|
||||
}
|
||||
|
||||
type OctopusMsgBuilder interface {
|
||||
Build(omType string, content interface{}) *OctopusMessage
|
||||
}
|
||||
|
||||
type OctopusMessage struct {
|
||||
UUID string `json:"uuid"`
|
||||
InitTime time.Time `json:"init_time" format:"2023-03-21 16:38:30"`
|
||||
Type string `json:"type"`
|
||||
Content interface{} `json:"content"`
|
||||
Result interface{} `json:"result"`
|
||||
ACTime time.Time `json:"ac_time" format:"2023-03-21 16:38:30"`
|
||||
}
|
||||
|
||||
func (om *OctopusMessage) Handle() {
|
||||
// 实际执行 OM handle进程
|
||||
doHandleOctopusMessage(om)
|
||||
}
|
||||
|
||||
func (om *OctopusMessage) Send(rabbitQueue *RabbitQueue, msg []byte) {
|
||||
rabbitQueue.Send(msg)
|
||||
}
|
||||
|
||||
func (om *OctopusMessage) Build(omType string, content interface{}) *OctopusMessage {
|
||||
|
||||
// 当前时间
|
||||
curTimeString := utils.CurTimeString()
|
||||
|
||||
// must write to string format, otherwise it's very hard to deserialize
|
||||
bytes, err := json.Marshal(content)
|
||||
if err != nil {
|
||||
fmt.Sprintf("OctopusMessage Build Error ! %v", err)
|
||||
}
|
||||
|
||||
return &OctopusMessage{
|
||||
UUID: curTimeString,
|
||||
InitTime: time.Now(),
|
||||
Type: omType,
|
||||
Content: string(bytes),
|
||||
Result: nil,
|
||||
ACTime: time.Time{},
|
||||
}
|
||||
}
|
||||
|
||||
func doHandleOctopusMessage(octopusMessage *OctopusMessage) {
|
||||
|
||||
switch octopusMessage.Type {
|
||||
case g.InitOmType:
|
||||
go func() {}()
|
||||
case g.ExecOmType:
|
||||
P.Submit(func() {
|
||||
executorOMHandler(octopusMessage)
|
||||
})
|
||||
case g.StatusOmType:
|
||||
P.Submit(func() {
|
||||
statusOMHandler(octopusMessage)
|
||||
})
|
||||
default:
|
||||
P.Submit(func() {
|
||||
blackHoleOMHandler(octopusMessage)
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func executorOMHandler(octopusMessage *OctopusMessage) {
|
||||
|
||||
executionMsgString := octopusMessage.Content.(string)
|
||||
|
||||
var executionMessage *executor.ExecutionMessage
|
||||
err := json.Unmarshal([]byte(executionMsgString), &executionMessage)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("execution message convert to json is wrong! msg is => %s", executionMsgString))
|
||||
return
|
||||
}
|
||||
|
||||
// 交给后端的实际处理器处理， 再次策略
|
||||
executor.Execute(executionMessage)
|
||||
}
|
||||
|
||||
func statusOMHandler(octopusMessage *OctopusMessage) {
|
||||
|
||||
}
|
||||
|
||||
func blackHoleOMHandler(octopusMessage *OctopusMessage) {
|
||||
log.Error(fmt.Sprintf("octopusMessage type wrong! msg is => %v", octopusMessage))
|
||||
}
|
||||
@@ -1,106 +0,0 @@
|
||||
package rabbitmq
|
||||
|
||||
import (
|
||||
"agent-go/config"
|
||||
"agent-go/executor"
|
||||
"agent-go/g"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
var P = g.G.P
|
||||
|
||||
func HandleOMsg(initOMsgFromServer *config.OctopusMessage) {
|
||||
|
||||
agentTopicName := initOMsgFromServer.Result.(string)
|
||||
|
||||
OctopusExchange := g.G.NacosConfig.GetString("octopus.message.octopus_exchange")
|
||||
|
||||
octopusConnectProp := &ConnectProperty{
|
||||
ExchangeName: OctopusExchange,
|
||||
QueueName: agentTopicName,
|
||||
ExchangeType: g.QueueTopic,
|
||||
TopicKey: agentTopicName + "*",
|
||||
}
|
||||
|
||||
octopusConn, err := NewRabbitMQConn(octopusConnectProp)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("Octopus Message Queue create Error ! => %v", octopusConnectProp))
|
||||
panic(err)
|
||||
}
|
||||
|
||||
// 开始接收消息
|
||||
channel := octopusConn.Channel
|
||||
deliveries, err := channel.Consume(
|
||||
agentTopicName,
|
||||
agentTopicName,
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
nil,
|
||||
)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// 死循环，处理Ocotpus Message
|
||||
for delivery := range deliveries {
|
||||
|
||||
var om *config.OctopusMessage
|
||||
err := json.Unmarshal(delivery.Body, &om)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("octopus message convert to json is wrong! msg is => %s", delivery.Body))
|
||||
// 保存到某处
|
||||
continue
|
||||
}
|
||||
|
||||
// 策略模式 处理消息
|
||||
doHandleOctopusMessage(om)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func doHandleOctopusMessage(octopusMessage *config.OctopusMessage) {
|
||||
|
||||
switch octopusMessage.Type {
|
||||
case g.InitOmType:
|
||||
go func() {}()
|
||||
case g.ExecOmType:
|
||||
P.Submit(func() {
|
||||
executorOMHandler(octopusMessage)
|
||||
})
|
||||
case g.StatusOmType:
|
||||
P.Submit(func() {
|
||||
statusOMHandler(octopusMessage)
|
||||
})
|
||||
default:
|
||||
P.Submit(func() {
|
||||
blackHoleOMHandler(octopusMessage)
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func executorOMHandler(octopusMessage *config.OctopusMessage) {
|
||||
|
||||
executionMsgString := octopusMessage.Content.(string)
|
||||
|
||||
var executionMessage *config.ExecutionMessage
|
||||
err := json.Unmarshal([]byte(executionMsgString), &executionMessage)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("execution message convert to json is wrong! msg is => %s", executionMsgString))
|
||||
return
|
||||
}
|
||||
|
||||
// 交给后端的实际处理器处理， 再次策略
|
||||
executor.Execute(octopusMessage, executionMessage)
|
||||
}
|
||||
|
||||
func statusOMHandler(octopusMessage *config.OctopusMessage) {
|
||||
|
||||
}
|
||||
|
||||
func blackHoleOMHandler(octopusMessage *config.OctopusMessage) {
|
||||
log.Error(fmt.Sprintf("octopusMessage type wrong! msg is => %v", octopusMessage))
|
||||
}
|
||||
@@ -1,154 +0,0 @@
|
||||
package rabbitmq
|
||||
|
||||
import (
|
||||
"agent-go/g"
|
||||
"fmt"
|
||||
"github.com/streadway/amqp"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
var log = g.G.LOG
|
||||
|
||||
// 定义全局唯一的 Singleton 实例
|
||||
var instance *amqp.Connection
|
||||
|
||||
// 用 sync.Once 变量确保初始化函数只会被调用一次
|
||||
var once sync.Once
|
||||
|
||||
// 初始化 Singleton 实例的函数
|
||||
func createInstance() {
|
||||
// 在这里进行 Singleton 的初始化操作
|
||||
|
||||
// 获取RabbitMQ的连接地址
|
||||
rabbitMQEndpointFromG := parseRabbitMQEndpointFromG()
|
||||
|
||||
// 创建全局唯一连接 RabbitMQ连接
|
||||
connection, err := amqp.Dial(rabbitMQEndpointFromG)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("failed to connect to RabbitMQ: %v", err))
|
||||
}
|
||||
|
||||
instance = connection
|
||||
}
|
||||
|
||||
// GetInstance 获取全局唯一的 Singleton 实例的函数
|
||||
func GetInstance() *amqp.Connection {
|
||||
// 使用 sync.Once 确保 createInstance 只会被调用一次
|
||||
once.Do(createInstance)
|
||||
return instance
|
||||
}
|
||||
|
||||
// NewRabbitMQConn creates a new RabbitMQ connection object
|
||||
func NewRabbitMQConn(property *ConnectProperty) (*RabbitMQConn, error) {
|
||||
|
||||
// 获取RabbitMQ的连接
|
||||
conn := GetInstance()
|
||||
// 获取RabbitMQ的连接地址
|
||||
//rabbitMQEndpointFromG := parseRabbitMQEndpointFromG()
|
||||
//conn, err := amqp.Dial(rabbitMQEndpointFromG)
|
||||
//if err != nil {
|
||||
// log.Error(fmt.Sprintf("failed to connect to RabbitMQ: %v", err))
|
||||
//}
|
||||
|
||||
ch, err := conn.Channel()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create RabbitMQ channel: %w", err)
|
||||
}
|
||||
|
||||
if err = ch.ExchangeDeclare(
|
||||
property.ExchangeName, // name of the exchange
|
||||
property.ExchangeType, // type of the exchange
|
||||
false, // durable
|
||||
false, // delete when complete
|
||||
false, // internal
|
||||
false, // noWait
|
||||
nil, // arguments
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("failed to declare RabbitMQ exchange: %w", err)
|
||||
}
|
||||
|
||||
_, err = ch.QueueDeclare(
|
||||
property.QueueName, // name of the queue
|
||||
false, // durable
|
||||
false, // delete when unused
|
||||
false, // exclusive
|
||||
false, // noWait
|
||||
nil, // arguments
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to declare RabbitMQ queue: %w", err)
|
||||
}
|
||||
|
||||
if err = ch.QueueBind(
|
||||
property.QueueName, // name of the queue
|
||||
property.TopicKey, // routing key - all topics
|
||||
property.ExchangeName, // name of the exchange
|
||||
false, // noWait
|
||||
nil, // arguments
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("failed to bind RabbitMQ queue: %w", err)
|
||||
}
|
||||
|
||||
return &RabbitMQConn{Connection: conn, Channel: ch}, nil
|
||||
}
|
||||
|
||||
// parseRabbitMQEndpoint 根据全局变量NacosConfig解析出RabbitMQ的连接地址
|
||||
func parseRabbitMQEndpointFromG() string {
|
||||
|
||||
nacosConfig := g.G.NacosConfig
|
||||
|
||||
var res strings.Builder
|
||||
|
||||
host := nacosConfig.GetString("spring.rabbitmq.host")
|
||||
port := nacosConfig.GetString("spring.rabbitmq.port")
|
||||
username := nacosConfig.GetString("spring.rabbitmq.username")
|
||||
password := nacosConfig.GetString("spring.rabbitmq.password")
|
||||
virtualHost := nacosConfig.GetString("spring.rabbitmq.virtual-host")
|
||||
|
||||
// amqp://{username}:{password}@{hostname}:{port}/{virtual_host}
|
||||
res.WriteString("amqp://")
|
||||
res.WriteString(username)
|
||||
res.WriteString(":")
|
||||
res.WriteString(password)
|
||||
res.WriteString("@")
|
||||
res.WriteString(host)
|
||||
res.WriteString(":")
|
||||
res.WriteString(port)
|
||||
res.WriteString("/")
|
||||
res.WriteString(virtualHost)
|
||||
|
||||
s := res.String()
|
||||
log.Debug(fmt.Sprintf("generate RabbitMQ endpoint is %s", s))
|
||||
return s
|
||||
}
|
||||
|
||||
func CloseChannel(conn *RabbitMQConn) error {
|
||||
var err error
|
||||
|
||||
if conn.Channel != nil {
|
||||
if err = conn.Channel.Close(); err != nil {
|
||||
log.Error(fmt.Sprintf("Failed to close RabbitMQ channel: %v", err))
|
||||
}
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// CloseRabbitMQAll closes the RabbitMQ connection and channel
|
||||
func (r *RabbitMQConn) CloseRabbitMQAll() error {
|
||||
var err error
|
||||
|
||||
if r.Channel != nil {
|
||||
if err = r.Channel.Close(); err != nil {
|
||||
log.Error(fmt.Sprintf("Failed to close RabbitMQ channel: %v", err))
|
||||
}
|
||||
}
|
||||
|
||||
if r.Connection != nil {
|
||||
if err = r.Connection.Close(); err != nil {
|
||||
log.Error(fmt.Sprintf("Failed to close RabbitMQ connection: %v", err))
|
||||
}
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
214
agent-go/rabbitmq/RabbitMsgQueue.go
Normal file
214
agent-go/rabbitmq/RabbitMsgQueue.go
Normal file
@@ -0,0 +1,214 @@
|
||||
package rabbitmq
|
||||
|
||||
import (
|
||||
"agent-go/g"
|
||||
logger2 "agent-go/logger"
|
||||
"fmt"
|
||||
"github.com/streadway/amqp"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type RabbitMQ interface {
|
||||
RabbitSendWriter
|
||||
|
||||
RabbitConnectCloser
|
||||
}
|
||||
|
||||
type RabbitSendWriter interface {
|
||||
Send(message []byte)
|
||||
|
||||
Read(autoAck bool) <-chan amqp.Delivery
|
||||
}
|
||||
|
||||
type RabbitConnectCloser interface {
|
||||
Connect()
|
||||
|
||||
Close() error
|
||||
}
|
||||
|
||||
type RabbitQueue struct {
|
||||
RabbitConn *RabbitMQConn
|
||||
RabbitProp *ConnectProperty
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
var log = logger2.Log
|
||||
|
||||
// 定义全局唯一的 Singleton 实例
|
||||
var instance *amqp.Connection
|
||||
|
||||
// 用 sync.Once 变量确保初始化函数只会被调用一次
|
||||
var once sync.Once
|
||||
|
||||
// 初始化 Singleton 实例的函数
|
||||
func createInstance() {
|
||||
// 在这里进行 Singleton 的初始化操作
|
||||
|
||||
// 获取RabbitMQ的连接地址
|
||||
rabbitMQEndpointFromG := parseRabbitMQEndpointFromG()
|
||||
|
||||
// 创建全局唯一连接 RabbitMQ连接
|
||||
connection, err := amqp.Dial(rabbitMQEndpointFromG)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("failed to connect to RabbitMQ: %v", err))
|
||||
}
|
||||
|
||||
instance = connection
|
||||
}
|
||||
|
||||
// GetInstance 获取全局唯一的 Singleton 实例的函数
|
||||
func GetInstance() *amqp.Connection {
|
||||
// 使用 sync.Once 确保 createInstance 只会被调用一次
|
||||
once.Do(createInstance)
|
||||
return instance
|
||||
}
|
||||
|
||||
// Connect creates a new RabbitMQ connection object
|
||||
func (r *RabbitQueue) Connect() {
|
||||
|
||||
// 获取RabbitMQ的连接
|
||||
conn := GetInstance()
|
||||
|
||||
ch, err := conn.Channel()
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("failed to create RabbitMQ channel: %w", err))
|
||||
}
|
||||
|
||||
if err = ch.ExchangeDeclare(
|
||||
r.RabbitProp.ExchangeName, // name of the exchange
|
||||
r.RabbitProp.ExchangeType, // type of the exchange
|
||||
false, // durable
|
||||
false, // delete when complete
|
||||
false, // internal
|
||||
false, // noWait
|
||||
nil, // arguments
|
||||
); err != nil {
|
||||
log.Error(fmt.Sprintf("failed to declare exchange !: %w", err))
|
||||
}
|
||||
|
||||
_, err = ch.QueueDeclare(
|
||||
r.RabbitProp.QueueName, // name of the queue
|
||||
false, // durable
|
||||
false, // delete when unused
|
||||
false, // exclusive
|
||||
false, // noWait
|
||||
nil, // arguments
|
||||
)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("failed to declare RabbitMQ queue: %w", err))
|
||||
}
|
||||
|
||||
if err = ch.QueueBind(
|
||||
r.RabbitProp.QueueName, // name of the queue
|
||||
r.RabbitProp.TopicKey, // routing key - all topics
|
||||
r.RabbitProp.ExchangeName, // name of the exchange
|
||||
false, // noWait
|
||||
nil, // arguments
|
||||
); err != nil {
|
||||
log.Error(fmt.Sprintf("failed to bind RabbitMQ queue: %w", err))
|
||||
}
|
||||
|
||||
r.RabbitConn = &RabbitMQConn{
|
||||
Connection: conn,
|
||||
Channel: ch,
|
||||
}
|
||||
}
|
||||
|
||||
func (r *RabbitQueue) Close() error {
|
||||
var err error
|
||||
|
||||
if r.RabbitConn.Channel != nil {
|
||||
if err = r.RabbitConn.Channel.Close(); err != nil {
|
||||
log.Error(fmt.Sprintf("Failed to close RabbitMQ channel: %v", err))
|
||||
}
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// Send 向RabbitMQ中发送消息
|
||||
func (r *RabbitQueue) Send(message []byte) {
|
||||
// 往哪里发
|
||||
channel := r.RabbitConn.Channel
|
||||
|
||||
// 发送
|
||||
err := channel.Publish(
|
||||
r.RabbitProp.ExchangeName,
|
||||
r.RabbitProp.TopicKey,
|
||||
false,
|
||||
false,
|
||||
amqp.Publishing{
|
||||
ContentType: "text/plain",
|
||||
Body: message,
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
log.Error(fmt.Sprintf("Failed to publish a message: %v", err))
|
||||
}
|
||||
}
|
||||
|
||||
func (r *RabbitQueue) Read(autoAck bool) <-chan amqp.Delivery {
|
||||
|
||||
// 拿到特定的Channel
|
||||
channel := r.RabbitConn.Channel
|
||||
|
||||
// 开始读取队列中的全部消息
|
||||
msgs, err := channel.Consume(
|
||||
r.RabbitProp.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
|
||||
}
|
||||
|
||||
// parseRabbitMQEndpoint 根据全局变量NacosConfig解析出RabbitMQ的连接地址
|
||||
func parseRabbitMQEndpointFromG() string {
|
||||
|
||||
nacosConfig := g.G.NacosConfig
|
||||
|
||||
var res strings.Builder
|
||||
|
||||
host := nacosConfig.GetString("spring.rabbitmq.host")
|
||||
port := nacosConfig.GetString("spring.rabbitmq.port")
|
||||
username := nacosConfig.GetString("spring.rabbitmq.username")
|
||||
password := nacosConfig.GetString("spring.rabbitmq.password")
|
||||
virtualHost := nacosConfig.GetString("spring.rabbitmq.virtual-host")
|
||||
|
||||
// amqp://{username}:{password}@{hostname}:{port}/{virtual_host}
|
||||
res.WriteString("amqp://")
|
||||
res.WriteString(username)
|
||||
res.WriteString(":")
|
||||
res.WriteString(password)
|
||||
res.WriteString("@")
|
||||
res.WriteString(host)
|
||||
res.WriteString(":")
|
||||
res.WriteString(port)
|
||||
res.WriteString("/")
|
||||
res.WriteString(virtualHost)
|
||||
|
||||
s := res.String()
|
||||
|
||||
log.Debug(fmt.Sprintf("generate RabbitMQ endpoint is %s", s))
|
||||
|
||||
return s
|
||||
}
|
||||
Reference in New Issue
Block a user