添加代理商奖励推送

src/main/java/com/lanren/huhu/partner/config/RedisConfiguration.java

@@ -2,17 +2,24 @@ package com.lanren.huhu.partner.config;
 
 import com.alibaba.fastjson.parser.ParserConfig;
 import com.alibaba.fastjson.support.spring.GenericFastJsonRedisSerializer;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.cache.RedisCacheConfiguration;
 import org.springframework.data.redis.cache.RedisCacheManager;
 import org.springframework.data.redis.cache.RedisCacheWriter;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisPassword;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.util.ObjectUtils;
 
 /**
  * @author houseme
@@ -60,4 +67,22 @@ public class RedisConfiguration {
         return redisTemplate;
     }
 
+    @Bean
+    public RedisTemplate redisTemplatePush(@Value("${spring.redis.database2}") int pushDb,
+                                           @Value("${spring.redis.host}") String host,
+                                           @Value("${spring.redis.port}") int port,
+                                           @Value("${spring.redis.password}") String password) {
+        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
+        configuration.setHostName(host);
+        configuration.setPort(port);
+        configuration.setDatabase(pushDb);
+        RedisPassword redisPassword = RedisPassword.of(password);
+        configuration.setPassword(redisPassword);
+        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
+        LettucePoolingClientConfiguration.LettucePoolingClientConfigurationBuilder builder = LettucePoolingClientConfiguration.builder();
+        builder.poolConfig(genericObjectPoolConfig);
+        LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(configuration, builder.build());
+        connectionFactory.afterPropertiesSet();
+        return redisTemplate(connectionFactory);
+    }
 }

src/main/java/com/lanren/huhu/partner/constants/Constants.java

@@ -68,6 +68,11 @@ public class Constants {
      */
     public static final BigDecimal MIN_REWARD = new BigDecimal("0.000001");
 
