d

fi-taxctl/app/base/service/impl/rpt/rpttypeSve.js

@@ -33,10 +33,15 @@ class RptTypeService extends ServiceBase{
           hisarray.push(h);
           hisarray.push(h);
         });
         });
         await self.db.models.rpthistory.bulkCreate(hisarray,{transaction:t});
         await self.db.models.rpthistory.bulkCreate(hisarray,{transaction:t});
+        //每保存一次历史数据，就是处理完成一个任务中的一个报表
+        var  currentval= await self.redisClient.decr("batch"+batchid);
+        if(currentval==0){
               await self.redisClient.rpush("riskmq",JSON.stringify({
               await self.redisClient.rpush("riskmq",JSON.stringify({
                 key:new Date().toLocaleString(),
                 key:new Date().toLocaleString(),
                 content:batchid+""
                 content:batchid+""
           }));
           }));
+          self.redisClient.delete("batch"+batchid);
+        }
         return {};
         return {};
     });
     });
   }
   }
@@ -89,6 +94,8 @@ class RptTypeService extends ServiceBase{
           //保存报表
           //保存报表
         await this.db.models.uploadrpt.bulkCreate(rptresult,{transaction:t});
         await this.db.models.uploadrpt.bulkCreate(rptresult,{transaction:t});
         var lstspts=await this.db.models.uploadrpt.findAll({where:{uploadbatch_id:batchtemp.id},transaction:t});
         var lstspts=await this.db.models.uploadrpt.findAll({where:{uploadbatch_id:batchtemp.id},transaction:t});
+        //保存当前批的报表数量到缓存，每保存一次历史，计数减1，当计数变为0时，改变批状态，给康发送消息，
+        await this.redisClient.set("batch"+batchtemp.id,lstspts.length);
         try{
         try{
           lstspts.forEach(async (rpt)=>{
           lstspts.forEach(async (rpt)=>{
             await this.redisClient.notifyConsume("k8stask","docurls",rpt.id+"~"+batchtemp.id+"~"+rpt.rptdate+"~"+rpt.rpttypecode+"~"+rpt.urlpath);
             await this.redisClient.notifyConsume("k8stask","docurls",rpt.id+"~"+batchtemp.id+"~"+rpt.rptdate+"~"+rpt.rpttypecode+"~"+rpt.urlpath);

fi-taxctl/app/base/utils/redisClient.js

@@ -222,6 +222,10 @@ class RedisClient {
   async incr(key) {
   async incr(key) {
     return this.client.incrAsync(key);
     return this.client.incrAsync(key);
   }
   }
+  async decr(key) {
+    return this.client.decrAsync(key);
+  }
+
 }
 }
 module.exports = RedisClient;
 module.exports = RedisClient;
 var client=new RedisClient();
 var client=new RedisClient();