first commit

apps/mqs/Dockerfile

@@ -0,0 +1,31 @@
+FROM golang:alpine AS builder
+
+LABEL stage=gobuilder
+
+ENV CGO_ENABLED 0
+ENV GOPROXY https://goproxy.cn,direct
+RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories
+
+RUN apk update --no-cache && apk add --no-cache tzdata
+
+WORKDIR /build
+
+ADD go.mod .
+ADD go.sum .
+RUN go mod download
+COPY . .
+COPY apps/mqs/etc /app/etc
+RUN go build -ldflags="-s -w" -o /app/main apps/mqs/.\main.go
+
+
+FROM scratch
+
+COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
+COPY --from=builder /usr/share/zoneinfo/Asia/Shanghai /usr/share/zoneinfo/Asia/Shanghai
+ENV TZ Asia/Shanghai
+
+WORKDIR /app
+COPY --from=builder /app/main /app/main
+COPY --from=builder /app/etc /app/etc
+
+CMD ["./main", "-f", "etc/mqs.yaml"]

apps/mqs/etc/mqs.yaml

@@ -0,0 +1,25 @@
+Name: mq
+Host: 0.0.0.0
+Port: 12001
+
+# kq 消费者组1 - 用于记录
+KqConsumerLog:
+  Name: kqConsumerLog
+  Brokers:
+    - 127.0.0.1:9092
+  Group: logGroup
+  Topic: apirequest
+  Offset: first
+  Consumers: 2  # 为了避免资源争夺，分配2个消费者
+  Processors: 2
+
+# kq 消费者组2 - 用于扣款
+KqConsumerCharge:
+  Name: kqConsumerCharge
+  Brokers:
+    - 127.0.0.1:9092
+  Group: chargeGroup
+  Topic: apirequest
+  Offset: first
+  Consumers: 2  # 同样分配2个消费者
+  Processors: 2

apps/mqs/internal/config/config.go

@@ -0,0 +1,12 @@
+package config
+
+import (
+	"github.com/zeromicro/go-queue/kq"
+	"github.com/zeromicro/go-zero/rest"
+)
+
+type Config struct {
+	rest.RestConf
+	KqConsumerLog    kq.KqConf
+	KqConsumerCharge kq.KqConf
+}

apps/mqs/internal/mqs/apirequest/charge.go

@@ -0,0 +1,24 @@
+package apirequest
+
+import (
+	"context"
+	"github.com/zeromicro/go-zero/core/logx"
+	"tianyuan-api/apps/mqs/internal/svc"
+)
+
+type Charge struct {
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+func NewCharge(ctx context.Context, svcCtx *svc.ServiceContext) *Charge {
+	return &Charge{
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *Charge) Consume(ctx context.Context, key, val string) error {
+	logx.Infof("Charge key :%s , val :%s", key, val)
+	return nil
+}

apps/mqs/internal/mqs/apirequest/log.go

@@ -0,0 +1,24 @@
+package apirequest
+
+import (
+	"context"
+	"github.com/zeromicro/go-zero/core/logx"
+	"tianyuan-api/apps/mqs/internal/svc"
+)
+
+type Log struct {
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+func NewLog(ctx context.Context, svcCtx *svc.ServiceContext) *Log {
+	return &Log{
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *Log) Consume(ctx context.Context, key, val string) error {
+	logx.Infof("log key :%s , val :%s", key, val)
+	return nil
+}

apps/mqs/internal/mqs/mqs.go

@@ -0,0 +1,18 @@
+package mqs
+
+import (
+	"context"
+	"tianyuan-api/apps/mqs/internal/config"
+	"tianyuan-api/apps/mqs/internal/mqs/apirequest"
+	"tianyuan-api/apps/mqs/internal/svc"
+
+	"github.com/zeromicro/go-queue/kq"
+	"github.com/zeromicro/go-zero/core/service"
+)
+
+func Consumers(c config.Config, ctx context.Context, svcContext *svc.ServiceContext) []service.Service {
+	return []service.Service{
+		kq.MustNewQueue(c.KqConsumerLog, apirequest.NewLog(ctx, svcContext)),
+		kq.MustNewQueue(c.KqConsumerCharge, apirequest.NewCharge(ctx, svcContext)),
+	}
+}

apps/mqs/internal/svc/servicecontext.go

@@ -0,0 +1,13 @@
+package svc
+
+import "tianyuan-api/apps/mqs/internal/config"
+
+type ServiceContext struct {
+	Config config.Config
+}
+
+func NewServiceContext(c config.Config) *ServiceContext {
+	return &ServiceContext{
+		Config: c,
+	}
+}

apps/mqs/main.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"github.com/zeromicro/go-zero/core/service"
+	"tianyuan-api/apps/mqs/internal/config"
+	"tianyuan-api/apps/mqs/internal/mqs"
+	"tianyuan-api/apps/mqs/internal/svc"
+
+	"github.com/zeromicro/go-zero/core/conf"
+	"github.com/zeromicro/go-zero/rest"
+)
+
+var configFile = flag.String("f", "etc/mqs.yaml", "the config file")
+
+func main() {
+	flag.Parse()
+
+	var c config.Config
+	conf.MustLoad(*configFile, &c)
+
+	server := rest.MustNewServer(c.RestConf)
+	defer server.Stop()
+
+	svcCtx := svc.NewServiceContext(c)
+	ctx := context.Background()
+	serviceGroup := service.NewServiceGroup()
+	defer serviceGroup.Stop()
+
+	for _, mq := range mqs.Consumers(c, ctx, svcCtx) {
+		serviceGroup.Add(mq)
+	}
+
+	serviceGroup.Start()
+}