+    /**
+     * 代理商奖励push队列
+     */
+    public static final String AGENT_REWARD_PUSH_KEY = "agent:reward:queue";
+
     static {
         /**
          * 代理商平推奖励比例配置

src/main/java/com/lanren/huhu/partner/manager/AgentManager.java

@@ -18,8 +18,7 @@ import java.math.BigDecimal;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 
-import static com.lanren.huhu.partner.constants.Constants.PARENT_COLUMN_NAME_CENGJI;
-import static com.lanren.huhu.partner.constants.Constants.PARENT_COLUMN_NAME_PINGTUI;
+import static com.lanren.huhu.partner.constants.Constants.*;
 
 /**
  * @author chen
@@ -66,12 +65,14 @@ public class AgentManager {
         agentRewardMessage.setReferenceId("9999");
         agentRewardMessage.setAgentId(33);
         agentRewardMessage.setAgentReward(new BigDecimal(5));
+        agentRewardMessage.setSourceUserPayment(new BigDecimal(5));
         agentRewardMessage.setSettleState(-100);
         agentRewardMessage.setOrderType("taobao");
         agentRewardMessage.setOrderSn("507596802910534728");
         agentRewardMessage.setSubOrderSn("507596802910534728");
         agentRewardMessage.setOrderCommission(new BigDecimal("35.400000"));
-        ops.leftPush("agent_reward_redis_city_key", JSON.toJSONString(agentRewardMessage));
+        logger.info("写入队列");
+        ops.leftPush(AGENT_REWARD_QUEUE_KEY, JSON.toJSONString(agentRewardMessage));
         return new Result<>();
     }
     /**

src/main/java/com/lanren/huhu/partner/model/AgentRewardQueueMessage.java

+package com.lanren.huhu.partner.model;
+
+import com.lanren.huhu.partner.model.base.BaseModel;
+import lombok.Data;
+
+import java.math.BigDecimal;
+
+/**
+ * @author chen
+ * @title: AgentRewardQueueMessage
+ * @projectName partner
+ * @description: TODO
+ * @package com.lanren.huhu.partner.model
+ * @date 2019-07-09 16:05
+ */
+@Data
+public class AgentRewardQueueMessage extends BaseModel {
+    private int userId;
+    private int sourceUserId;
+    private String money;
+    private int partnerLevel;
+    private long time;
+    private int rewardType;
+}

src/main/java/com/lanren/huhu/partner/schedule/AgentRewardQueueTask.java

@@ -3,8 +3,10 @@ package com.lanren.huhu.partner.schedule;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.lanren.huhu.partner.domain.AgentReward;
+import com.lanren.huhu.partner.domain.PartnerAccount;
 import com.lanren.huhu.partner.domain.UserAgent;
 import com.lanren.huhu.partner.model.AgentRewardMessage;
+import com.lanren.huhu.partner.model.AgentRewardQueueMessage;
 import com.lanren.huhu.partner.model.ParentAgent;
 import com.lanren.huhu.partner.service.AgentRewardService;
 import com.lanren.huhu.partner.service.PartnerAccountService;
@@ -15,10 +17,16 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.ListOperations;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.annotation.Transactional;
+import sun.management.resources.agent;
 
 import java.math.BigDecimal;
 import java.time.LocalDateTime;
@@ -51,6 +59,12 @@ public class AgentRewardQueueTask {
     AgentRewardService agentRewardService;
     @Autowired
     PartnerAccountService partnerAccountService;
+    @Autowired
+    RedisTemplate redisTemplatePush;
+    @Autowired
+    DataSourceTransactionManager dataSourceTransactionManager;
+    @Autowired
+    TransactionDefinition transactionDefinition;
 
     @Scheduled(fixedDelay = 5000L)
     public void runScheduledTask() {
@@ -111,8 +125,6 @@ public class AgentRewardQueueTask {
          * 这里的逻辑有问题, 不能用这个接口找上级, 因为会先找一次直接邀请人,
          * 只能按message.getAgentId()的代理商id, 找user_agent表里的平推关系
          */
-//        User user = userService.getRelationByUserId(userAgent.getUserId());
-//        ArrayList<ParentAgent> parentList = user.getAgentList();
         ArrayList<ParentAgent> parentList = (ArrayList<ParentAgent>) userAgentService.getParentAgentListByAgentId(userAgent.getAgentId(), PARENT_COLUMN_NAME_PINGTUI);
         ArrayList<ParentAgent> level4List = new ArrayList<ParentAgent>();
         if (parentList.size() > 0) {
@@ -159,24 +171,57 @@ public class AgentRewardQueueTask {
                 rateList.add(rateMap.get(1));
                 rateList.add(rateMap.get(2));
             }
+
+            TransactionStatus transactionStatus = null;
+            List<AgentReward> pushList = new ArrayList<AgentReward>();
+            try {
+                /**
+                 * 开启事务, 如果rewardList中的奖励 都插入成功, 才提交事务
+                 * 并发送redis消息, 否则, 进行回滚, 并把读入的消息塞回队列
+                 */
+                transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
                 for (int i=0; i<rewardList.size(); i++) {
                     BigDecimal commissionRate = rateList.get(i);
-                if (doInsert(commissionRate, rewardList.get(i), message)) {
-                    logger.info("插入代理商分成奖励,奖励比例: {}, 代理商: {}", commissionRate, rewardList.get(i));
-                } else {
-                    logger.error("插入失败, 写回到队列");
+                    AgentReward reward = doInsert(commissionRate, rewardList.get(i), message);
+                    if (reward != null) {
+                        pushList.add(reward);
+                    }
+                }
+                /**
+                 * 提交事务
+                 */
+                dataSourceTransactionManager.commit(transactionStatus);
+                /**
+                 * 数据库写入成功后 发推送
+                 */
+                for (AgentReward reward : pushList) {
+                    doPush(reward);
+                }
+            } catch (Exception e) {
+                logger.error("插入代理商分成奖励失败, 奖励 {}", message);
+                /**
+                 * 回滚事务
+                 */
+                if (transactionStatus != null) {
+                    dataSourceTransactionManager.rollback(transactionStatus);
+                }
+                /**
+                 * 消息处理失败 丢回队列
+                 */
                 ops.leftPush(AGENT_REWARD_QUEUE_KEY, JSON.toJSONString(message));
+                /**
+                 * 休眠
+                 */
                 try {
                     Thread.sleep(5 * 1000L);
-                    } catch (InterruptedException e) {
-                        logger.error(e.getMessage(), e);
-                    }
+                } catch (Exception e1) {
+                    logger.error(e.getMessage(), e1);
                 }
             }
         }
     }
 
-    private boolean doInsert(BigDecimal commissionRate, ParentAgent agent, AgentRewardMessage message) {
+    private AgentReward doInsert(BigDecimal commissionRate, ParentAgent agent, AgentRewardMessage message) {
         try {
             int rewardType = message.getRewardType();
             int sourceUserLevel = partnerAccountService.getOneByUserId(message.getSourceUserId()).getPartnerLevel();
@@ -219,24 +264,45 @@ public class AgentRewardQueueTask {
                 agentReward.setAllMoneyOri(message.getOrderCommission());
                 agentReward.setRatioAll(message.getOrderRateArray());
             }
+            logger.info("rewardBasement is: {}", rewardBasement);
+            logger.info("commissionRate is: {}", commissionRate);
             BigDecimal reward = rewardBasement.multiply(commissionRate);
             /**
              * 小于0.000001的奖励 不写入
              */
             if (reward.compareTo(MIN_REWARD) < 0) {
                 logger.info("代理商({})平推奖励金额小于0.000001, 忽略不写入. msg:{}", agent.getAgentId(), message);
-                return true;
+                return null;
             } else {
                 agentReward.setMoney(rewardBasement);
                 agentReward.setCommissionAcount(rewardBasement);
                 agentReward.setAmount(reward);
                 agentReward.setCommission(reward);
                 agentReward.setCashCode("0");
-                return agentRewardService.save(agentReward);
+                doPush(agentReward);
+                agentRewardService.save(agentReward);
+                return agentReward;
+            }
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
+            return null;
         }
+    }
+
+    private void doPush(AgentReward agentReward) {
+        try {
+            JSONObject json = new JSONObject();
+            PartnerAccount partnerAccount = partnerAccountService.getOneByUserId(agentReward.getUserId());
+            AgentRewardQueueMessage message = new AgentRewardQueueMessage();
+            message.setUserId(agentReward.getUserId());
+            message.setSourceUserId(agentReward.getSourceUserId());
+            message.setMoney(agentReward.getCommission().toString());
+            message.setPartnerLevel(partnerAccount.getIsSuperPartner() == 1 ? 30 : partnerAccount.getPartnerLevel());
+            message.setTime(agentReward.getCreatedAt());
+            message.setRewardType(agentReward.getRewardType());
+            redisTemplatePush.opsForList().leftPush(AGENT_REWARD_PUSH_KEY, message);
         } catch (Exception e) {
             logger.error(e.getMessage(), e);
-            return false;
         }
     }
 }

src/main/resources/application.yml

@@ -54,6 +54,7 @@ spring:
   ## Redis数据库索引（默认为0）
   redis:
     database: 13
+    database2: 5
     ## Redis测试服务器地址
     host: @redis.host@
     ## Redis测试服务器连接端口