ip购买bug

src/main/java/com/edgec/browserbackend/browser/domain/IpTransaction.java

@@ -23,6 +23,10 @@ public class IpTransaction {
 
     private List<String> ipIds;
 
+    private boolean isLocked;
+
+    private long lockTimestamp;
+
     public String getUsername() {
         return username;
     }
@@ -62,4 +66,20 @@ public class IpTransaction {
     public void setIpIds(List<String> ipIds) {
         this.ipIds = ipIds;
     }
+
+    public void setLocked(boolean locked) {
+        isLocked = locked;
+    }
+
+    public boolean isLocked() {
+        return isLocked;
+    }
+
+    public long getLockTimestamp() {
+        return lockTimestamp;
+    }
+
+    public void setLockTimestamp(long lockTimestamp) {
+        this.lockTimestamp = lockTimestamp;
+    }
 }

src/main/java/com/edgec/browserbackend/browser/repository/IpTransactionRepository.java

@@ -3,11 +3,13 @@ package com.edgec.browserbackend.browser.repository;
 import com.edgec.browserbackend.browser.domain.IpTransaction;
 import org.springframework.data.mongodb.repository.MongoRepository;
 
+import java.util.List;
+
 /**
  * @Desc
  * @Author jason
  * @CreateTime 2020/3/18 5:21 下午
  **/
-public interface IpTransactionRepository extends MongoRepository<IpTransaction, String> {
-
+public interface IpTransactionRepository extends MongoRepository<IpTransaction, String>, IpTransactionRepositoryCustom {
+    List<IpTransaction> findByStatus(int status);
 }

src/main/java/com/edgec/browserbackend/browser/repository/IpTransactionRepositoryCustom.java

+package com.edgec.browserbackend.browser.repository;
+
+import com.edgec.browserbackend.browser.domain.IpResource;
+import com.edgec.browserbackend.browser.domain.IpTransaction;
+
+import java.util.List;
+
+public interface IpTransactionRepositoryCustom {
+
+    boolean lockTask(IpTransaction ipTransaction);
+
+    boolean unLockTask(String id);
+
+    List<IpTransaction> sampleTasks(int status, long timestamp);
+}

src/main/java/com/edgec/browserbackend/browser/repository/IpTransactionRepositoryCustomImpl.java

+package com.edgec.browserbackend.browser.repository;
+
+import com.edgec.browserbackend.browser.domain.IpResource;
+import com.edgec.browserbackend.browser.domain.IpTransaction;
+import com.mongodb.client.result.UpdateResult;
+import org.bson.Document;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.data.mongodb.core.aggregation.MatchOperation;
+import org.springframework.data.mongodb.core.aggregation.SampleOperation;
+import org.springframework.data.mongodb.core.query.BasicQuery;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Update;
+
+import java.time.Instant;
+import java.util.List;
+
+import static org.springframework.data.mongodb.core.query.Criteria.where;
+
+public class IpTransactionRepositoryCustomImpl implements IpTransactionRepositoryCustom {
+
+    @Autowired
+    MongoTemplate mongoTemplate;
+
+
+    @Override
+    public boolean lockTask(IpTransaction ipTransaction) {
+        Document doc = new Document();
+        BasicQuery basicQuery = new BasicQuery(doc);
+        Criteria criteria = new Criteria();
+        criteria.orOperator(where("tid").is(ipTransaction.getTid()).and("isLocked").is(false).and("status").is(ipTransaction.getStatus()),
+                where("lockTimestamp").lte(Instant.now().minusSeconds(300).toEpochMilli()).and("status").is(ipTransaction.getStatus()));
+        basicQuery.addCriteria(criteria);
+        Update update = new Update();
+        update.set("isLocked", true).set("lockTimestamp", Instant.now().toEpochMilli());
+        UpdateResult result = mongoTemplate.updateFirst(basicQuery, update, IpResource.class);
+
+        if (result.getModifiedCount() < 1)
+            return false;
+        else
+            return true;
+    }
+
+    @Override
+    public boolean unLockTask(String id) {
+        Document doc = new Document();
+        BasicQuery basicQuery = new BasicQuery(doc);
+        basicQuery.addCriteria(where("id").is(id));
+        Update update = new Update();
+        update.set("isLocked", false).set("lockTimestamp", Instant.now().toEpochMilli());
+        UpdateResult result = mongoTemplate.updateFirst(basicQuery, update, IpResource.class);
+
+        if (result.getModifiedCount() < 1)
+            return false;
+        else
+            return true;
+    }
+
+    @Override
+    public List<IpTransaction> sampleTasks(int status, long timestamp) {
+        Criteria matchCriteria  = new Criteria();
+        matchCriteria.orOperator(where("status").is(status).and("isLocked").is(false),
+                where("isLocked").is(true).and("lockTimestamp").lte(timestamp));
+
+        MatchOperation match = Aggregation.match(matchCriteria);
+
+        SampleOperation sample = Aggregation.sample(20);
+        AggregationResults<IpTransaction> results = mongoTemplate.aggregate(Aggregation.newAggregation(match, sample), IpTransaction.class, IpTransaction.class);
+        List<IpTransaction> mappedResults = results.getMappedResults();
+
+        return mappedResults;
+    }
+}

src/main/java/com/edgec/browserbackend/browser/task/BrowserTask.java

@@ -50,6 +50,9 @@ public class BrowserTask {
     @Autowired
     private AccountService accountService;
 
+    @Autowired
+    private IpTransactionRepository ipTransactionRepository;
+
     @Value("${spring.profiles.active}")
     private String profiles;
 
@@ -92,12 +95,12 @@ public class BrowserTask {
         String URL = (profiles.equals("dev") || profiles.equals("staging")) ? TESTURL : CLOUDAMURL;
         long time = Instant.now().minusSeconds(300).toEpochMilli();
         List<IpResource> ipResources = ipResourceRepository.sampleTasks(6, time);
-        AtomicBoolean result = new AtomicBoolean(false);
         for (IpResource ipResource : ipResources) {
             long start = System.currentTimeMillis();
             CompletableFuture.runAsync(() -> {
                 if (ipResourceRepository.lockTask(ipResource)) {
                     try {
+                        boolean result = false;
                         RestTemplate restTemplate = new RestTemplate();
                         HttpHeaders header = buildPostHeader();
                         HashMap<String, Object> map = new HashMap<>();
@@ -121,7 +124,6 @@ public class BrowserTask {
                             if (StringUtils.isNotBlank(ipBuyResultDto.getErrorCode())) {
                                 log.error("fail to buy ip");
                                 log.error(ipBuyResultDto.getErrorCode());
-                                result.set(false);
                             }
                             if (ipBuyResultDto != null && ipBuyResultDto.getIplist() != null && ipBuyResultDto.getIplist().size() >= 1) {
                                 AtomicInteger index = new AtomicInteger();
@@ -133,19 +135,16 @@ public class BrowserTask {
                                         ipResourceRepository.save(ipResource);
                                     } else {
                                         log.error("no ipResource");
-                                        result.set(false);
                                     }
                                     index.getAndIncrement();
                                 });
+                                result = true;
                             }
-//                        ipTransaction.setStatus(1);
-//                        ipTransactionRepository.save(ipTransaction);
-                            result.set(true);
                         } catch (Exception e) {
                             log.error(e.getMessage());
-                            result.set(false);
+                            result = false;
                         }
-                        if (result.get() == false && (ipResource.getPurchasedTime() < Instant.now().minusSeconds(7200).toEpochMilli())) {
+                        if (result == false && (ipResource.getPurchasedTime() < Instant.now().minusSeconds(7200).toEpochMilli())) {
                             IpChargeRequestDto ipChargeRequestDto = buildIpChargeRequestDto(ipResource, 3, 0);
                             accountService.chargeByMoney(ipResource.getUsername(), -ipResource.getPrice(), ipChargeRequestDto);
                             ipResourceRepository.deleteById(ipResource.getId());
@@ -207,4 +206,33 @@ public class BrowserTask {
         }
     }
 
+//    @Scheduled(cron = "0 0/1 * * * ?")
+//    public void queryIpTransaction() {
+//        String URL = (profiles.equals("dev") || profiles.equals("staging")) ? TESTURL : CLOUDAMURL;
+//        long time = Instant.now().minusSeconds(300).toEpochMilli();
+//        List<IpTransaction> ipTransactions = ipTransactionRepository.findByStatus(0);
+//        for (IpTransaction ipTransaction : ipTransactions) {
+//            long start = System.currentTimeMillis();
+//            CompletableFuture.runAsync(() -> {
+//                List<String> ipIds = ipTransaction.getIpIds();
+//                boolean result = true;
+//                for (String ipId : ipIds) {
+//                    IpResource ipResource = ipResourceRepository.findById(ipId).orElse(null);
+//                    if (ipResource == null)
+//                        continue;
+//                    if (ipResource.getStatus() == 3 || ipResource.getStatus() == 6) {
+//                        result = false;
+//                        break;
+//                    }
+//                }
+//                if (result == true) {
+//                    ipTransaction.setStatus(1);
+//                    ipTransactionRepository.save(ipTransaction);
+//                }
+//            }, ThreadPoolUtils.queryIpTransactionPool);
+//        }
+//    }
+
+
+
 }

src/main/java/com/edgec/browserbackend/common/utils/ThreadPoolUtils.java

@@ -38,9 +38,19 @@ public abstract class ThreadPoolUtils {
 
         @Override
         public Thread newThread(Runnable runnable) {
-            return new Thread(runnable, "browser-queyrIp-taks-" + count++);
+            return new Thread(runnable, "browser-queyrIp-task-" + count++);
         }
     });
 
 
+    private static final int IPTRANSACTION_POOL_COUNT = 10;
+
+    public static final ScheduledExecutorService queryIpTransactionPool = Executors.newScheduledThreadPool(IPTRANSACTION_POOL_COUNT, new ThreadFactory() {
+        int count = 1;
+
+        @Override
+        public Thread newThread(Runnable runnable) {
+            return new Thread(runnable, "browser-ipTransaction-task-" + count++);
+        }
+    });
 }