sweet队列迁移到base

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/pom.xml

@@ -47,6 +47,21 @@
             <artifactId>liquidnet-service-goblin-api</artifactId>
             <version>1.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-service-sweet-do</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-api-feign-stone</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-service-sweet-api</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
 
         <!-- third party api -->
 

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

@@ -3,6 +3,8 @@ package com.liquidnet.service.consumer.base.config;
 import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
 import com.liquidnet.service.base.constant.MQConst;
 import com.liquidnet.service.consumer.base.receiver.*;
+import com.liquidnet.service.consumer.base.receiver.sweet.ConsumerSweetStoneIntegralReceiver;
+import com.liquidnet.service.consumer.base.receiver.sweet.ConsumerSweetTemplateMsgReceiver;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -40,6 +42,43 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
     ConsumerGalaxyJsonNftUserRegisterReceiver jsonNftUserRegisterReceiver;
     @Autowired
     ConsumerGoblinBizIntegralReceiver consumerGoblinBizIntegralReceiver;
+    /*------sweet------*/
+    @Autowired
+    ConsumerSweetTemplateMsgReceiver consumerSweetTemplateMsgReceiver;
+    @Autowired
+    ConsumerSweetStoneIntegralReceiver consumerSweetStoneIntegralReceiver;
+
+    /*------sweet------*/
+    @Bean // 发送模版消息
+    public List<Subscription> subscriptionSweetSqlApiTemplateMsg(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_TEMPLATE_MSG;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 3; 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()), consumerSweetTemplateMsgReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+    @Bean // stone积分处理
+    public List<Subscription> subscriptionSweetSqlApiStoneIntgral(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_STONE_INTEGRAL;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 3; 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()), consumerSweetStoneIntegralReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
 
     @Bean// 短信通知
     public List<Subscription> subscriptionCommonSmsNotice(RedisConnectionFactory factory) {

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/sweet/WechatMaConfigure.java

+package com.liquidnet.service.consumer.base.config.sweet;
+
+import cn.binarywang.wx.miniapp.api.WxMaService;
+import cn.binarywang.wx.miniapp.api.impl.WxMaServiceImpl;
+import cn.binarywang.wx.miniapp.config.impl.WxMaRedisBetterConfigImpl;
+import me.chanjar.weixin.common.redis.RedisTemplateWxRedisOps;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import javax.annotation.PostConstruct;
+
+@Configuration
+public class WechatMaConfigure {
+    private static String appletAppidZhengzai;
+    private static String appletSecretZhengzai;
+
+    private static String appletAppidStrawberry;
+    private static String appletSecretStrawberry;
+
+    private static String appletAppidFive;
+    private static String appletSecretFive;
+
+    private static String appletAppidMdsk;
+    private static String appletSecretMdsk;
+
+    @Value("${liquidnet.wechat.applet.zhengzai.appid}")
+    public void setAppletAppidZhengzai(String appletAppidZhengzai) {
+        WechatMaConfigure.appletAppidZhengzai = appletAppidZhengzai;
+    }
+
+    @Value("${liquidnet.wechat.applet.zhengzai.secret}")
+    public void setAppletSecretZhengzai(String appletSecretZhengzai) {
+        WechatMaConfigure.appletSecretZhengzai = appletSecretZhengzai;
+    }
+
+    @Value("${liquidnet.wechat.applet.strawberry.appid}")
+    public void setAppletAppidStrawberry(String appletAppidStrawberry) {
+        WechatMaConfigure.appletAppidStrawberry = appletAppidStrawberry;
+    }
+
+    @Value("${liquidnet.wechat.applet.strawberry.secret}")
+    public void setAppletSecretStrawberry(String appletSecretStrawberry) {
+        WechatMaConfigure.appletSecretStrawberry = appletSecretStrawberry;
+    }
+
+    @Value("${liquidnet.wechat.applet.five.appid}")
+    public void setAppletAppidFive(String appletAppidFive) {
+        WechatMaConfigure.appletAppidFive = appletAppidFive;
+    }
+
+    @Value("${liquidnet.wechat.applet.five.secret}")
+    public void setAppletSecretFive(String appletSecretFive) {
+        WechatMaConfigure.appletSecretFive = appletSecretFive;
+    }
+
+    @Value("${liquidnet.wechat.applet.mdsk.appid}")
+    public void setAppletAppidMdsk(String appletAppidMdsk) {
+        WechatMaConfigure.appletAppidMdsk = appletAppidMdsk;
+    }
+
+    @Value("${liquidnet.wechat.applet.mdsk.secret}")
+    public void setAppletSecretMdsk(String appletSecretMdsk) {
+        WechatMaConfigure.appletSecretMdsk = appletSecretMdsk;
+    }
+
+    private WxMaService wxMaAppletZhengzaiService;
+    private WxMaService wxMaAppletStrawberryService;
+    private WxMaService wxMaAppletFiveService;
+    private WxMaService wxMaAppletMdskService;
+
+    @Autowired
+    private StringRedisTemplate stringRedisTemplate;
+
+    @PostConstruct
+    public void init() {
+        wxMaAppletZhengzaiService = new WxMaServiceImpl() {
+            {
+                RedisTemplateWxRedisOps redisTemplateWxRedisOps = new RedisTemplateWxRedisOps(stringRedisTemplate);
+                WxMaRedisBetterConfigImpl wxMaDefaultConfig = new WxMaRedisBetterConfigImpl(redisTemplateWxRedisOps, "sweet:accessToken:applet:zhengzai");
+
+//                WxMaDefaultConfigImpl wxMaDefaultConfig = new WxMaDefaultConfigImpl();
+                wxMaDefaultConfig.setAppid(appletAppidZhengzai);
+                wxMaDefaultConfig.setSecret(appletSecretZhengzai);
+                wxMaDefaultConfig.setMsgDataFormat("JSON");
+
+                this.setWxMaConfig(wxMaDefaultConfig);
+            }
+        };
+        wxMaAppletStrawberryService = new WxMaServiceImpl() {
+            {
+                RedisTemplateWxRedisOps redisTemplateWxRedisOps = new RedisTemplateWxRedisOps(stringRedisTemplate);
+                WxMaRedisBetterConfigImpl wxMaDefaultConfig = new WxMaRedisBetterConfigImpl(redisTemplateWxRedisOps, "sweet:accessToken:applet:strawberry");
+//                WxMaDefaultConfigImpl wxMaDefaultConfig = new WxMaDefaultConfigImpl();
+                wxMaDefaultConfig.setAppid(appletAppidStrawberry);
+                wxMaDefaultConfig.setSecret(appletSecretStrawberry);
+                wxMaDefaultConfig.setMsgDataFormat("JSON");
+
+                this.setWxMaConfig(wxMaDefaultConfig);
+            }
+        };
+        wxMaAppletFiveService = new WxMaServiceImpl() {
+            {
+                RedisTemplateWxRedisOps redisTemplateWxRedisOps = new RedisTemplateWxRedisOps(stringRedisTemplate);
+                WxMaRedisBetterConfigImpl wxMaDefaultConfig = new WxMaRedisBetterConfigImpl(redisTemplateWxRedisOps, "sweet:accessToken:applet:five");
+//                WxMaDefaultConfigImpl wxMaDefaultConfig = new WxMaDefaultConfigImpl();
+                wxMaDefaultConfig.setAppid(appletAppidFive);
+                wxMaDefaultConfig.setSecret(appletSecretFive);
+                wxMaDefaultConfig.setMsgDataFormat("JSON");
+
+                this.setWxMaConfig(wxMaDefaultConfig);
+            }
+        };
+        wxMaAppletMdskService = new WxMaServiceImpl() {
+            {
+                RedisTemplateWxRedisOps redisTemplateWxRedisOps = new RedisTemplateWxRedisOps(stringRedisTemplate);
+                WxMaRedisBetterConfigImpl wxMaDefaultConfig = new WxMaRedisBetterConfigImpl(redisTemplateWxRedisOps, "sweet:accessToken:applet:mdsk");
+//                WxMaDefaultConfigImpl wxMaDefaultConfig = new WxMaDefaultConfigImpl();
+                wxMaDefaultConfig.setAppid(appletAppidMdsk);
+                wxMaDefaultConfig.setSecret(appletSecretMdsk);
+                wxMaDefaultConfig.setMsgDataFormat("JSON");
+
+                this.setWxMaConfig(wxMaDefaultConfig);
+            }
+        };
+    }
+
+    public WxMaService getWxMaService(Integer anum) {
+        switch (anum) {
+            case 4:
+                return wxMaAppletZhengzaiService;
+            case 1:
+                return wxMaAppletStrawberryService;
+            case 2:
+                return wxMaAppletFiveService;
+            case 3:
+                return wxMaAppletMdskService;
+        }
+        return null;
+    }
+
+    public String getWxMaAppid(Integer anum) {
+        switch (anum) {
+            case 4:
+                return appletAppidZhengzai;
+            case 1:
+                return appletAppidStrawberry;
+            case 2:
+                return appletAppidFive;
+            case 3:
+                return appletAppidMdsk;
+        }
+        return "";
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/sweet/WechatMpConfigure.java

+package com.liquidnet.service.consumer.base.config.sweet;
+
+import me.chanjar.weixin.common.redis.RedisTemplateWxRedisOps;
+import me.chanjar.weixin.mp.api.WxMpService;
+import me.chanjar.weixin.mp.api.impl.WxMpServiceImpl;
+import me.chanjar.weixin.mp.config.impl.WxMpRedisConfigImpl;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import javax.annotation.PostConstruct;
+
+@Configuration
+public class WechatMpConfigure {
+    private static String appidZhengzai;
+    private static String secretZhengzai;
+    private static String tokenZhengzai;
+    private static String aeskeyZhengzai;
+
+    private static String appidModernsky;
+    private static String secretModernsky;
+
+    @Value("${liquidnet.wechat.service.zhengzai.appid}")
+    public void setAppidZhengzai(String appidZhengzai) { WechatMpConfigure.appidZhengzai = appidZhengzai; }
+    @Value("${liquidnet.wechat.service.zhengzai.secret}")
+    public void setSecretZhengzai(String secretZhengzai) {
+        WechatMpConfigure.secretZhengzai = secretZhengzai;
+    }
+    @Value("${liquidnet.wechat.service.zhengzai.token}")
+    public void setTokenZhengzai(String tokenZhengzai) {
+        WechatMpConfigure.tokenZhengzai = tokenZhengzai;
+    }
+    @Value("${liquidnet.wechat.service.zhengzai.aeskey}")
+    public void setAeskeyZhengzai(String aeskeyZhengzai) {
+        WechatMpConfigure.aeskeyZhengzai = aeskeyZhengzai;
+    }
+
+    @Value("${liquidnet.wechat.service.modernsky.appid}")
+    public void setAppidModernsky(String appidModernsky) {
+        WechatMpConfigure.appidModernsky = appidModernsky;
+    }
+    @Value("${liquidnet.wechat.service.modernsky.secret}")
+    public void setSecretModernsky(String secretModernsky) {
+        WechatMpConfigure.secretModernsky = secretModernsky;
+    }
+
+    private WxMpService wxMpZhengzaiService;
+    private WxMpService wxMpModernskyService;
+
+    @Autowired
+    private StringRedisTemplate stringRedisTemplate;
+
+    @PostConstruct
+    public void init() {
+        wxMpZhengzaiService = new WxMpServiceImpl() {
+            {
+                RedisTemplateWxRedisOps redisTemplateWxRedisOps = new RedisTemplateWxRedisOps(stringRedisTemplate);
+                WxMpRedisConfigImpl wxMpDefaultConfig = new WxMpRedisConfigImpl(redisTemplateWxRedisOps, "sweet:accessToken:service:zhengzai");
+
+                wxMpDefaultConfig.setAppId(appidZhengzai);
+                wxMpDefaultConfig.setSecret(secretZhengzai);
+                wxMpDefaultConfig.setToken(tokenZhengzai);
+                wxMpDefaultConfig.setAesKey(aeskeyZhengzai);
+
+                this.setWxMpConfigStorage(wxMpDefaultConfig);
+            }
+        };
+        wxMpModernskyService = new WxMpServiceImpl() {
+            {
+                RedisTemplateWxRedisOps redisTemplateWxRedisOps = new RedisTemplateWxRedisOps(stringRedisTemplate);
+                WxMpRedisConfigImpl wxMpDefaultConfig = new WxMpRedisConfigImpl(redisTemplateWxRedisOps, "sweet:accessToken:service:modernsky");
+
+                wxMpDefaultConfig.setAppId(appidModernsky);
+                wxMpDefaultConfig.setSecret(secretModernsky);
+
+                this.setWxMpConfigStorage(wxMpDefaultConfig);
+            }
+        };
+    }
+
+    public WxMpService getWxMpService(Integer pnum) {
+        switch (pnum) {
+            case 1:
+                return wxMpZhengzaiService;
+            case 2:
+                return wxMpModernskyService;
+        }
+        return null;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetAnswerRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetAntigenicQuestionRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetCityVoteRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetIntegralActivityDrawRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetMaoDengReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetPosterRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetRemindRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetSqlLuckDrawRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetSqlUserRelationMDSKRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetSqlUserRelationRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetStoneIntegralReceiver.java

+package com.liquidnet.service.consumer.base.receiver.sweet;
+
+import com.liquidnet.commons.lang.util.CollectionUtil;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.ResponseDto;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.base.receiver.AbstractSqlRedisReceiver;
+import com.liquidnet.service.feign.stone.api.FeignStoneIntegralClient;
+import com.liquidnet.service.sweet.param.SweetStoneIntegralParam;
+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.connection.stream.StreamRecords;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+
+@Slf4j
+@Component
+public class ConsumerSweetStoneIntegralReceiver extends AbstractSqlRedisReceiver {
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Autowired
+    private FeignStoneIntegralClient feignStoneIntegralClient;
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.SweetQueue.SWEET_STONE_INTEGRAL.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.SweetQueue.SWEET_STONE_INTEGRAL.getGroup();
+    }
+
+    @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);
+        }
+    }
+
+    private boolean consumerMessageHandler(String msg) {
+        boolean aBoolean = false;
+        try {
+            SweetStoneIntegralParam param = JsonUtils.fromJson(msg, SweetStoneIntegralParam.class);
+            if (param == null) {
+                aBoolean = true;
+            } else {
+                ResponseDto<String> res = null;
+                if (null == param.getType()) {
+                    aBoolean = true;
+                    return true;
+                }
+                if (param.getType() == 1) {
+                    res = feignStoneIntegralClient.in2112(
+                            param.getUid(),
+                            param.getScore(),
+                            param.getContent()
+                    );
+                } else if (param.getType() == 2) {
+                    res = feignStoneIntegralClient.de2111(
+                            param.getUid(),
+                            param.getScore(),
+                            param.getContent()
+                    );
+                }
+                if (null != res && res.getCode().equals("0")) {
+                    aBoolean = true;
+                }
+            }
+        } catch (Exception e) {
+            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
+        } finally {
+            if (!aBoolean) {
+                HashMap<String, String> map = CollectionUtil.mapStringString();
+                map.put("message", msg);
+                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
+            }
+        }
+        return aBoolean;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetTemplateMsgReceiver.java

+package com.liquidnet.service.consumer.base.receiver.sweet;
+
+import com.liquidnet.commons.lang.util.CollectionUtil;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.base.receiver.AbstractSqlRedisReceiver;
+import com.liquidnet.service.consumer.base.service.impl.sweet.SweetWechatTemplateMsgServiceImpl;
+import com.liquidnet.service.sweet.dto.param.SweetOpenSendMsgParam;
+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.connection.stream.StreamRecords;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+
+@Slf4j
+@Component
+public class ConsumerSweetTemplateMsgReceiver extends AbstractSqlRedisReceiver {
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Autowired
+    SweetWechatTemplateMsgServiceImpl sweetWechatTemplateService;
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.SweetQueue.SWEET_TEMPLATE_MSG.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.SweetQueue.SWEET_TEMPLATE_MSG.getGroup();
+    }
+
+    @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);
+        }
+    }
+
+    private boolean consumerMessageHandler(String msg) {
+        boolean aBoolean = false;
+        try {
+            SweetOpenSendMsgParam sweetOpenSendMsgParam = JsonUtils.fromJson(msg, SweetOpenSendMsgParam.class);
+            aBoolean = null == sweetOpenSendMsgParam || sweetWechatTemplateService.openSendMsg(sweetOpenSendMsgParam);
+        } catch (Exception e) {
+            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
+        } finally {
+            if (!aBoolean) {
+                HashMap<String, String> map = CollectionUtil.mapStringString();
+                map.put("message", msg);
+                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
+            }
+        }
+        return aBoolean;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/sweet/ConsumerSweetUserRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/service/impl/sweet/SweetWechatTemplateMsgServiceImpl.java

+package com.liquidnet.service.consumer.base.service.impl.sweet;
+
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.liquidnet.service.consumer.base.config.sweet.WechatMaConfigure;
+import com.liquidnet.service.consumer.base.config.sweet.WechatMpConfigure;
+import com.liquidnet.service.sweet.dto.param.SweetOpenSendMsgParam;
+import com.liquidnet.service.sweet.entity.SweetAppletUsers;
+import com.liquidnet.service.sweet.mapper.SweetAppletUsersMapper;
+import com.liquidnet.service.sweet.vo.SweetWechatUsersVo;
+import lombok.extern.slf4j.Slf4j;
+import me.chanjar.weixin.common.error.WxErrorException;
+import me.chanjar.weixin.mp.api.WxMpService;
+import me.chanjar.weixin.mp.bean.template.WxMpTemplateData;
+import me.chanjar.weixin.mp.bean.template.WxMpTemplateMessage;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * <p>
+ * 公众号模版消息 服务实现类
+ * </p>
+ *
+ * @author liquidnet
+ * @since 2021-07-27
+ */
+@Slf4j
+@Service
+public class SweetWechatTemplateMsgServiceImpl {
+
+    @Autowired
+    WechatMpConfigure wechatMpConfigure;
+
+    @Autowired
+    WechatMaConfigure wechatMaConfigure;
+
+    @Autowired
+    private SweetAppletUsersMapper sweetAppletUserMapper;
+
+//    @Autowired
+//    private RedisDataUtils redisDataUtils;
+
+    public Boolean openSendMsg(SweetOpenSendMsgParam param) {
+        String openId = getOpenId(param.getPhone(), param.getSendTargetType());
+        if (openId.isEmpty()) {
+            return true;
+        }
+        WxMpTemplateMessage templateMessage = null;
+        if (param.getJumpType() == 5) {
+            templateMessage = getTemplateMessage(param.getTemplateId(), openId, param.getJumpUrl());
+        } else {
+            templateMessage = getTemplateMessage(param.getTemplateId(), openId, param.getJumpType(), param.getJumpAppletPagePath(), param.getJumpAppletUsePath());
+        }
+        // 添加模板数据
+        String templateData = param.getTemplateData();
+        JSONObject templateDataJObj = JSONObject.parseObject(templateData);
+        for (Map.Entry<String, Object> entry : templateDataJObj.entrySet()) {
+            templateMessage.addData(new WxMpTemplateData(entry.getKey(), String.valueOf(entry.getValue())));
+        }
+        String msgId = sendTmpMsg(templateMessage, param.getSendTargetType());
+        if (null == msgId) {
+            return false;
+        } else {
+            return true;
+        }
+    }
+
+    /**
+     * 获取templateMessage
+     *
+     * @param templateId
+     * @param openId
+     * @param url
+     * @return
+     */
+    private WxMpTemplateMessage getTemplateMessage(String templateId, String openId, String url) {
+        WxMpTemplateMessage templateMessage = WxMpTemplateMessage.builder()
+                .toUser(openId) // 接收者openid
+                .templateId(templateId) // 模板id
+                .url(url) // 模板跳转链接
+                .build();
+        return templateMessage;
+    }
+
+    private WxMpTemplateMessage getTemplateMessage(String templateId, String openId, Integer jumpType, String jumpAppletPagePath, Boolean jumpAppletUsePath) {
+        //配置小程序信息
+        WxMpTemplateMessage.MiniProgram miniProgram = new WxMpTemplateMessage.MiniProgram();
+        miniProgram.setAppid(wechatMaConfigure.getWxMaAppid(jumpType));//小程序appid
+        miniProgram.setUsePath(jumpAppletUsePath);//true的话需要前端配合处理跳转
+        miniProgram.setPagePath(jumpAppletPagePath);//用户点击时需要跳转的小程序页面
+
+        WxMpTemplateMessage templateMessage = WxMpTemplateMessage.builder()
+                .toUser(openId) // 接收者openid
+                .templateId(templateId) // 模板id
+                .miniProgram(miniProgram) // 模板跳转链接
+                .build();
+        return templateMessage;
+    }
+
+    private String getOpenId(String phone, Integer sendTargetType) { // sendTargetType 将来如果加入其他服务号用获取不同openId
+        List<SweetAppletUsers> sweetAppletUsers = sweetAppletUserMapper.selectList(Wrappers.lambdaQuery(SweetAppletUsers.class).eq(SweetAppletUsers::getPurePhoneNumber, phone));
+        if (CollectionUtils.isEmpty(sweetAppletUsers)) {
+            log.info("getOpenId没有此applet用户：[phone=[{}]", phone);
+            return "";
+        } else {
+            SweetAppletUsers sweetAppletUser = sweetAppletUsers.get(0);
+//            SweetWechatUsersVo sweetWechatUsers = redisDataUtils.getSweetWechatUsersOfUnionId(sweetAppletUser.getUnionId(), 1);
+//            if (null == sweetWechatUsers) {
+//                log.info("getOpenId没有此WechatUser用户：[phone={},unionId={}]", phone, sweetAppletUser.getUnionId());
+                return "";
+//            } else {
+//                return sweetWechatUsers.getOpenId();
+//            }
+        }
+    }
+
+    private String sendTmpMsg(WxMpTemplateMessage templateMessage, Integer sendTargetType) {
+        String msgId = null;
+        try {
+            // 发送模板消息
+            WxMpService wxMpService = wechatMpConfigure.getWxMpService(sendTargetType);
+            log.info("sendTmpMsg-isAccessTokenExpired：[{}] ", wxMpService.getWxMpConfigStorage().isAccessTokenExpired());
+            log.info("sendTmpMsg-getAccessToken：[{}] ", wxMpService.getWxMpConfigStorage().getAccessToken());
+            msgId = wxMpService.getTemplateMsgService().sendTemplateMsg(templateMessage);
+            log.info("模版消息发送成功：[msgId={}]", msgId);
+        } catch (WxErrorException e) {
+            log.error("模版消息发送异常", e);
+        }
+        return msgId;
+    }
+
+}