+queue:藏品生成队列及消费逻辑实现;

liquidnet-bus-api/liquidnet-service-goblin-api/src/main/java/com/liquidnet/service/goblin/constant/GoblinRedisConst.java

@@ -166,6 +166,25 @@ public class GoblinRedisConst {
      */
     public static final String USER_COUPON = PREFIX.concat("user_coupon:");
 
+    /* ----------------------------------------------------------------- */
+
+    /**
+     * 我的藏品ID列表（首页）
+     * {goblin:u_d_art:${uid}, JsonUtils.toJson(List<String:artwork_id>)}
+     */
+    public static final String USER_DIGITAL_ARTWORK_IDS = PREFIX.concat("u_d_art_ids:");
+
+    /**
+     * 我的藏品信息
+     * {goblin:u_d_art:${artwork_id}, JsonUtils.toJson(com.liquidnet.service.goblin.dto.vo.GoblinDigitalArtworkVo)}
+     */
+    public static final String USER_DIGITAL_ARTWORK = PREFIX.concat("u_d_art:");
+    /**
+     * 我的藏品生成标记（有时效）
+     * {goblin:u_d_art:${uid+skuId+orderId},1}
+     */
+    public static final String USER_DIGITAL_ARTWORK_GENMARK = PREFIX.concat("u_d_art_gm:");
+
     /* ----------------------------------------------------------------- */
     /* ----------------------------------------------------------------- */
 }

liquidnet-bus-api/liquidnet-service-goblin-api/src/main/java/com/liquidnet/service/goblin/dto/GoblinQueueBizArtworkGenDto.java

+package com.liquidnet.service.goblin.dto;
+
+import com.liquidnet.commons.lang.util.JsonUtils;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 购买藏品、开启盲盒后，生成藏品队列消息体
+ *
+ * @author zhanggb
+ * Created by IntelliJ IDEA at 2022/3/31
+ */
+@Data
+public class GoblinQueueBizArtworkGenDto implements Serializable, Cloneable {
+    private static final long serialVersionUID = 8267639695935038399L;
+    private String uid;
+    private String skuId;
+    private String orderId;
+    private Integer source;
+
+    private static final GoblinQueueBizArtworkGenDto obj = new GoblinQueueBizArtworkGenDto();
+
+    public static GoblinQueueBizArtworkGenDto getNew() {
+        try {
+            return (GoblinQueueBizArtworkGenDto) obj.clone();
+        } catch (CloneNotSupportedException e) {
+            return new GoblinQueueBizArtworkGenDto();
+        }
+    }
+
+    public String toJson() {
+        return JsonUtils.toJson(this);
+    }
+}

liquidnet-bus-api/liquidnet-service-goblin-api/src/main/java/com/liquidnet/service/goblin/dto/manage/GoblinStoreMgtDigitalGoodsAddSkuParam.java

