ufo / ufo-platform · Commits

Go to a project

Authored by LUOXC
Commit 6c403757b5f7d45ac55053b582bd28174a092d48 1 parent 5b16cf26

fail tradebills recovery

Inline Side-by-side
Showing 7 changed files with 165 additions and 124 deletions
... ... @@ -17,6 +17,8 @@ public class UserHelper {
public static final UserHelper CUSTOMS_SYSTEM = new UserHelper(0,"海关系统");
public static final UserHelper UFO_PLATFORM_SYSTEM = new UserHelper(1,"UFO平台端系统");
private int userId;
private String userName;
... ...
package com.yoho.order.dal;
import com.yoho.order.model.TradeBillsReq;
import com.yoho.order.model.TradeBills;
import com.yoho.order.model.TradeBillsReq;
import com.yoho.order.model.TradeBillsTransferOutTradeNo;
import org.apache.ibatis.annotations.Param;
... ... @@ -12,11 +12,13 @@ import java.util.List;
*/
public interface TradeBillsMapper {
int selectCountByCondition(@Param("billsTradeReq") TradeBillsReq req);
int selectCountByCondition(@Param("billsTradeReq") TradeBillsReq req);
List<TradeBills> selectByConditionWithPage(@Param("billsTradeReq") TradeBillsReq req);
List<TradeBills> selectByConditionWithPage(@Param("billsTradeReq") TradeBillsReq req);
List<TradeBillsTransferOutTradeNo> selectTransferOutTradeNoByOrderCodes(@Param("orderCodes") List<Long> orderCodes);
List<TradeBillsTransferOutTradeNo> selectTransferOutTradeNoByOrderCodes(@Param("orderCodes") List<Long> orderCodes);
List<TradeBills> selectByFailTradeBills(@Param("fromCreateTime") int fromCreateTime, @Param("fromTradeBillsId") int fromTradeBillsId, @Param("limit") int limit);
}
... ...
... ... @@ -105,4 +105,18 @@
and t.interface_type = 2
</select>
<select id="selectByFailTradeBills">
select <include refid="Base_Column_List" />
from trade_bills where
<!-- 失败的 -->
income_outcome = 1 AND deal_status != 1
<!-- 没有人工干预的 -->
AND (deal_uid is NULL or deal_uid in (0,1))
<!-- 退款失败和转账失败 -->
AND trade_status in (200,299)
AND create_time > #{fromCreateTime}
AND id > #{fromTradeBillsId}
limit #{limit}
</select>
</mapper>
\ No newline at end of file
... ...
package com.yoho.ufo.order.monitor;
import com.yoho.core.redis.cluster.annotation.Redis;
import com.yoho.core.redis.cluster.operations.nosync.YHHashOperations;
import com.yoho.core.redis.cluster.operations.nosync.YHValueOperations;
import com.yoho.core.redis.cluster.operations.serializer.RedisKeyBuilder;
import com.yoho.order.dal.OrderStatusMonitorMapper;
import com.yoho.quartz.annotation.JobType;
import com.yoho.quartz.annotation.MisfiredPolicy;
import com.yoho.quartz.annotation.YhJobDef;
import com.yoho.quartz.domain.JobProcessResult;
import com.yoho.quartz.domain.JobResultCode;
import com.yoho.quartz.job.YhJob;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.TimeUnit;
/**
* @author LUOXC
* @date 2019/1/7 11:02
*/
@Component
@Slf4j
@YhJobDef(desc = "卖家未支付订单超时取消监控",
jobName = "BuyerOrderWithoutPaidTimeoutCancelMonitor",
cron = "0 0/5 * * * ?",
misfiredPolicy = MisfiredPolicy.CRON_DO_NOTHING,
jobType = JobType.CRON,
jobGroup = "ufoPlatform",
needUpdate = true)
public class BuyerOrderWithoutPaidTimeoutCancelMonitor implements YhJob {
@Redis("gwNoSyncRedis")
private YHValueOperations yhValueOperations;
@Redis("gwNoSyncRedis")
private YHHashOperations hashOperations;
@Autowired
private OrderStatusMonitorMapper orderStatusMonitorMapper;
private RedisKeyBuilder lastMonitorEndTimeKey = RedisKeyBuilder.newInstance()
.appendFixed("ufo:platform:order:monitor:buyerOrderWithoutPaidTimeoutCancelLastMonitorEndTime");
private RedisKeyBuilder alertKey = RedisKeyBuilder.newInstance()
.appendFixed("ufo:platform:order:monitor:buyerOrderWithoutPaidTimeoutCancelAlert");
@Override
public JobProcessResult process(String s) {
monitor();
JobProcessResult result = new JobProcessResult();
result.setJobResultCode(JobResultCode.SUCCESS);
return result;
}
private void monitor() {
Mono.zip(getLastMonitorEndTime(), getThisMonitorEndTime())
.doOnNext(this::cacheThisMonitorEndTime)
.flatMapIterable(monitorTime ->
orderStatusMonitorMapper.selectBuyerOrderCodeForBuyerOrderWithoutPaidTimeoutCancelFail(monitorTime.getT1(), monitorTime.getT2())
)
.filter(this::cacheIfNotAlert)
.map(orderCode -> Tuples.of("卖家未支付订单超时取消异常", "订单号" + orderCode))
.subscribe(
message -> alert(message),
e -> log.warn("Buyer order without paid timeout cancel monitor fail", e),
() -> log.info("Buyer order without paid timeout cancel monitor complete")
);
}
private Mono<Long> getLastMonitorEndTime() {
return Mono.fromSupplier(() -> yhValueOperations.get(lastMonitorEndTimeKey))
.doOnError(e -> log.warn("get last monitor end time fail", e))
.onErrorReturn(StringUtils.EMPTY)
.map(NumberUtils::toLong);
}
private Mono<Long> getThisMonitorEndTime() {
return Mono.just(Instant.now())
.map(now -> now.minus(20, ChronoUnit.MINUTES).toEpochMilli());
}
private void cacheThisMonitorEndTime(Tuple2<Long, Long> monitorTime) {
try {
yhValueOperations.set(lastMonitorEndTimeKey, String.valueOf(monitorTime.getT1()), 20, TimeUnit.MINUTES);
} catch (Exception e) {
log.warn("cache this monitor end time fail", e);
}
}
private Boolean cacheIfNotAlert(Long orderCode) {
try {
return hashOperations.putIfAbsent(
alertKey,
orderCode.toString(),
LocalDateTime.now().toString());
} catch (Exception e) {
log.warn("cache if not alert fail for {}", orderCode, e);
return true;
}
}
private void alert(Tuple2<String, String> message) {
val title = message.getT1();
val body = message.getT2();
log.warn("Buyer order without paid timeout cancel fail, title:{},body:", title, body);
}
}
package com.yoho.ufo.order.task;
import com.alibaba.fastjson.JSONObject;
import com.yoho.core.rest.client.ServiceCaller;
import com.yoho.order.model.TradeBills;
import com.yoho.ufo.service.impl.UserHelper;
import com.yohobuy.ufo.model.order.req.ManualDealRequest;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class FailTradeBillsRecovery {
private final ServiceCaller serviceCaller;
private final TradeBills failTradeBills;
public FailTradeBillsRecovery(ServiceCaller serviceCaller, TradeBills failTradeBills) {
this.serviceCaller = serviceCaller;
this.failTradeBills = failTradeBills;
}
public void recover() {
if (failTradeBills.getTradeStatus().intValue() == 100) {
return;
}
ManualDealRequest req = new ManualDealRequest();
req.setOperateUid(UserHelper.UFO_PLATFORM_SYSTEM.getUserId());
req.setOperateUname(UserHelper.UFO_PLATFORM_SYSTEM.getUserName());
req.setTradeBillsId(failTradeBills.getId());
req.setUid(failTradeBills.getUid());
req.setOrderCode(failTradeBills.getOrderCode());
req.setAmount(failTradeBills.getAmount().toString());
log.info("fail trade bills recovery, param is {}", JSONObject.toJSONString(req));
JSONObject jsonObject = serviceCaller.asyncCall("ufo-gateway.manualDealMon", req, JSONObject.class).get(5);
log.info("fail trade bills recovery, result is {}", jsonObject.toJSONString());
}
}
\ No newline at end of file
... ...
package com.yoho.ufo.order.task;
import com.yoho.core.rest.client.ServiceCaller;
import com.yoho.order.dal.TradeBillsMapper;
import com.yoho.order.model.TradeBills;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@Slf4j
public class FailTradeBillsScanner implements Runnable {
private final ScheduledExecutorService scheduler;
private final int pollingInterval;
private final TradeBillsMapper tradeBillsMapper;
private final ServiceCaller serviceCaller;
private final Integer fromCreateTime;
private Integer fromTradeBillsId = 0;
public FailTradeBillsScanner(ScheduledExecutorService scheduler, int pollingInterval, TradeBillsMapper tradeBillsMapper, ServiceCaller serviceCaller, Integer fromCreateTime) {
this.scheduler = scheduler;
this.pollingInterval = pollingInterval;
this.tradeBillsMapper = tradeBillsMapper;
this.serviceCaller = serviceCaller;
this.fromCreateTime = fromCreateTime;
}
@Override
public void run() {
try {
// Need to catch the exception to keep the event scanner running.
pollEvents();
} catch (Exception ex) {
log.warn("Got the exception {} when poll.", ex.getMessage(), ex);
}
}
private void pollEvents() {
scheduler.scheduleWithFixedDelay(
() -> recoverFailTradeBills(),
0,
pollingInterval,
MILLISECONDS);
}
private void recoverFailTradeBills() {
List<TradeBills> failTradeBillsList = tradeBillsMapper.selectByFailTradeBills(fromCreateTime, fromTradeBillsId, 1);
if (CollectionUtils.isEmpty(failTradeBillsList)) {
log.info("Non found fail trade bills, reset from id.");
fromTradeBillsId = 0;
} else {
fromTradeBillsId = failTradeBillsList.stream().map(TradeBills::getId).max(Integer::compareTo).get();
log.info("Found fail trade bills, the last id is {}", fromTradeBillsId);
}
failTradeBillsList.stream().forEach(failTradeBills -> {
new FailTradeBillsRecovery(serviceCaller, failTradeBills).recover();
});
}
}
\ No newline at end of file
... ...
package com.yoho.ufo.order.task;
import com.yoho.core.rest.client.ServiceCaller;
import com.yoho.order.dal.TradeBillsMapper;
import com.yoho.ufo.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@Slf4j
@Configuration
public class TaskConfig {
@Value("${failTradeBills.scanner.enabled:true}")
boolean failTradeBillsScannerEnabled;
@Value("${failTradeBills.scanner.pollingInterval:500}")
int failTradeBillsScannerPollingInterval;
@Autowired
TradeBillsMapper tradeBillsMapper;
@Autowired
ServiceCaller serviceCaller;
@PostConstruct
void init() {
if (failTradeBillsScannerEnabled) {
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
Integer fromCreateTime = DateUtil.getTimeSecondsFromStr("20191220", "yyyyMMdd");
new FailTradeBillsScanner(scheduler,
failTradeBillsScannerPollingInterval,
tradeBillsMapper,
serviceCaller,
fromCreateTime).run();
log.info("Starting the FailTradeBillsScanner.");
}
}
}
\ No newline at end of file
... ...