pro/im_manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add im kafka admin integration

Browse Source
This commit is contained in:
richjohnrambo
2026-09-04 11:57:38 +08:00
parent a2abaef3a6
commit f7f8cc0ef0
74 changed files with 6254 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/imadmin/config.go
+26 -8
View File
@@ -10,10 +10,13 @@ import (
)
type Config struct {
BaseUrl string `json:"baseUrl"`
ServiceId string `json:"serviceId"`
ServiceSecret string `json:"serviceSecret"`
Timeout time.Duration `json:"timeout"`
BaseUrl string `json:"baseUrl"`
ServiceId string `json:"serviceId"`
ServiceSecret string `json:"serviceSecret"`
Timeout time.Duration `json:"timeout"`
InternalBaseUrl string `json:"internalBaseUrl"`
InternalServiceName string `json:"internalServiceName"`
InternalServiceSecret string `json:"internalServiceSecret"`
}
func LoadConfig(ctx context.Context) Config {
@@ -28,10 +31,25 @@ func LoadConfig(ctx context.Context) Config {
if serviceSecret == "" {
serviceSecret = os.Getenv("IM_ADMIN_API_SECRET")
}
internalSecret := g.Cfg().MustGet(ctx, "imInternalApi.serviceSecret").String()
if strings.HasPrefix(internalSecret, "${") && strings.HasSuffix(internalSecret, "}") {
internalSecret = os.Getenv(strings.TrimSuffix(strings.TrimPrefix(internalSecret, "${"), "}"))
}
if internalSecret == "" {
internalSecret = os.Getenv("IM_INTERNAL_API_SECRET")
}
if internalSecret == "" {
internalSecret = serviceSecret
}
baseUrl := g.Cfg().MustGet(ctx, "imAdminApi.baseUrl").String()
serviceId := g.Cfg().MustGet(ctx, "imAdminApi.serviceId", "hotgo-admin").String()
return Config{
BaseUrl: g.Cfg().MustGet(ctx, "imAdminApi.baseUrl").String(),
ServiceId: g.Cfg().MustGet(ctx, "imAdminApi.serviceId", "hotgo-admin").String(),
ServiceSecret: serviceSecret,
Timeout: timeout,
BaseUrl: baseUrl,
ServiceId: serviceId,
ServiceSecret: serviceSecret,
Timeout: timeout,
InternalBaseUrl: g.Cfg().MustGet(ctx, "imInternalApi.baseUrl", baseUrl).String(),
InternalServiceName: g.Cfg().MustGet(ctx, "imInternalApi.serviceName", serviceId).String(),
InternalServiceSecret: internalSecret,
}
}
server/internal/library/imadmin/internal.go
+101
View File
@@ -0,0 +1,101 @@
package imadmin
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"io"
"net/http"
"strconv"
"strings"
"time"
"github.com/gogf/gf/v2/errors/gerror"
)
func (c *Client) InternalGet(ctx context.Context, path string, out any) error {
req, err := c.newInternalRequest(ctx, http.MethodGet, path, nil)
if err != nil {
return err
}
resp, err := c.http.Do(req)
if err != nil {
return gerror.New("IM内部查询服务暂不可用")
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return c.parseError(resp)
}
if out == nil {
return nil
}
body, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBody))
if err != nil {
return gerror.Wrap(err, "读取IM内部查询响应失败")
}
var envelope struct {
Code int `json:"code"`
Message string `json:"message"`
Reason string `json:"reason"`
Data json.RawMessage `json:"data"`
}
if err = json.Unmarshal(body, &envelope); err == nil && len(envelope.Data) > 0 {
if envelope.Code != 0 {
return stableError(envelope.Message, envelope.Reason)
}
return json.Unmarshal(envelope.Data, out)
}
return json.Unmarshal(body, out)
}
func (c *Client) newInternalRequest(ctx context.Context, method string, path string, bodyBytes []byte) (*http.Request, error) {
baseUrl := strings.TrimRight(c.conf.InternalBaseUrl, "/")
if baseUrl == "" {
baseUrl = strings.TrimRight(c.conf.BaseUrl, "/")
}
if baseUrl == "" {
return nil, gerror.New("IM内部查询API未配置")
}
serviceName := strings.TrimSpace(c.conf.InternalServiceName)
if serviceName == "" {
serviceName = strings.TrimSpace(c.conf.ServiceId)
}
if serviceName == "" {
return nil, gerror.New("IM内部查询服务名未配置")
}
secret := c.conf.InternalServiceSecret
if secret == "" {
secret = c.conf.ServiceSecret
}
if secret == "" {
return nil, gerror.New("IM内部查询API密钥未配置")
}
req, err := http.NewRequestWithContext(ctx, method, baseUrl+path, bytes.NewReader(bodyBytes))
if err != nil {
return nil, gerror.Wrap(err, "创建IM内部查询请求失败")
}
timestamp := strconv.FormatInt(time.Now().UTC().Unix(), 10)
nonce, err := randomNonce()
if err != nil {
return nil, err
}
bodyHash := sha256.Sum256(bodyBytes)
signingText := strings.Join([]string{
method,
requestTarget(req),
hex.EncodeToString(bodyHash[:]),
timestamp,
nonce,
}, "\n")
mac := hmac.New(sha256.New, []byte(secret))
_, _ = mac.Write([]byte(signingText))
req.Header.Set("Accept", "application/json")
req.Header.Set("X-Service-Name", serviceName)
req.Header.Set("X-Timestamp", timestamp)
req.Header.Set("X-Nonce", nonce)
req.Header.Set("X-Signature", hex.EncodeToString(mac.Sum(nil)))
return req, nil
}
server/internal/library/iminternal/auth.go
+135
View File
@@ -0,0 +1,135 @@
package iminternal
import (
"bytes"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"io"
"net/http"
"os"
"strconv"
"strings"
"sync"
"time"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/ghttp"
)
const (
headerService = "X-Service-Name"
headerTimestamp = "X-Timestamp"
headerNonce = "X-Nonce"
headerSignature = "X-Signature"
)
var nonces sync.Map
func Middleware(r *ghttp.Request) {
if !Authorize(r) {
return
}
r.Middleware.Next()
}
func Authorize(r *ghttp.Request) bool {
ctx := r.GetCtx()
serviceName := strings.TrimSpace(r.Header.Get(headerService))
timestampRaw := strings.TrimSpace(r.Header.Get(headerTimestamp))
nonce := strings.TrimSpace(r.Header.Get(headerNonce))
signature := strings.TrimSpace(r.Header.Get(headerSignature))
expectedService := g.Cfg().MustGet(ctx, "imConfigApi.serviceName", "im-server").String()
secret := configuredSecret(r)
if serviceName == "" || timestampRaw == "" || nonce == "" || signature == "" || serviceName != expectedService || secret == "" {
writeAuthError(r, http.StatusUnauthorized, "internal signature required")
return false
}
timestamp, err := strconv.ParseInt(timestampRaw, 10, 64)
if err != nil {
writeAuthError(r, http.StatusUnauthorized, "invalid internal timestamp")
return false
}
window := g.Cfg().MustGet(ctx, "imConfigApi.timeWindow", "300s").Duration()
if window <= 0 {
window = 5 * time.Minute
}
delta := time.Since(time.Unix(timestamp, 0))
if delta > window || delta < -window {
writeAuthError(r, http.StatusUnauthorized, "internal signature expired")
return false
}
body, err := readBody(r)
if err != nil {
writeAuthError(r, http.StatusInternalServerError, "read internal body failed")
return false
}
bodyHash := sha256.Sum256(body)
signingText := strings.Join([]string{
r.Method,
r.URL.RequestURI(),
hex.EncodeToString(bodyHash[:]),
timestampRaw,
nonce,
}, "\n")
mac := hmac.New(sha256.New, []byte(secret))
_, _ = mac.Write([]byte(signingText))
if !hmac.Equal([]byte(strings.ToLower(signature)), []byte(hex.EncodeToString(mac.Sum(nil)))) {
writeAuthError(r, http.StatusUnauthorized, "invalid internal signature")
return false
}
if !claimNonce(serviceName, nonce, time.Now().Add(window)) {
writeAuthError(r, http.StatusUnauthorized, "internal nonce replayed")
return false
}
return true
}
func configuredSecret(r *ghttp.Request) string {
ctx := r.GetCtx()
secret := g.Cfg().MustGet(ctx, "imConfigApi.serviceSecret").String()
if strings.HasPrefix(secret, "${") && strings.HasSuffix(secret, "}") {
secret = os.Getenv(strings.TrimSuffix(strings.TrimPrefix(secret, "${"), "}"))
}
if secret == "" {
secret = os.Getenv("IM_CONFIG_API_SECRET")
}
if secret == "" {
secret = g.Cfg().MustGet(ctx, "imInternalApi.serviceSecret").String()
}
return secret
}
func readBody(r *ghttp.Request) ([]byte, error) {
if r.Body == nil {
return nil, nil
}
body, err := io.ReadAll(io.LimitReader(r.Body, 4<<20))
if err != nil {
return nil, err
}
r.Body = io.NopCloser(bytes.NewReader(body))
return body, nil
}
func claimNonce(serviceName string, nonce string, expiresAt time.Time) bool {
now := time.Now()
nonces.Range(func(key, value any) bool {
if expires, ok := value.(time.Time); ok && expires.Before(now) {
nonces.Delete(key)
}
return true
})
key := serviceName + ":" + nonce
_, loaded := nonces.LoadOrStore(key, expiresAt)
return !loaded
}
func writeAuthError(r *ghttp.Request, status int, message string) {
r.Response.WriteStatus(status)
r.Response.WriteJson(g.Map{
"code": status,
"message": message,
})
r.Exit()
}
server/internal/library/imkafka/command.go
+378
View File
@@ -0,0 +1,378 @@
package imkafka
import (
"context"
"encoding/json"
"hotgo/internal/consts"
"hotgo/internal/library/queue"
"time"
"github.com/gogf/gf/v2/database/gdb"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/gctx"
"github.com/gogf/gf/v2/os/gtime"
"github.com/gogf/gf/v2/util/gconv"
"github.com/gogf/gf/v2/util/guid"
)
const (
adminCommandTable = "im_admin_commands"
adminOutboxTable = "im_admin_outbox"
)
func PublishAdminCommand(ctx context.Context, in AdminCommandInput) (res *AdminCommandModel, err error) {
if in.TenantID <= 0 || in.TenantCode == "" {
return nil, gerror.New("IM租户信息不完整")
}
if in.EventType == "" {
return nil, gerror.New("IM命令类型不能为空")
}
if in.IdempotencyKey == "" {
in.IdempotencyKey = "cmd_" + guid.S()
}
commandID := "cmd_" + guid.S()
eventID := "evt_" + guid.S()
now := time.Now().UTC().Round(time.Millisecond)
payload := in.Payload
if payload == nil {
payload = map[string]any{}
}
payload["command_id"] = commandID
payloadBytes, err := json.Marshal(payload)
if err != nil {
return nil, gerror.Wrap(err, "序列化IM命令Payload失败")
}
envelope := Envelope{
EventID: eventID,
TraceID: gctx.CtxId(ctx),
TenantCode: in.TenantCode,
SourceService: consts.ImKafkaSourceService,
TargetService: consts.ImKafkaTargetService,
EventType: in.EventType,
BizType: in.BizType,
BizID: in.BizID,
IdempotencyKey: in.IdempotencyKey,
Version: in.Version,
OccurredAt: now,
Actor: in.Actor,
Payload: payloadBytes,
}
envelopeBytes, err := json.Marshal(envelope)
if err != nil {
return nil, gerror.Wrap(err, "序列化IM命令Envelope失败")
}
db := g.DB()
err = db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
data := g.Map{
"command_id": commandID,
"tenant_id": in.TenantID,
"tenant_code": in.TenantCode,
"event_id": eventID,
"event_type": in.EventType,
"biz_type": in.BizType,
"biz_id": in.BizID,
"idempotency_key": in.IdempotencyKey,
"status": consts.ImAdminCommandStatusPending,
"payload": string(payloadBytes),
"envelope": string(envelopeBytes),
"created_by": in.CreatedBy,
"created_at": gtime.Now(),
"updated_at": gtime.Now(),
}
if _, err := tx.Model(adminCommandTable).Safe().Ctx(ctx).Data(data).Insert(); err != nil {
return gerror.Wrap(err, "创建IM后台命令失败")
}
outbox := g.Map{
"event_id": eventID,
"command_id": commandID,
"topic": consts.QueueImAdminCommandTopic,
"event_type": in.EventType,
"biz_type": in.BizType,
"biz_id": in.BizID,
"key_value": in.BizID,
"envelope": string(envelopeBytes),
"status": consts.ImAdminOutboxStatusPending,
"created_at": gtime.Now(),
"updated_at": gtime.Now(),
}
if _, err := tx.Model(adminOutboxTable).Safe().Ctx(ctx).Data(outbox).Insert(); err != nil {
return gerror.Wrap(err, "创建IM后台命令Outbox失败")
}
return nil
})
if err != nil {
return nil, err
}
res = &AdminCommandModel{
CommandID: commandID,
EventID: eventID,
Status: consts.ImAdminCommandStatusPending,
TenantID: in.TenantID,
TenantCode: in.TenantCode,
EventType: in.EventType,
IdempotencyKey: in.IdempotencyKey,
}
if err = DispatchOutboxByEvent(ctx, eventID); err != nil {
return res, err
}
res.Status = consts.ImAdminCommandStatusRunning
return res, nil
}
func DispatchOutboxByEvent(ctx context.Context, eventID string) error {
var row gdb.Record
if err := g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).
Where("event_id", eventID).
Where("status", consts.ImAdminOutboxStatusPending).
Scan(&row); err != nil {
return gerror.Wrap(err, "读取IM命令Outbox失败")
}
if row.IsEmpty() {
return nil
}
return dispatchOutboxRecord(ctx, row)
}
func PublishConfigEvent(ctx context.Context, in ConfigEventInput) error {
if in.EventType == "" {
return gerror.New("IM配置事件类型不能为空")
}
eventID := "evt_" + guid.S()
now := time.Now().UTC().Round(time.Millisecond)
payload := map[string]any{
"tenant_code": in.TenantCode,
"tenant_id": in.TenantID,
"group": in.Group,
"keys": in.Keys,
}
payloadBytes, err := json.Marshal(payload)
if err != nil {
return gerror.Wrap(err, "序列化IM配置事件Payload失败")
}
envelope := Envelope{
EventID: eventID,
TraceID: gctx.CtxId(ctx),
TenantCode: in.TenantCode,
SourceService: consts.ImKafkaSourceService,
TargetService: consts.ImKafkaTargetService,
EventType: in.EventType,
BizType: "config",
BizID: in.BizID,
OccurredAt: now,
Actor: in.Actor,
Payload: payloadBytes,
}
envelopeBytes, err := json.Marshal(envelope)
if err != nil {
return gerror.Wrap(err, "序列化IM配置事件Envelope失败")
}
_, err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).Data(g.Map{
"event_id": eventID,
"command_id": "",
"topic": consts.QueueImConfigEventTopic,
"event_type": in.EventType,
"biz_type": "config",
"biz_id": in.BizID,
"key_value": in.TenantCode,
"envelope": string(envelopeBytes),
"status": consts.ImAdminOutboxStatusPending,
"created_at": gtime.Now(),
"updated_at": gtime.Now(),
}).Insert()
if err != nil {
return gerror.Wrap(err, "创建IM配置事件Outbox失败")
}
return DispatchOutboxByEvent(ctx, eventID)
}
func PublishEvent(ctx context.Context, in EventInput) (eventID string, err error) {
if in.Topic == "" {
return "", gerror.New("IM Kafka Topic不能为空")
}
if in.EventType == "" {
return "", gerror.New("IM Kafka事件类型不能为空")
}
eventID = "evt_" + guid.S()
now := time.Now().UTC().Round(time.Millisecond)
payload := in.Payload
if payload == nil {
payload = map[string]any{}
}
payloadBytes, err := json.Marshal(payload)
if err != nil {
return "", gerror.Wrap(err, "序列化IM Kafka事件Payload失败")
}
targetService := in.TargetService
if targetService == "" {
targetService = consts.ImKafkaTargetService
}
envelope := Envelope{
EventID: eventID,
TraceID: gctx.CtxId(ctx),
TenantCode: in.TenantCode,
SourceService: consts.ImKafkaSourceService,
TargetService: targetService,
EventType: in.EventType,
BizType: in.BizType,
BizID: in.BizID,
IdempotencyKey: in.IdempotencyKey,
Version: in.Version,
OccurredAt: now,
Actor: in.Actor,
Payload: payloadBytes,
}
envelopeBytes, err := json.Marshal(envelope)
if err != nil {
return "", gerror.Wrap(err, "序列化IM Kafka事件Envelope失败")
}
_, err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).Data(g.Map{
"event_id": eventID,
"command_id": "",
"topic": in.Topic,
"event_type": in.EventType,
"biz_type": in.BizType,
"biz_id": in.BizID,
"key_value": in.BizID,
"envelope": string(envelopeBytes),
"status": consts.ImAdminOutboxStatusPending,
"created_at": gtime.Now(),
"updated_at": gtime.Now(),
}).Insert()
if err != nil {
return "", gerror.Wrap(err, "创建IM Kafka事件Outbox失败")
}
if err = DispatchOutboxByEvent(ctx, eventID); err != nil {
return eventID, err
}
return eventID, nil
}
func DispatchPendingOutbox(ctx context.Context, limit int) (sent int, err error) {
if limit <= 0 {
limit = 100
}
var rows []gdb.Record
err = g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).
WhereIn("status", []string{consts.ImAdminOutboxStatusPending, consts.ImAdminOutboxStatusFailed}).
Where("next_retry_at IS NULL OR next_retry_at <= NOW()").
OrderAsc("id").
Limit(limit).
Scan(&rows)
if err != nil {
return 0, gerror.Wrap(err, "读取待投递IM命令Outbox失败")
}
for _, row := range rows {
if err = dispatchOutboxRecord(ctx, row); err != nil {
g.Log().Warningf(ctx, "IM命令Outbox投递失败 event:%s err:%+v", row["event_id"].String(), err)
continue
}
sent++
}
return sent, nil
}
func dispatchOutboxRecord(ctx context.Context, row gdb.Record) error {
eventID := row["event_id"].String()
commandID := row["command_id"].String()
topic := row["topic"].String()
body := row["envelope"].String()
if topic == "" || body == "" {
return markOutboxFailed(ctx, eventID, "MALFORMED_OUTBOX", "topic or envelope is empty")
}
if g.Cfg().MustGet(ctx, "queue.driver").String() != "kafka" {
err := gerror.New("IM Kafka未启用，请将queue.driver配置为kafka")
_ = markOutboxFailed(ctx, eventID, "KAFKA_DISABLED", err.Error())
return err
}
if err := queue.Push(topic, body); err != nil {
_ = markOutboxFailed(ctx, eventID, "KAFKA_SEND_FAILED", err.Error())
return err
}
now := gtime.Now()
_, err := g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).
Where("event_id", eventID).
Data(g.Map{
"status": consts.ImAdminOutboxStatusSent,
"attempts": gdb.Raw("attempts + 1"),
"sent_at": now,
"last_reason": "",
"last_error": "",
"updated_at": now,
}).Update()
if err != nil {
return gerror.Wrap(err, "更新IM命令Outbox投递状态失败")
}
if commandID != "" {
_, err = g.DB().Model(adminCommandTable).Safe().Ctx(ctx).
Where("command_id", commandID).
Data(g.Map{
"status": consts.ImAdminCommandStatusRunning,
"sent_at": now,
"updated_at": now,
}).Update()
if err != nil {
return gerror.Wrap(err, "更新IM后台命令投递状态失败")
}
}
return nil
}
func markOutboxFailed(ctx context.Context, eventID string, reason string, message string) error {
now := gtime.Now()
_, err := g.DB().Model(adminOutboxTable).Safe().Ctx(ctx).
Where("event_id", eventID).
Data(g.Map{
"status": consts.ImAdminOutboxStatusFailed,
"attempts": gdb.Raw("attempts + 1"),
"next_retry_at": now.Add(30 * time.Second),
"last_reason": reason,
"last_error": message,
"updated_at": now,
}).Update()
if err != nil {
return gerror.Wrap(err, "更新IM命令Outbox失败状态失败")
}
return nil
}
func ApplyCommandResult(ctx context.Context, result CommandResult) error {
commandID := result.CommandID
if commandID == "" {
commandID = result.CommandEventID
}
if commandID == "" && result.CommandEventID == "" {
return gerror.New("IM命令结果缺少command_id")
}
status := gconv.String(result.Status)
switch status {
case consts.ImAdminCommandStatusSucceeded, consts.ImAdminCommandStatusFailed, consts.ImAdminCommandStatusDead:
default:
status = consts.ImAdminCommandStatusFailed
}
data := g.Map{
"status": status,
"reason": result.Reason,
"message": result.Message,
"completed_at": gtime.Now(),
"updated_at": gtime.Now(),
}
if len(result.Result) > 0 {
data["result"] = string(result.Result)
}
mod := g.DB().Model(adminCommandTable).Safe().Ctx(ctx)
if result.CommandID != "" {
mod = mod.Where("command_id", result.CommandID)
} else {
mod = mod.Where("event_id", result.CommandEventID)
}
_, err := mod.Data(data).Update()
if err != nil {
return gerror.Wrap(err, "更新IM后台命令结果失败")
}
return nil
}
server/internal/library/imkafka/envelope.go
+90
View File
@@ -0,0 +1,90 @@
package imkafka
import (
"encoding/json"
"time"
)
type Actor struct {
Type string `json:"type,omitempty"`
ID string `json:"id,omitempty"`
DisplayName string `json:"display_name,omitempty"`
}
type Envelope struct {
EventID string `json:"event_id"`
TraceID string `json:"trace_id,omitempty"`
TenantCode string `json:"tenant_code,omitempty"`
SourceService string `json:"source_service,omitempty"`
TargetService string `json:"target_service,omitempty"`
EventType string `json:"event_type"`
BizType string `json:"biz_type,omitempty"`
BizID string `json:"biz_id,omitempty"`
IdempotencyKey string `json:"idempotency_key,omitempty"`
Version int64 `json:"version,omitempty"`
OccurredAt time.Time `json:"occurred_at"`
Actor *Actor `json:"actor,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
ParentEventID string `json:"parent_event_id,omitempty"`
}
type CommandResult struct {
EventID string `json:"event_id"`
TraceID string `json:"trace_id,omitempty"`
TenantCode string `json:"tenant_code,omitempty"`
CommandID string `json:"command_id"`
CommandEventID string `json:"command_event_id"`
Status string `json:"status"`
Reason string `json:"reason,omitempty"`
Message string `json:"message,omitempty"`
Result json.RawMessage `json:"result,omitempty"`
OccurredAt time.Time `json:"occurred_at"`
}
type AdminCommandInput struct {
TenantID int64
TenantCode string
EventType string
BizType string
BizID string
IdempotencyKey string
Version int64
Actor *Actor
Payload map[string]any
CreatedBy int64
}
type ConfigEventInput struct {
TenantID int64
TenantCode string
EventType string
BizID string
Group string
Keys []string
Actor *Actor
CreatedBy int64
}
type EventInput struct {
Topic string
TenantCode string
TargetService string
EventType string
BizType string
BizID string
IdempotencyKey string
Version int64
Actor *Actor
Payload map[string]any
CreatedBy int64
}
type AdminCommandModel struct {
CommandID string `json:"commandId" dc:"命令ID"`
EventID string `json:"eventId" dc:"Kafka事件ID"`
Status string `json:"status" dc:"状态"`
TenantID int64 `json:"tenantId" dc:"租户ID"`
TenantCode string `json:"tenantCode" dc:"租户编码"`
EventType string `json:"eventType" dc:"事件类型"`
IdempotencyKey string `json:"idempotencyKey" dc:"幂等键"`
}
server/internal/library/queue/disk.go
+1 -1
View File
@@ -125,7 +125,7 @@ func (d *DiskProducerMq) getProducer(topic string) *disk.Queue {
func NewDiskQueue(topic string, config *disk.Config) *disk.Queue {
conf := &disk.Config{
Path: fmt.Sprintf(config.Path + "/" + config.GroupName + "/" + topic),
Path: fmt.Sprintf("%s/%s/%s", config.Path, config.GroupName, topic),
BatchSize: config.BatchSize,
BatchTime: config.BatchTime * time.Second,
SegmentSize: config.SegmentSize,
server/internal/library/queue/kafkamq.go
+81 -20
View File
@@ -7,13 +7,16 @@ package queue
import (
"context"
"fmt"
"crypto/tls"
"strings"
"time"
"github.com/IBM/sarama"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
"github.com/xdg-go/scram"
"hotgo/internal/consts"
"hotgo/utility/simple"
"time"
)
type KafkaMq struct {
@@ -23,14 +26,17 @@ type KafkaMq struct {
}
type KafkaConfig struct {
ClientId string
Brokers []string
GroupID string
Partitions int32
Replication int16
Version string
UserName string
Password string
ClientId string
Brokers []string
GroupID string
Partitions int32
Replication int16
Version string
UserName string
Password string
SASLEnabled bool
SASLMechanism string
TLSEnabled bool
}
// SendMsg 按字符串类型生产数据
@@ -98,7 +104,7 @@ func (r *KafkaMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg))
}
if consumerCtx.Err() != nil {
Logger().Debugf(ctx, fmt.Sprintf("kafka consoumer stop : %v", consumerCtx.Err()))
Logger().Debugf(ctx, "kafka consoumer stop : %v", consumerCtx.Err())
return
}
consumer.ready = make(chan bool)
@@ -134,10 +140,8 @@ func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error)
conf := sarama.NewConfig()
conf.Consumer.Return.Errors = true
conf.Version = kfkVersion
if connOpt.UserName != "" {
conf.Net.SASL.Enable = true
conf.Net.SASL.User = connOpt.UserName
conf.Net.SASL.Password = connOpt.Password
if err = configureKafkaSecurity(conf, connOpt); err != nil {
return
}
// 默认按随机方式消费
@@ -157,7 +161,9 @@ func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error)
// RegisterKafkaProducer 注册并启动生产者接口实现
func RegisterKafkaProducer(connOpt KafkaConfig) (client MqProducer, err error) {
mqIns := &KafkaMq{}
connOpt.ClientId = "HOTGO-Producer"
if connOpt.ClientId == "" {
connOpt.ClientId = "hotgo-producer"
}
// 这里如果使用go程需要处理chan同步问题
if err = doRegisterKafkaProducer(connOpt, mqIns); err != nil {
@@ -191,10 +197,8 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
conf.ClientID = connOpt.ClientId
conf.Version = kfkVersion
if connOpt.UserName != "" {
conf.Net.SASL.Enable = true
conf.Net.SASL.User = connOpt.UserName
conf.Net.SASL.Password = connOpt.Password
if err = configureKafkaSecurity(conf, connOpt); err != nil {
return
}
mqIns.producerIns, err = sarama.NewAsyncProducer(brokers, conf)
@@ -209,6 +213,63 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
return
}
func configureKafkaSecurity(conf *sarama.Config, connOpt KafkaConfig) error {
if connOpt.TLSEnabled {
conf.Net.TLS.Enable = true
conf.Net.TLS.Config = &tls.Config{MinVersion: tls.VersionTLS12}
}
if !connOpt.SASLEnabled && connOpt.UserName == "" {
return nil
}
if connOpt.UserName == "" || connOpt.Password == "" {
return gerror.New("queue kafka sasl username or password is empty")
}
conf.Net.SASL.Enable = true
conf.Net.SASL.User = connOpt.UserName
conf.Net.SASL.Password = connOpt.Password
mechanism := strings.ToUpper(strings.ReplaceAll(connOpt.SASLMechanism, "_", "-"))
switch mechanism {
case "", "PLAIN":
conf.Net.SASL.Mechanism = sarama.SASLTypePlaintext
case "SCRAM-SHA-256":
conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA256
conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
return &xdgSCRAMClient{HashGeneratorFcn: scram.SHA256}
}
case "SCRAM-SHA-512":
conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
return &xdgSCRAMClient{HashGeneratorFcn: scram.SHA512}
}
default:
return gerror.New("queue kafka sasl mechanism is not support")
}
return nil
}
type xdgSCRAMClient struct {
*scram.Client
*scram.ClientConversation
scram.HashGeneratorFcn
}
func (x *xdgSCRAMClient) Begin(userName, password, authzID string) (err error) {
x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID)
if err != nil {
return err
}
x.ClientConversation = x.Client.NewConversation()
return nil
}
func (x *xdgSCRAMClient) Step(challenge string) (response string, err error) {
return x.ClientConversation.Step(challenge)
}
func (x *xdgSCRAMClient) Done() bool {
return x.ClientConversation.Done()
}
// validateVersion 验证版本是否有效
func validateVersion(version sarama.KafkaVersion) bool {
for _, item := range sarama.SupportedVersions {
server/internal/library/queue/queue.go
+51 -7
View File
@@ -11,6 +11,8 @@ import (
"github.com/gogf/gf/v2/os/gctx"
"hotgo/internal/library/queue/disk"
"hotgo/utility/charset"
"os"
"strings"
"sync"
"time"
)
@@ -59,6 +61,12 @@ type KafkaConf struct {
Version string `json:"version"`
RandClient bool `json:"randClient"`
MultiConsumer bool `json:"multiConsumer"`
ClientId string `json:"clientId"`
UserName string `json:"userName"`
Password string `json:"password"`
SASLEnabled bool `json:"saslEnabled"`
SASLMechanism string `json:"saslMechanism"`
TLSEnabled bool `json:"tlsEnabled"`
}
type MqMsg struct {
@@ -85,6 +93,7 @@ func init() {
if err := g.Cfg().MustGet(ctx, "queue").Scan(&config); err != nil {
Logger().Warningf(ctx, "queue init err:%+v", err)
}
normalizeKafkaConf()
}
// InstanceConsumer 实例化消费者
@@ -121,9 +130,15 @@ func NewProducer(groupName string) (mqClient MqProducer, err error) {
return
}
mqClient, err = RegisterKafkaProducer(KafkaConfig{
Brokers: config.Kafka.Address,
GroupID: groupName,
Version: config.Kafka.Version,
Brokers: config.Kafka.Address,
GroupID: groupName,
Version: config.Kafka.Version,
ClientId: firstKafkaClientId(config.Kafka.ClientId, "hotgo-producer"),
UserName: config.Kafka.UserName,
Password: config.Kafka.Password,
SASLEnabled: config.Kafka.SASLEnabled,
SASLMechanism: config.Kafka.SASLMechanism,
TLSEnabled: config.Kafka.TLSEnabled,
})
case "redis":
if _, err = g.Redis().Do(ctx, "ping"); err == nil {
@@ -185,10 +200,15 @@ func NewConsumer(groupName string) (mqClient MqConsumer, err error) {
}
mqClient, err = RegisterKafkaMqConsumer(KafkaConfig{
Brokers: config.Kafka.Address,
GroupID: groupName,
Version: config.Kafka.Version,
ClientId: clientId,
Brokers: config.Kafka.Address,
GroupID: groupName,
Version: config.Kafka.Version,
ClientId: clientId,
UserName: config.Kafka.UserName,
Password: config.Kafka.Password,
SASLEnabled: config.Kafka.SASLEnabled,
SASLMechanism: config.Kafka.SASLMechanism,
TLSEnabled: config.Kafka.TLSEnabled,
})
case "redis":
if _, err = g.Redis().Do(ctx, "ping"); err == nil {
@@ -213,6 +233,30 @@ func NewConsumer(groupName string) (mqClient MqConsumer, err error) {
return
}
func normalizeKafkaConf() {
config.Kafka.UserName = resolveEnvPlaceholder(config.Kafka.UserName)
config.Kafka.Password = resolveEnvPlaceholder(config.Kafka.Password)
if config.Kafka.SASLMechanism == "" {
config.Kafka.SASLMechanism = "SCRAM-SHA-512"
}
}
func resolveEnvPlaceholder(value string) string {
value = strings.TrimSpace(value)
if strings.HasPrefix(value, "${") && strings.HasSuffix(value, "}") {
return os.Getenv(strings.TrimSuffix(strings.TrimPrefix(value, "${"), "}"))
}
return value
}
func firstKafkaClientId(value string, fallback string) string {
value = strings.TrimSpace(value)
if value != "" {
return value
}
return fallback
}
// BodyString 返回消息体
func (m *MqMsg) BodyString() string {
return string(m.Body)