@@ -31,7 +31,8 @@ public class GoblinStoreMgtDigitalGoodsAddSkuParam implements Serializable {
     @NotNull(message = "藏品名称不能为空")
     @Size(max = 36, message = "藏品名称内容过长")
     private String name;
-    @ApiModelProperty(position = 14, required = false, value = "款式名称[36]，`unbox=0`时必传", example = "款式名称...")
+    @ApiModelProperty(position = 14, required = true, value = "款式名称[36]，`unbox=0`时必传", example = "款式名称...")
+    @NotNull(message = "款式名称不能为空")
     @Size(max = 36, message = "款式名称内容过长")
     private String subtitle;
     @ApiModelProperty(position = 15, required = true, value = "藏品封面图片URL[256]", example = "http://123")
@@ -100,9 +101,12 @@ public class GoblinStoreMgtDigitalGoodsAddSkuParam implements Serializable {
     @NotNull(message = "藏品详情不能为空")
     private String details;
 
-    @ApiModelProperty(position = 31, required = true, value = "上架处理方式[1-等待手动上架|2-直接上架售卖|3-预约定时上架]", example = "1")
+//    @ApiModelProperty(position = 31, required = true, value = "上架处理方式[1-等待手动上架|2-直接上架售卖|3-预约定时上架]", example = "1")
+//    @NotNull(message = "上架处理方式不能为空")
+//    @Pattern(regexp = "\\b(1|2|3)\\b", message = "上架处理方式参数无效")
+    @ApiModelProperty(position = 31, required = true, value = "上架处理方式[2-直接上架售卖|3-预约定时上架]", example = "1")
     @NotNull(message = "上架处理方式不能为空")
-    @Pattern(regexp = "\\b(1|2|3)\\b", message = "上架处理方式参数无效")
+    @Pattern(regexp = "\\b(2|3)\\b", message = "上架处理方式参数无效")
     private String shelvesHandle;
     @ApiModelProperty(position = 32, required = false, value = "预约上架时间[yyyy-MM-dd HH:mm:ss][上架处理方式为3-预约定时上架时需要指定]", example = "2022-03-25 00:00:00")
     @Pattern(regexp = LnsRegex.Valid.DATETIME_FULL, message = "预约上架时间格式有误")

liquidnet-bus-api/liquidnet-service-goblin-api/src/main/java/com/liquidnet/service/goblin/dto/vo/GoblinDigitalArtworkListVo.java

@@ -15,11 +15,11 @@ import java.time.LocalDateTime;
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class GoblinDigitalArtworkListVo implements Serializable, Cloneable {
     private static final long serialVersionUID = -1510553574954846520L;
-    @ApiModelProperty(position = 10, value = "是否盲盒[0-否|1-是]")
+    @ApiModelProperty(position = 10, value = "藏品Id")
+    private String artworkId;
+    @ApiModelProperty(position = 11, value = "是否盲盒[0-否|1-是]")
     private Integer unbox;
 
-    @ApiModelProperty(position = 11, value = "藏品Id")
-    private String artworkId;
     @ApiModelProperty(position = 12, value = "藏品名称")
     private String name;
     @ApiModelProperty(position = 13, value = "藏品副名称")
@@ -30,8 +30,8 @@ public class GoblinDigitalArtworkListVo implements Serializable, Cloneable {
     private Integer editionSn;
     @ApiModelProperty(position = 16, value = "藏品发行量")
     private Integer edition;
-    @ApiModelProperty(position = 17, value = "获得方式[0-购买|1-兑换]")
-    private Integer getway;
+    @ApiModelProperty(position = 17, value = "获得方式[1-购买|2-兑换|3-赠送|5-受赠]")
+    private Integer source;
 
     @ApiModelProperty(position = 18, value = "藏品状态，根据`unbox`区分盲盒来判断[0-生成中/未开启|1-已生成/已开启|2-生成失败/开启失败]")
     private Integer state;

liquidnet-bus-api/liquidnet-service-goblin-api/src/main/java/com/liquidnet/service/goblin/dto/vo/GoblinDigitalArtworkVo.java

@@ -11,8 +11,8 @@ import java.time.LocalDateTime;
 public class GoblinDigitalArtworkVo implements Serializable, Cloneable {
     private static final long serialVersionUID = -5920952408993245963L;
     private String artworkId;
-    private String skuId;
     private String uid;
+    private String skuId;
     private String orderId;
     private Integer editionSn;
     private String nftId;

liquidnet-bus-common/liquidnet-common-service-base/src/main/java/com/liquidnet/service/base/constant/MQConst.java

@@ -239,6 +239,7 @@ public class MQConst {
         SQL_STORE("goblin:stream:sql.store", "group.sql.ustore", "店铺相关"),
         SQL_GOODS("goblin:stream:sql.goods", "group.sql.ustore", "商品相关"),
         SQL_COUPON("goblin:stream:sql.coupon", "group.sql.ustore", "店铺优惠券相关"),
+
         GOBLIN_STORE_MARKET("goblin:stream:store.market", "group.store.market", "店铺活动"),
         GOBLIN_SELF_MARKET("goblin:stream:self.market", "group.self.market", "平台活动"),
         GOBLIN_ORDER_CREATE_PAY("goblin:stream:order:create_pay", "group.order:create_pay", "订单创建&支付"),
@@ -260,6 +261,9 @@ public class MQConst {
         GOBLIN_UN_PAY_7("goblin:stream:order:back:7", "group.order:back", "回滚关闭订单库存队列"),
         GOBLIN_UN_PAY_8("goblin:stream:order:back:8", "group.order:back", "回滚关闭订单库存队列"),
         GOBLIN_UN_PAY_9("goblin:stream:order:back:9", "group.order:back", "回滚关闭订单库存队列"),
+
+        BIZ_ARTWORK_GEN("goblin:stream:biz_art:gen", "group.biz.artwork", "藏品生成"),
+        SQL_ARTWORK_GEN("goblin:stream:sql_art:gen", "group.biz.artwork", "藏品生成"),
         ;
 
         private final String key;

liquidnet-bus-do/liquidnet-service-goblin-do/src/main/java/com/liquidnet/service/goblin/entity/GoblinDigitalArtwork.java

@@ -30,14 +30,14 @@ public class GoblinDigitalArtwork implements Serializable {
     private String artworkId;
 
     /**
-     * 单品ID
+     * 用户UID
      */
-    private String skuId;
+    private String uid;
 
     /**
-     * 用户UID
+     * 单品ID
      */
-    private String uid;
+    private String skuId;
 
     /**
      * 藏品订单号

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/test/java/TestAdam.java

-import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.DESUtils;
 import com.liquidnet.commons.lang.util.HttpUtil;
 import com.liquidnet.commons.lang.util.IdentityUtils;
@@ -74,12 +73,16 @@ public class TestAdam {
 
                 "sweet:manual:pushList",
                 "goblin:sku:relation:226416501783461229",
-                "goblin:bsc:goods:2264185547679ZZ87427",
+                "goblin:bsc:goods:2284183012778",
                 "goblin:bsc:goods_sku:226418554767964895ZZ87427",
                 "goblin:user_coupon:926591841436344323959830",
                 "goblin:bsc:ustore:809406",
                 "goblin:bsc:ustore:922833055536291845022973",
                 "goblin:bsc:store:839fa8dde17b5bf248575370f1249eab",
+                "goblin:bsc:self_g_c",
+                "goblin:u_d_art:2290114757532",
+                "goblin:u_d_art_ids:481073045099397123024514",
+                "goblin:u_d_art:TEST123228418301277894015TEST1234567890",
         };
         for (String key : keys) {
             long value = key.hashCode();
@@ -104,12 +107,5 @@ public class TestAdam {
 
     @Test
     public void testTmp() {
-//        String post = HttpUtil.post("http://ttestkylin.zhengzai.tv/kylin/inner/cache/member?uid=925802662655180800214832", CollectionUtil.linkedMultiValueMapStringString());
-//        System.out.println("===" + post);
-
-        LocalDateTime now = LocalDateTime.now();
-        for (int i = 0; i < 15; i++) {
-            System.out.println(now.getSecond() + "----" + now.getSecond()%10);
-        }
     }
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerCommonBizRedisStreamConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerGoblinBizArtworkGenRdsReceiver;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Configuration
+public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+    @Autowired
+    ConsumerGoblinBizArtworkGenRdsReceiver consumerGoblinBizArtworkGenRdsReceiver;
+
+    @Bean// 藏品生成
+    public List<Subscription> subscriptionBizArtworkGen(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.BIZ_ARTWORK_GEN;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 10; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinBizArtworkGenRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerCommonSqlRedisStreamConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerGoblinSqlArtworkGenRdsReceiver;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Configuration
+public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+    @Autowired
+    ConsumerGoblinSqlArtworkGenRdsReceiver consumerGoblinSqlArtworkGenRdsReceiver;
+
+    @Bean// 藏品生成
+    public List<Subscription> subscriptionSqlArtworkGen(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_ARTWORK_GEN;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 5; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinSqlArtworkGenRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/AbstractBizRedisReceiver.java

+package com.liquidnet.service.consumer.kylin.receiver;
+
+import com.liquidnet.service.consumer.kylin.service.IBaseDao;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamListener;
+
+/**
+ * 公共的业务队列消息监听器，具体业务消费逻辑通过`consumerMessageHandler`实现
+ *
+ * @author zhanggb
+ * Created by IntelliJ IDEA at 2022/3/31
+ */
+@Slf4j
+public abstract class AbstractBizRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
+    @Autowired
+    public IBaseDao baseDao;
+    @Autowired
+    public StringRedisTemplate stringRedisTemplate;
+
+    @Override
+    public void onMessage(MapRecord<String, String, String> message) {
+        String redisStreamKey = this.getRedisStreamKey();
+        log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
+        boolean result = this.consumerMessageHandler(message.getValue().get("message"));
+        log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
+
+        try {
+            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
+        } catch (Exception e) {
+            log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
+        }
+        try {
+            stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
+        } catch (Exception e) {
+            log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
+        }
+    }
+
+    protected abstract boolean consumerMessageHandler(String msg);
+
+    protected abstract String getRedisStreamKey();
+
+    protected abstract String getRedisStreamGroup();
+}
\ No newline at end of file

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/AbstractSqlRedisReceiver.java

@@ -3,7 +3,7 @@ package com.liquidnet.service.consumer.kylin.receiver;
 import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.JsonUtils;
 import com.liquidnet.service.base.SqlMapping;
-import com.liquidnet.service.consumer.service.IBaseDao;
+import com.liquidnet.service.consumer.kylin.service.IBaseDao;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.connection.stream.MapRecord;
@@ -13,6 +13,12 @@ import org.springframework.data.redis.stream.StreamListener;
 
 import java.util.HashMap;
 
+/**
+ * 公共的SQL队列消息监听器，具体SQL消费逻辑统一使用`consumerMessageHandler`
+ *
+ * @author zhanggb
+ * Created by IntelliJ IDEA at 2022/3/31
+ */
 @Slf4j
 public abstract class AbstractSqlRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
     @Autowired

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/AbstractXlsRedisReceiver.java

@@ -6,7 +6,7 @@ import com.alibaba.fastjson.JSON;
 import com.liquidnet.common.cache.redis.util.RedisUtil;
 import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.service.consumer.kylin.dto.PhoneDto;
-import com.liquidnet.service.consumer.service.IBaseDao;
+import com.liquidnet.service.consumer.kylin.service.IBaseDao;
 import com.liquidnet.service.goblin.constant.GoblinRedisConst;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/ConsumerGoblinSqlArtworkGenRdsReceiver.java

+package com.liquidnet.service.consumer.kylin.receiver;
+
+import com.liquidnet.service.base.constant.MQConst;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ConsumerGoblinSqlArtworkGenRdsReceiver extends AbstractSqlRedisReceiver {
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GoblinQueue.SQL_ARTWORK_GEN.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GoblinQueue.SQL_ARTWORK_GEN.getGroup();
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/service/IBaseDao.java

-package com.liquidnet.service.consumer.service;
+package com.liquidnet.service.consumer.kylin.service;
 
 import java.util.LinkedList;
 

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/service/impl/BaseDao.java

 package com.liquidnet.service.consumer.kylin.service.impl;
 
 import com.liquidnet.commons.lang.util.JsonUtils;
-import com.liquidnet.service.consumer.service.IBaseDao;
+import com.liquidnet.service.consumer.kylin.service.IBaseDao;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.jdbc.core.JdbcTemplate;

liquidnet-bus-service/liquidnet-service-goblin/liquidnet-service-goblin-impl/docu/db_0315_jxl_nft.sql

@@ -83,8 +83,8 @@ create table goblin_digital_artwork
 (
     mid            bigint auto_increment primary key,
     artwork_id     varchar(64)  not null comment '藏品ID',
-    sku_id         varchar(64)  not null comment '单品ID',
     uid            varchar(64)  not null comment '用户UID',
+    sku_id         varchar(64)  not null comment '单品ID',
     order_id       varchar(64)  not null comment '藏品订单号',
     edition_sn     int          null comment '序列号',
     nft_id         varchar(256) null comment '藏品NFT ID',