add rabbitmq deploy files

deploy/rabbitmq-config.yaml

@@ -0,0 +1,30 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: rabbitmq-config
+data:
+  rabbitmq.conf: |
+    # 确保允许PLAIN认证
+    auth_mechanisms.1 = PLAIN
+    auth_mechanisms.2 = AMQPLAIN
+    # 允许admin用户通过远程方式连接
+    loopback_users.admin = false
+    # 默认心跳和监听配置可在此扩展
+    # ssl 新增部分
+    # 确定 ssl 连接时验证使用的用户名
+    ssl_cert_login_from = common_name
+    # 开启此项配置会导致只能通过TLS端口访问
+    listeners.tcp = none
+    listeners.ssl.default = 5671
+    # ssl config
+    ssl_options.cacertfile = /etc/rabbitmq/certs/ca_certificate.pem
+    ssl_options.certfile   = /etc/rabbitmq/certs/server_certificate.pem
+    ssl_options.keyfile    = /etc/rabbitmq/certs/server_key.pem
+    # management ssl config
+    management.ssl.port       = 15671
+    management.ssl.cacertfile = /etc/rabbitmq/certs/ca_certificate.pem
+    management.ssl.certfile   = /etc/rabbitmq/certs/server_certificate.pem
+    management.ssl.keyfile    = /etc/rabbitmq/certs/server_key.pem
+    # 启用双向认证
+    ssl_options.verify     = verify_peer
+    ssl_options.fail_if_no_peer_cert = true

deploy/rabbitmq-deployment.yaml

@@ -0,0 +1,63 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: eventrt-rabbitmq
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: rabbitmq
+  template:
+    metadata:
+      labels:
+        app: rabbitmq
+    spec:
+      containers:
+        - name: rabbitmq
+          image: rabbitmq:4.1.1-management-alpine
+          ports:
+            - containerPort: 4369
+            - containerPort: 5671
+            - containerPort: 5672 # AMQP
+            - containerPort: 15671
+            - containerPort: 15672 # Management UI
+            - containerPort: 15691
+            - containerPort: 15692
+            - containerPort: 25672
+          env:
+            - name: RABBITMQ_DEFAULT_USER
+              valueFrom:
+                secretKeyRef:
+                  name: rabbitmq-secret
+                  key: rabbitmq-user
+            - name: RABBITMQ_DEFAULT_PASS
+              valueFrom:
+                secretKeyRef:
+                  name: rabbitmq-secret
+                  key: rabbitmq-pass
+            - name: RABBITMQ_ERLANG_COOKIE
+              valueFrom:
+                secretKeyRef:
+                  name: rabbitmq-secret
+                  key: erlang-cookie
+            - name: RABBITMQ_DEFAULT_VHOST
+              value: "/"
+          volumeMounts:
+            - name: rabbitmq-certs-volume
+              mountPath: /etc/rabbitmq/certs
+              readOnly: true
+            - name: rabbitmq-config-volume
+              mountPath: /etc/rabbitmq/rabbitmq.conf
+              subPath: rabbitmq.conf
+              readOnly: true
+            - name: rabbitmq-data
+              mountPath: /var/lib/rabbitmq
+      volumes:
+        - name: rabbitmq-certs-volume
+          secret:
+            secretName: rabbitmq-certs
+        - name: rabbitmq-config-volume
+          configMap:
+            name: rabbitmq-config
+        - name: rabbitmq-data
+          emptyDir: {}

deploy/rabbitmq-secret.yaml

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: rabbitmq-secret
+type: Opaque
+stringData:
+  rabbitmq-user: "coslight"
+  rabbitmq-pass: "coslight@tj"
+  erlang-cookie: "secret-erlang-cookie"

deploy/rabbitmq-service.yaml

@@ -0,0 +1,29 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: rabbitmq-service
+spec:
+  type: NodePort # 在 Minikube 中使用 NodePort 方便外部访问
+  selector:
+    app: rabbitmq
+  ports:
+    - name: amqp-ssl
+      protocol: TCP
+      port: 5671
+      targetPort: 5671
+      nodePort: 30671
+    - name: amqp
+      protocol: TCP
+      port: 5672
+      targetPort: 5672
+      nodePort: 30672
+    - name: management-ssl
+      protocol: TCP
+      port: 15671
+      targetPort: 15671
+      nodePort: 31671
+    - name: management
+      protocol: TCP
+      port: 15672
+      targetPort: 15672
+      nodePort: 31672

mq/mq_init.go

@@ -57,7 +57,7 @@ func initRabbitMQ(ctx context.Context, rabbitMQURI string) *amqp.Connection {
 func (p *RabbitMQProxy) handleReconnect(ctx context.Context, rabbitMQURI string) {
 	for {
 		closeChan := make(chan *amqp.Error)
-		p.Conn.NotifyClose(closeChan)
+		GetConn().NotifyClose(closeChan)
 
 		err, ok := <-closeChan