拆开配置文件

factory/rabbitmq/consumer_factory.go

@@ -0,0 +1,80 @@
+package rabbitmq
+
+import (
+	"fmt"
+
+	"git.x2erp.com/qdy/go-base/types"
+	amqp "github.com/streadway/amqp"
+)
+
+// MessageHandler 消息处理器接口
+type MessageHandler interface {
+	Process(ueueRequest *types.QueueRequest, body []byte) types.QueryResult
+}
+
+// Consumer 消费者结构体
+type Consumer struct {
+	Channel      *amqp.Channel
+	QueueName    string
+	Handler      MessageHandler
+	QueueRequest *types.QueueRequest
+}
+
+// CreateConsumer 创建消费者
+// 返回Consumer对象，创建后自动开始消费
+func (f *RabbitMQFactory) CreateConsumer(queueRequest *types.QueueRequest, consumerID string, handler MessageHandler) (*Consumer, error) {
+	// 生成唯一的通道名称
+	channelName := fmt.Sprintf("%s_%s", queueRequest.QueueName, consumerID)
+
+	// 创建新通道
+	channel, err := f.CreateChannel(channelName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create channel: %v", err)
+	}
+
+	// 设置QoS（一次处理一条）
+	err = channel.Qos(1, 0, false)
+	if err != nil {
+		f.CloseChannel(channelName) // 清理
+		return nil, fmt.Errorf("failed to set qos: %v", err)
+	}
+
+	// 开始消费
+	deliveries, err := channel.Consume(
+		queueRequest.QueueName,
+		channelName, // 使用通道名称作为消费者标签
+		false,       // 手动确认
+		false,       // 非独占
+		false,       // 不排除本地连接
+		false,       // 阻塞等待
+		nil,         // 额外参数
+	)
+
+	if err != nil {
+		f.CloseChannel(channelName) // 清理
+		return nil, fmt.Errorf("failed to start consumer: %v", err)
+	}
+
+	// 创建Consumer对象
+	consumer := &Consumer{
+		Channel:      channel,
+		QueueName:    queueRequest.QueueName,
+		Handler:      handler,
+		QueueRequest: queueRequest,
+	}
+
+	// 启动goroutine处理消息
+	go func() {
+		for delivery := range deliveries {
+			queryResult := handler.Process(queueRequest, delivery.Body)
+			if queryResult.Success {
+				delivery.Ack(false)
+			} else {
+				delivery.Reject(true)
+			}
+		}
+		// deliveries通道关闭时，goroutine自动结束
+	}()
+
+	return consumer, nil
+}

factory/rabbitmq/rabbitmq_factory.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"git.x2erp.com/qdy/go-base/config"
+	"git.x2erp.com/qdy/go-base/config/subconfigs"
 	amqp "github.com/streadway/amqp"
 )
 
@@ -14,7 +15,7 @@ type RabbitMQFactory struct {
 	mu      sync.RWMutex
 	clients map[string]*amqp.Channel
 	conn    *amqp.Connection
-	config  config.RabbitMQConfig
+	config  *subconfigs.RabbitMQConfig
 }
 
 // NewRabbitMQFactory 创建RabbitMQ工厂
@@ -38,8 +39,14 @@ func NewRabbitMQFactory() (*RabbitMQFactory, error) {
 		rabbitConfig.Port,
 		rabbitConfig.Vhost)
 
+	// 简单的连接配置
+	mqConfig := amqp.Config{
+		Properties: amqp.Table{
+			"connection_name": cfg.GetService().InstanceName,
+		}}
+
 	// 建立连接
-	conn, err := amqp.Dial(url)
+	conn, err := amqp.DialConfig(url, mqConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to rabbitmq: %v", err)
 	}

gct.sh

@@ -1,82 +1,25 @@
 #!/bin/bash
 
-# 脚本用法：./git-commit-and-tag.sh "你的提交描述" "版本号"
+# 使用命令行参数，如果没有提供则使用默认值
+COMMIT_MSG="${1:-初始提交}"
+TAG_VERSION="${2:-v0.1.0}"
 
-# 检查参数数量
-if [ $# -ne 2 ]; then
-    echo "错误: 脚本需要2个参数。"
-    echo "用法: $0 \"提交描述\" \"版本号\""
-    echo "示例: $0 \"修复了登录问题\" \"v1.2.3\""
-    exit 1
-fi
+echo "提交信息: $COMMIT_MSG"
+echo "标签版本: $TAG_VERSION"
 
-# 分配参数
-COMMIT_MESSAGE="$1"
-VERSION_TAG="$2"
+# 提交和推送
+git add .
+git commit -m "$COMMIT_MSG"
+git push -u origin master
 
-# 检查当前目录是否为Git仓库
-if ! git rev-parse --git-dir > /dev/null 2>&1; then
-    echo "错误: 当前目录不是一个Git仓库。"
-    exit 1
-fi
+# 创建标签（关键修改在这里）
+# 方法1：使用注释标签（推荐，不会打开编辑器）
+git tag -a "$TAG_VERSION" -m "Release $TAG_VERSION"
 
-echo "开始处理提交和版本标签..."
-echo "提交描述: $COMMIT_MESSAGE"
-echo "版本标签: $VERSION_TAG"
+# 或者方法2：如果你需要GPG签名，使用这个
+# GIT_EDITOR=true git tag -s "$TAG_VERSION" -m "Release $TAG_VERSION"
 
-# 检查是否有未提交的更改
-if [ -n "$(git status --porcelain)" ]; then
-    echo "检测到未提交的更改，正在提交..."
-    
-    # 添加所有更改到暂存区
-    git add .
-    
-    # 进行提交
-    git commit -m "$COMMIT_MESSAGE"
-    if [ $? -ne 0 ]; then
-        echo "错误: 提交失败。"
-        exit 1
-    fi
-    echo "✅ 更改已提交"
-else
-    echo "提示: 没有未提交的更改，跳过提交步骤"
-    
-    # 检查是否有未提交的commit但未推送
-    LOCAL_COMMITS=$(git log @{u}..HEAD --oneline 2>/dev/null | wc -l)
-    if [ $LOCAL_COMMITS -eq 0 ]; then
-        echo "错误: 没有需要推送的提交。"
-        exit 1
-    else
-        echo "检测到 $LOCAL_COMMITS 个本地提交等待推送"
-    fi
-fi
+# 推送标签
+git push origin "$TAG_VERSION"
 
-# 检查标签是否已存在
-#if git rev-parse "$VERSION_TAG" >/dev/null 2>&1; then
- #   echo "错误: 标签 '$VERSION_TAG' 已经存在。"
-  #  exit 1
-#fi
-
-# 创建标签
-#git tag "$VERSION_TAG"
-#if [ $? -ne 0 ]; then
-#    echo "错误: 创建标签失败。"
-#    exit 1
-#fi
-#echo "✅ 标签 '$VERSION_TAG' 已创建"
-
-# 推送到远程仓库并推送标签
-echo "正在推送到远程仓库..."
-git push
-if [ $? -ne 0 ]; then
-    echo "错误: 推送提交失败。"
-    exit 1
-fi
-
-#git push origin "$VERSION_TAG"
-#if [ $? -ne 0 ]; then
-#    echo "错误: 推送标签失败。"
-#    exit 1
-#fi
-
-#echo "✅ 完成！提交已推送，版本标签 $VERSION_TAG 已创建并推送。"
+echo "完成！"