优化

githubzhaoqian · Sep 2, 2024 · d62c971 · d62c971
1 parent 842fed1
commit d62c971
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 5 deletions.
diff --git a/pkg/rediszset/publisher.go b/pkg/rediszset/publisher.go
@@ -93,17 +93,16 @@ func (p *Publisher) Publish(topic string, msgs ...*message.Message) error {
 		if err != nil {
 			return errors.Wrapf(err, "cannot marshal message %s", msg.UUID)
 		}
-
-		id, err := p.client.ZAdd(msg.Context(), topic, redis.Z{Score: score, Member: msg.UUID}).Result()
-		if err != nil {
-			return errors.Wrapf(err, "cannot zadd message %s", msg.UUID)
-		}
 		// 设置string 值
 		valueKey := getKey(topic, msg.UUID)
 		_, err = p.client.Set(msg.Context(), valueKey, value, 0).Result()
 		if err != nil {
 			return errors.Wrapf(err, "cannot set message %s", msg.UUID)
 		}
+		id, err := p.client.ZAdd(msg.Context(), topic, redis.Z{Score: score, Member: msg.UUID}).Result()
+		if err != nil {
+			return errors.Wrapf(err, "cannot zadd message %s", msg.UUID)
+		}
 		logFields["zadd_id"] = id
 		p.logger.Trace("Message sent to redis stream", logFields)
 	}

diff --git a/pkg/rediszset/subscriber.go b/pkg/rediszset/subscriber.go
@@ -398,6 +398,10 @@ func (h *messageHandler) processMessage(ctx context.Context, topic string, read
 	src := read.Z.Member.(string)
 	srcKey := getKey(topic, src)
 	member, err := h.rc.Get(ctx, srcKey).Result()
+	if errors.Is(err, redis.Nil) {
+		h.logger.Error("Message nil", err, watermill.LogFields{"key": srcKey})
+		return nil
+	}
 	if err != nil {
 		return errors.Wrapf(err, "message %s get failed", srcKey)
 	}