集成ES

src/main/java/com/sqx/modules/app/entity/UserMoneyDetails.java

@@ -20,7 +20,7 @@ import java.math.BigDecimal;
 @Data
 @AllArgsConstructor
 @TableName("user_money_details")
-@IndexName("user_money_details")
+@IndexName(value = "user_money_details", keepGlobalPrefix = true)
 @ApiModel("钱包详情")
 public class UserMoneyDetails implements Serializable {
     /**

src/main/java/com/sqx/modules/app/service/impl/UserServiceImpl.java

@@ -1,6 +1,7 @@
 package com.sqx.modules.app.service.impl;
 
 
+import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.date.DateTime;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.thread.ThreadUtil;
@@ -17,10 +18,8 @@ import com.aliyuncs.DefaultAcsClient;
 import com.aliyuncs.IAcsClient;
 import com.aliyuncs.http.MethodType;
 import com.aliyuncs.profile.DefaultProfile;
-import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
-import com.baomidou.mybatisplus.core.toolkit.IdWorker;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.getui.push.v2.sdk.ApiHelper;
 import com.getui.push.v2.sdk.GtApiConfiguration;
@@ -50,11 +49,15 @@ import com.sqx.modules.app.utils.JwtUtils;
 import com.sqx.modules.app.utils.UserConstantInterface;
 import com.sqx.modules.common.entity.CommonInfo;
 import com.sqx.modules.common.service.CommonInfoService;
+import com.sqx.modules.course.dao.CourseDao;
+import com.sqx.modules.course.entity.Course;
 import com.sqx.modules.discSpinning.service.DiscSpinningService;
+import com.sqx.modules.es.mapper.EsOrdersMapper;
 import com.sqx.modules.file.utils.Md5Utils;
 import com.sqx.modules.invite.service.InviteService;
 import com.sqx.modules.message.entity.MessageInfo;
 import com.sqx.modules.message.service.MessageService;
+import com.sqx.modules.orders.entity.Orders;
 import com.sqx.modules.sys.entity.SysUserEntity;
 import com.sqx.modules.sys.service.SysUserService;
 import com.sqx.modules.utils.HttpClientUtil;
@@ -71,6 +74,7 @@ import com.tencentcloudapi.sms.v20210111.models.SendSmsResponse;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.lang.StringUtils;
+import org.dromara.easyes.core.conditions.select.LambdaEsQueryWrapper;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Lazy;
@@ -79,14 +83,14 @@ import org.springframework.transaction.annotation.Transactional;
 import weixin.popular.api.SnsAPI;
 import weixin.popular.util.JsonUtil;
 
+import javax.annotation.Resource;
 import javax.servlet.http.HttpServletRequest;
 import java.math.BigDecimal;
 import java.text.SimpleDateFormat;
-import java.util.Date;
+import java.util.*;
-import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
-import java.util.List;
-import java.util.Map;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
 
 /**
  * 用户
@@ -124,6 +128,10 @@ public class UserServiceImpl extends ServiceImpl<UserDao, UserEntity> implements
     private SysUserService sysUserService;
     private final AliService aliService;
     private final UserInfoService userInfoService;
+    @Resource
+    private CourseDao courseDao;
+    @Resource
+    private EsOrdersMapper esOrdersMapper;
 
     @Value("${spring.profiles.active}")
     private String profiles;
@@ -1410,9 +1418,41 @@ public class UserServiceImpl extends ServiceImpl<UserDao, UserEntity> implements
             endTime = DateUtil.format(DateUtil.endOfYear(parse), "yyyy-MM-dd 23:59:59");
         }
 
-        PageHelper.startPage(page.intValue(), limit.intValue());
+        List<Map<String, Object>> list = new ArrayList<>();
-        List<Map<String, Object>> list = baseMapper.queryCourseOrder(type, startTime, endTime, sysUserId);
+        // 改为从es库查询数据
-        PageInfo<Map<String, Object>> pageInfo = new PageInfo<>(list);
+        LambdaEsQueryWrapper<Orders> wrapper = new LambdaEsQueryWrapper<>();
+        if (sysUserId != null) {
+            wrapper.eq(Orders::getSysUserId, sysUserId);
+        }
+        wrapper.ge(Orders::getCreateTime, startTime);
+        wrapper.le(Orders::getCreateTime, endTime);
+        wrapper.eq(Orders::getStatus, 1);
+        wrapper.eq(Orders::getOrdersType, 1);
+        List<Orders> orders = esOrdersMapper.selectList(wrapper);
+        if(CollUtil.isEmpty(orders)){
+            PageInfo<Map<String, Object>> pageInfo = new PageInfo<>(new ArrayList<>());
+            return PageUtils.page(pageInfo, true);
+        }
+        Map<Long, List<Orders>> collect = orders.stream().collect(Collectors.groupingBy(Orders::getCourseId));
+        List<Course> courses = courseDao.selectList(Wrappers.<Course>lambdaQuery().select(Course::getCourseId, Course::getTitle).in(Course::getCourseId, collect.keySet()));
+        Map<Long, String> courseMap = courses.stream().collect(Collectors.toMap(Course::getCourseId, Course::getTitle));
+        collect.forEach((k, v) -> {
+            Long courseId = k;
+            String courseName = courseMap.get(k);
+            BigDecimal courseMoney = v.stream().map(Orders::getPayMoney).reduce(BigDecimal.ZERO, BigDecimal::add);
+            int courseNum = v.size();
+            Map<String, Object> data = new ConcurrentHashMap();
+            data.put("courseId", courseId);
+            data.put("coursemoney", courseMoney);
+            data.put("coursenum", courseNum);
+            data.put("coursename", courseName);
+            list.add(data);
+        });
+        List<List<Map<String, Object>>> split = CollUtil.split(list, limit.intValue());
+        List<Map<String, Object>> splitList = split.get(page.intValue() - 1);
+        // PageHelper.startPage(page.intValue(), limit.intValue());
+        // List<Map<String, Object>> list = baseMapper.queryCourseOrder(type, startTime, endTime, sysUserId);
+        PageInfo<Map<String, Object>> pageInfo = new PageInfo<>(splitList);
 
         return PageUtils.page(pageInfo, true);
     }
@@ -1495,8 +1535,8 @@ public class UserServiceImpl extends ServiceImpl<UserDao, UserEntity> implements
 
     @Override
     public Result selectInviteUserList(Integer page, Integer limit, String userName, String phone) {
-        PageHelper.startPage(page,limit);
+        PageHelper.startPage(page, limit);
-        return Result.success().put("data", PageUtils.page(new PageInfo<>(baseMapper.selectInviteUserList(userName, phone)),true));
+        return Result.success().put("data", PageUtils.page(new PageInfo<>(baseMapper.selectInviteUserList(userName, phone)), true));
     }
 
     @Override

src/main/java/com/sqx/modules/course/entity/CourseCollect.java

@@ -18,7 +18,7 @@ import java.io.Serializable;
 @Data
 @Accessors(chain = true)
 @TableName("course_collect")
-@IndexName("course_collect")
+@IndexName(value = "course_collect", keepGlobalPrefix = true)
 public class CourseCollect implements Serializable {
 
     private static final long serialVersionUID = 1L;

src/main/java/com/sqx/modules/course/entity/CourseUser.java

@@ -16,7 +16,7 @@ import java.io.Serializable;
  */
 @Data
 @TableName("course_user")
-@IndexName("course_user")
+@IndexName(value = "course_user", keepGlobalPrefix = true)
 public class CourseUser implements Serializable {
 
     private static final long serialVersionUID = 1L;

src/main/java/com/sqx/modules/discSpinning/entity/DiscSpinningRecord.java

@@ -18,7 +18,7 @@ import java.math.BigDecimal;
  */
 @Data
 @TableName("disc_spinning_record")
-@IndexName("disc_spinning_record")
+@IndexName(value = "disc_spinning_record", keepGlobalPrefix = true)
 public class DiscSpinningRecord extends Model<DiscSpinningRecord> {
     //主键
     @TableId(type = IdType.ID_WORKER)

src/main/java/com/sqx/modules/es/mapper/EsTaskCenterRecordMapper.java

@@ -0,0 +1,11 @@
+package com.sqx.modules.es.mapper;
+
+import com.sqx.modules.taskCenter.entity.TaskCenterRecord;
+import org.dromara.easyes.core.kernel.BaseEsMapper;
+
+/**
+ * @author tankaikai
+ * @since 2025-01-08 16:31
+ */
+public interface EsTaskCenterRecordMapper extends BaseEsMapper<TaskCenterRecord> {
+}

src/main/java/com/sqx/modules/es/service/EsOrderService.java

@@ -0,0 +1,22 @@
+package com.sqx.modules.es.service;
+
+import com.sqx.modules.orders.entity.Orders;
+
+import java.io.Serializable;
+
+/**
+ * @author tankaikai
+ * @since 2025-01-06 19:30
+ */
+public interface EsOrderService {
+    Orders selectById(Serializable id);
+
+    Orders selectOrdersByCourseIdAndUserId(Long userId, Long courseId);
+
+    Orders selectOrderByOrdersNo(String ordersNo);
+
+    Orders selectOrderByTradeNo(String tradeNo);
+
+    Orders selectOrderByOrderIdAndUserId(String orderId, Long userId);
+
+}

src/main/java/com/sqx/modules/es/service/impl/EsCoreServiceImpl.java

@@ -1,13 +1,9 @@
 package com.sqx.modules.es.service.impl;
 
-import cn.hutool.core.lang.Dict;
+import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.util.PageUtil;
-import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.sqx.modules.app.dao.UserMoneyDetailsDao;
-import com.sqx.modules.app.entity.UserEntity;
 import com.sqx.modules.app.entity.UserMoneyDetails;
-import com.sqx.modules.app.service.UserService;
 import com.sqx.modules.course.dao.CourseCollectDao;
 import com.sqx.modules.course.dao.CourseUserDao;
 import com.sqx.modules.course.entity.CourseCollect;
@@ -30,6 +26,7 @@ import org.springframework.stereotype.Service;
 import javax.annotation.Resource;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * @author tankaikai
@@ -76,151 +73,134 @@ public class EsCoreServiceImpl implements EsCoreService {
     private EsUserMoneyDetailsMapper esUserMoneyDetailsMapper;
     @Resource
     private EsTaskCenterRecordMapper esTaskCenterRecordMapper;
-    @Resource
-    private UserService userService;
 
     @Override
     public void sync() {
-        syncUserMoneyDetails();
+        List<String> list = new ArrayList<>();
-        syncOrders();
+        for (int i = 0; i < 10; i++) {
-        syncCourseCollect();
+            list.add(i + "");
-        syncPayDetails();
-        syncDiscSpinningRecord();
-        syncCashOut();
-        syncCourseUser();
-        syncTaskCenterRecord();
-    }
-
-    private List<Dict> buildPageList(int total) {
-        int size = 300;
-        int totalPage = PageUtil.totalPage(total, size);
-        List<Dict> pageList = new ArrayList<>();
-        for (int page = 1; page <= totalPage; page++) {
-            Dict dict = new Dict();
-            dict.set("page", (page - 1) * size);
-            dict.set("size", size);
-            pageList.add(dict);
         }
-        return pageList;
+        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
+            // 执行方法1的代码
+            System.out.println("方法1正在执行...");
+            syncUserMoneyDetails(list);
+        });
+        CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
+            // 执行方法2的代码
+            System.out.println("方法2正在执行...");
+            syncOrders(list);
+        });
+        CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> {
+            // 执行方法3的代码
+            System.out.println("方法3正在执行...");
+            syncPayDetails(list);
+        });
+        CompletableFuture<Void> future4 = CompletableFuture.runAsync(() -> {
+            // 执行方法4的代码
+            System.out.println("方法4正在执行...");
+            syncDiscSpinningRecord(list);
+        });
+        CompletableFuture<Void> future5 = CompletableFuture.runAsync(() -> {
+            // 执行方法6的代码
+            System.out.println("方法5正在执行...");
+            syncCashOut(list);
+        });
+        CompletableFuture<Void> future6 = CompletableFuture.runAsync(() -> {
+            // 执行方法6的代码
+            System.out.println("方法6正在执行...");
+            syncCourseCollect(list);
+        });
+        CompletableFuture<Void> future7 = CompletableFuture.runAsync(() -> {
+            // 执行方法7的代码
+            System.out.println("方法7正在执行...");
+            syncCourseUser(list);
+        });
+        CompletableFuture<Void> future8 = CompletableFuture.runAsync(() -> {
+            // 执行方法8的代码
+            System.out.println("方法8正在执行...");
+            syncTaskCenterRecord(list);
+        });
+        // 等待多个任务完成
+        CompletableFuture.allOf(future1, future2, future3, future4, future5, future6, future7).join();
+        System.out.println("所有方法已并行执行完毕。");
+        /*
+        syncUserMoneyDetails(list);
+        syncOrders(list);
+        syncCourseCollect(list);
+        syncPayDetails(list);
+        syncDiscSpinningRecord(list);
+        syncCashOut(list);
+        syncCourseUser(list);
+        syncTaskCenterRecord(list);
+
+        System.out.println("所有方法已并行执行完毕。");
+         */
     }
 
-    private void syncOrders() {
+    private void syncOrders(List<String> list) {
-        int total = userService.count(Wrappers.emptyWrapper());
+        list.parallelStream().forEach(i -> {
-        List<Dict> pageList = buildPageList(total);
+            System.out.println("同步订单数据：" + i);
-        pageList.parallelStream().forEach(param -> {
+            List<Orders> batchList = ordersDao.selectList(Wrappers.<Orders>lambdaQuery().likeLeft(Orders::getOrdersId, i));
-            Integer p = param.getInt("page");
+            esOrdersMapper.insertBatch(batchList);
-            Integer s = param.getInt("size");
-            List<UserEntity> list = userService.list(Wrappers.<UserEntity>lambdaQuery().orderByAsc(UserEntity::getUserId).last(StrUtil.format("limit {},{}", p, s)));
-            list.parallelStream().forEach(user -> {
-                System.out.println("同步订单数据：" + user.getUserId());
-                List<Orders> batchList = ordersDao.selectList(Wrappers.<Orders>lambdaQuery().eq(Orders::getUserId, user.getUserId()));
-                esOrdersMapper.insertBatch(batchList);
-            });
         });
     }
 
-    private void syncCashOut() {
+    private void syncCashOut(List<String> list) {
-        int total = userService.count(Wrappers.emptyWrapper());
+        list.parallelStream().forEach(i -> {
-        List<Dict> pageList = buildPageList(total);
+            System.out.println("同步提现数据：" + i);
-        pageList.parallelStream().forEach(param -> {
+            List<CashOut> batchList = cashOutDao.selectList(Wrappers.<CashOut>lambdaQuery().likeLeft(CashOut::getId, i));
-            Integer p = param.getInt("page");
+            if (CollUtil.isNotEmpty(batchList)) {
-            Integer s = param.getInt("size");
-            List<UserEntity> list = userService.list(Wrappers.<UserEntity>lambdaQuery().orderByAsc(UserEntity::getUserId).last(StrUtil.format("limit {},{}", p, s)));
-            list.parallelStream().forEach(user -> {
-                System.out.println("同步提现数据：" + user.getUserId());
-                List<CashOut> batchList = cashOutDao.selectList(Wrappers.<CashOut>lambdaQuery().eq(CashOut::getUserId, user.getUserId()));
                 esCashOutMapper.insertBatch(batchList);
-            });
+            }
         });
     }
 
-    private void syncCourseCollect() {
+
-        int total = userService.count(Wrappers.emptyWrapper());
+    private void syncCourseCollect(List<String> list) {
-        List<Dict> pageList = buildPageList(total);
+        list.parallelStream().forEach(i -> {
-        pageList.parallelStream().forEach(param -> {
+            System.out.println("同步短剧收藏数据：" + i);
-            Integer p = param.getInt("page");
+            List<CourseCollect> batchList = courseCollectDao.selectList(Wrappers.<CourseCollect>lambdaQuery().likeLeft(CourseCollect::getCourseCollectId, i));
-            Integer s = param.getInt("size");
+            esCourseCollectMapper.insertBatch(batchList);
-            List<UserEntity> list = userService.list(Wrappers.<UserEntity>lambdaQuery().orderByAsc(UserEntity::getUserId).last(StrUtil.format("limit {},{}", p, s)));
-            list.parallelStream().forEach(user -> {
-                System.out.println("同步短剧收藏数据：" + user.getUserId());
-                List<CourseCollect> batchList = courseCollectDao.selectList(Wrappers.<CourseCollect>lambdaQuery().eq(CourseCollect::getUserId, user.getUserId()));
-                esCourseCollectMapper.insertBatch(batchList);
-            });
         });
     }
 
-    private void syncCourseUser() {
+    private void syncCourseUser(List<String> list) {
-        int total = userService.count(Wrappers.emptyWrapper());
+        list.parallelStream().forEach(i -> {
-        List<Dict> pageList = buildPageList(total);
+            System.out.println("同步短剧用户数据：" + i);
-        pageList.parallelStream().forEach(param -> {
+            List<CourseUser> batchList = courseUserDao.selectList(Wrappers.<CourseUser>lambdaQuery().likeLeft(CourseUser::getCourseUserId, i));
-            Integer p = param.getInt("page");
+            esCourseUserMapper.insertBatch(batchList);
-            Integer s = param.getInt("size");
-            List<UserEntity> list = userService.list(Wrappers.<UserEntity>lambdaQuery().orderByAsc(UserEntity::getUserId).last(StrUtil.format("limit {},{}", p, s)));
-            list.parallelStream().forEach(user -> {
-                System.out.println("同步短剧用户数据：" + user.getUserId());
-                List<CourseUser> batchList = courseUserDao.selectList(Wrappers.<CourseUser>lambdaQuery().eq(CourseUser::getUserId, user.getUserId()));
-                esCourseUserMapper.insertBatch(batchList);
-            });
         });
     }
 
-    private void syncTaskCenterRecord() {
+    private void syncTaskCenterRecord(List<String> list) {
-        int total = userService.count(Wrappers.emptyWrapper());
+        list.parallelStream().forEach(i -> {
-        List<Dict> pageList = buildPageList(total);
+            System.out.println("同步任务中心数据：" + i);
-        pageList.parallelStream().forEach(param -> {
+            List<TaskCenterRecord> batchList = taskCenterRecordDao.selectList(Wrappers.<TaskCenterRecord>lambdaQuery().likeLeft(TaskCenterRecord::getId, i));
-            Integer p = param.getInt("page");
+            esTaskCenterRecordMapper.insertBatch(batchList);
-            Integer s = param.getInt("size");
-            List<UserEntity> list = userService.list(Wrappers.<UserEntity>lambdaQuery().orderByAsc(UserEntity::getUserId).last(StrUtil.format("limit {},{}", p, s)));
-            list.parallelStream().forEach(user -> {
-                System.out.println("同步任务中心数据：" + user.getUserId());
-                List<TaskCenterRecord> batchList = taskCenterRecordDao.selectList(Wrappers.<TaskCenterRecord>lambdaQuery().eq(TaskCenterRecord::getUserId, user.getUserId()));
-                esTaskCenterRecordMapper.insertBatch(batchList);
-            });
         });
     }
 
-    private void syncDiscSpinningRecord() {
+    private void syncDiscSpinningRecord(List<String> list) {
-        int total = userService.count(Wrappers.emptyWrapper());
+        list.parallelStream().forEach(i -> {
-        List<Dict> pageList = buildPageList(total);
+            System.out.println("同步抽奖记录数据：" + i);
-        pageList.parallelStream().forEach(param -> {
+            List<DiscSpinningRecord> batchList = discSpinningRecordDao.selectList(Wrappers.<DiscSpinningRecord>lambdaQuery().likeLeft(DiscSpinningRecord::getId, i));
-            Integer p = param.getInt("page");
+            esDiscSpinningRecordMapper.insertBatch(batchList);
-            Integer s = param.getInt("size");
-            List<UserEntity> list = userService.list(Wrappers.<UserEntity>lambdaQuery().orderByAsc(UserEntity::getUserId).last(StrUtil.format("limit {},{}", p, s)));
-            list.parallelStream().forEach(user -> {
-                System.out.println("同步抽奖记录数据：" + user.getUserId());
-                List<DiscSpinningRecord> batchList = discSpinningRecordDao.selectList(Wrappers.<DiscSpinningRecord>lambdaQuery().eq(DiscSpinningRecord::getUserId, user.getUserId()));
-                esDiscSpinningRecordMapper.insertBatch(batchList);
-            });
         });
     }
 
-    private void syncPayDetails() {
+    private void syncPayDetails(List<String> list) {
-        int total = userService.count(Wrappers.emptyWrapper());
+        list.parallelStream().forEach(i -> {
-        List<Dict> pageList = buildPageList(total);
+            System.out.println("同步支付详情数据：" + i);
-        pageList.parallelStream().forEach(param -> {
+            List<PayDetails> batchList = payDetailsDao.selectList(Wrappers.<PayDetails>lambdaQuery().likeLeft(PayDetails::getId, i));
-            Integer p = param.getInt("page");
+            esPayDetailsMapper.insertBatch(batchList);
-            Integer s = param.getInt("size");
-            List<UserEntity> list = userService.list(Wrappers.<UserEntity>lambdaQuery().orderByAsc(UserEntity::getUserId).last(StrUtil.format("limit {},{}", p, s)));
-            list.parallelStream().forEach(user -> {
-                System.out.println("同步支付详情数据：" + user.getUserId());
-                List<PayDetails> batchList = payDetailsDao.selectList(Wrappers.<PayDetails>lambdaQuery().eq(PayDetails::getUserId, user.getUserId()));
-                esPayDetailsMapper.insertBatch(batchList);
-            });
         });
     }
 
-    private void syncUserMoneyDetails() {
+    private void syncUserMoneyDetails(List<String> list) {
-        int total = userService.count(Wrappers.emptyWrapper());
+        list.parallelStream().forEach(i -> {
-        List<Dict> pageList = buildPageList(total);
+            System.out.println("同步用户资金明细数据：" + i);
-        pageList.parallelStream().forEach(param -> {
+            List<UserMoneyDetails> batchList = userMoneyDetailsDao.selectList(Wrappers.<UserMoneyDetails>lambdaQuery().likeLeft(UserMoneyDetails::getId, i));
-            Integer p = param.getInt("page");
+            esUserMoneyDetailsMapper.insertBatch(batchList);
-            Integer s = param.getInt("size");
-            List<UserEntity> list = userService.list(Wrappers.<UserEntity>lambdaQuery().orderByAsc(UserEntity::getUserId).last(StrUtil.format("limit {},{}", p, s)));
-            list.parallelStream().forEach(user -> {
-                System.out.println("同步用户资金明细数据：" + user.getUserId());
-                List<UserMoneyDetails> batchList = userMoneyDetailsDao.selectList(Wrappers.<UserMoneyDetails>lambdaQuery().eq(UserMoneyDetails::getUserId, user.getUserId()));
-                esUserMoneyDetailsMapper.insertBatch(batchList);
-            });
         });
     }
 

src/main/java/com/sqx/modules/es/service/impl/EsOrderServiceImpl.java

@@ -0,0 +1,89 @@
+package com.sqx.modules.es.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.sqx.modules.es.mapper.EsOrdersMapper;
+import com.sqx.modules.es.service.EsOrderService;
+import com.sqx.modules.orders.dao.OrdersDao;
+import com.sqx.modules.orders.entity.Orders;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.easyes.core.conditions.select.LambdaEsQueryWrapper;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.io.Serializable;
+
+/**
+ * @author tankaikai
+ * @since 2025-01-06 19:30
+ */
+@Service
+@Slf4j
+public class EsOrderServiceImpl implements EsOrderService {
+
+    @Resource
+    private OrdersDao ordersDao;
+
+    @Resource
+    private EsOrdersMapper esOrdersMapper;
+
+    @Override
+    public Orders selectById(Serializable id) {
+        Orders orders = esOrdersMapper.selectById(id);
+        if (orders != null) {
+            return orders;
+        }
+        return ordersDao.selectById(id);
+    }
+
+    @Override
+    public Orders selectOrdersByCourseIdAndUserId(Long userId, Long courseId) {
+        LambdaEsQueryWrapper<Orders> wrapper = new LambdaEsQueryWrapper<>();
+        wrapper.eq(Orders::getUserId, userId);
+        wrapper.eq(Orders::getCourseId, courseId);
+        Orders orders = esOrdersMapper.selectOne(wrapper);
+        if (orders != null) {
+            return orders;
+        }
+        return ordersDao.selectOrdersByCourseIdAndUserId(userId, courseId);
+    }
+
+    @Override
+    public Orders selectOrderByOrdersNo(String ordersNo) {
+        LambdaEsQueryWrapper<Orders> wrapper = new LambdaEsQueryWrapper<>();
+        wrapper.eq(Orders::getOrdersNo, ordersNo);
+        Orders orders = esOrdersMapper.selectOne(wrapper);
+        if (orders != null) {
+            return orders;
+        }
+        return ordersDao.selectOne(new QueryWrapper<Orders>().eq("orders_no", ordersNo));
+    }
+
+    @Override
+    public Orders selectOrderByTradeNo(String tradeNo) {
+        LambdaEsQueryWrapper<Orders> wrapper = new LambdaEsQueryWrapper<>();
+        wrapper.eq(Orders::getTradeNo, tradeNo);
+        Orders orders = esOrdersMapper.selectOne(wrapper);
+        if (orders != null) {
+            return orders;
+        }
+        return ordersDao.selectOne(new QueryWrapper<Orders>().eq("trade_no", tradeNo));
+    }
+
+    @Override
+    public Orders selectOrderByOrderIdAndUserId(String orderId, Long userId) {
+        LambdaEsQueryWrapper<Orders> wrapper = new LambdaEsQueryWrapper<>();
+        wrapper.eq(Orders::getOrdersNo, orderId);
+        wrapper.eq(Orders::getUserId, userId);
+        wrapper.limit(1);
+        Orders orders = esOrdersMapper.selectOne(wrapper);
+        if (orders != null) {
+            return orders;
+        }
+        QueryWrapper<Orders> queryWrapper = new QueryWrapper<>();
+        queryWrapper.eq("orders_no", orderId).eq("user_id", userId);
+        queryWrapper.last("limit 1");
+        return ordersDao.selectOne(queryWrapper);
+    }
+
+
+}

src/main/java/com/sqx/modules/orders/entity/Orders.java

@@ -18,7 +18,7 @@ import java.math.BigDecimal;
  */
 @Data
 @TableName("orders")
-@IndexName("orders")
+@IndexName(value = "orders",keepGlobalPrefix = true)
 public class Orders implements Serializable {
 
     private static final long serialVersionUID = 1L;

src/main/java/com/sqx/modules/orders/service/impl/OrdersServiceImpl.java

@@ -29,6 +29,7 @@ import com.sqx.modules.course.entity.CourseDetails;
 import com.sqx.modules.course.entity.CourseUser;
 import com.sqx.modules.course.service.CourseDetailsService;
 import com.sqx.modules.course.service.CourseUserService;
+import com.sqx.modules.es.service.EsOrderService;
 import com.sqx.modules.invite.service.InviteMoneyService;
 import com.sqx.modules.invite.service.InviteService;
 import com.sqx.modules.orders.dao.OrdersDao;
@@ -45,6 +46,7 @@ import org.apache.commons.lang.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import javax.annotation.Resource;
 import java.math.BigDecimal;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
@@ -87,13 +89,15 @@ public class OrdersServiceImpl extends ServiceImpl<OrdersDao, Orders> implements
     private DyService dyService;
     @Autowired
     private CourseDetailsDao courseDetailsDao;
+    @Resource
+    private EsOrderService esOrderService;
 
     private ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock(true);
 
 
     @Override
     public Orders selectOrdersByCourseIdAndUserId(Long userId, Long courseId) {
-        return baseMapper.selectOrdersByCourseIdAndUserId(userId, courseId);
+        return esOrderService.selectOrdersByCourseIdAndUserId(userId,courseId);
     }
 
 
@@ -489,7 +493,7 @@ public class OrdersServiceImpl extends ServiceImpl<OrdersDao, Orders> implements
     public Result payMoney(Long ordersId) {
         reentrantReadWriteLock.writeLock().lock();
         try {
-            Orders orders = baseMapper.selectById(ordersId);
+            Orders orders = esOrderService.selectById(ordersId);
             if (orders == null || !orders.getStatus().equals(0)) {
                 return Result.error("订单错误，请刷新后重试！");
             }
@@ -573,7 +577,7 @@ public class OrdersServiceImpl extends ServiceImpl<OrdersDao, Orders> implements
 
     @Override
     public Result refundOrder(Long ordersId, String refundContent) {
-        Orders bean = baseMapper.selectById(ordersId);
+        Orders bean = esOrderService.selectById(ordersId);
         if (!bean.getStatus().equals(1)) {
             return Result.error("订单未支付或已退款！");
         }
@@ -698,17 +702,17 @@ public class OrdersServiceImpl extends ServiceImpl<OrdersDao, Orders> implements
 
     @Override
     public Orders selectOrderById(Long orderId) {
-        return baseMapper.selectById(orderId);
+        return esOrderService.selectById(orderId);
     }
 
     @Override
     public Orders selectOrderByOrdersNo(String ordersNo) {
-        return baseMapper.selectOne(new QueryWrapper<Orders>().eq("orders_no", ordersNo));
+        return esOrderService.selectOrderByOrdersNo(ordersNo);
     }
 
     @Override
     public Orders selectOrderByTradeNo(String tradeNo) {
-        return baseMapper.selectOne(new QueryWrapper<Orders>().eq("trade_no", tradeNo));
+        return esOrderService.selectOrderByTradeNo(tradeNo);
     }
 
     @Override
@@ -718,8 +722,8 @@ public class OrdersServiceImpl extends ServiceImpl<OrdersDao, Orders> implements
 
     @Override
     public Result selectOrdersMoneyList(Integer page, Integer limit, Integer flag, String time) {
-        PageHelper.startPage(page,limit);
+        PageHelper.startPage(page, limit);
-        return Result.success().put("data", PageUtils.page(new PageInfo<>(baseMapper.selectOrdersMoneyList( flag, time)),true));
+        return Result.success().put("data", PageUtils.page(new PageInfo<>(baseMapper.selectOrdersMoneyList(flag, time)), true));
     }
 
 

src/main/java/com/sqx/modules/pay/controller/app/AliPayController.java

@@ -23,6 +23,7 @@ import com.sqx.modules.app.service.UserMoneyService;
 import com.sqx.modules.app.service.UserService;
 import com.sqx.modules.common.entity.CommonInfo;
 import com.sqx.modules.common.service.CommonInfoService;
+import com.sqx.modules.es.service.EsOrderService;
 import com.sqx.modules.invite.service.InviteService;
 import com.sqx.modules.orders.dao.OrdersDao;
 import com.sqx.modules.orders.entity.Orders;
@@ -42,6 +43,7 @@ import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RestController;
 
+import javax.annotation.Resource;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.math.BigDecimal;
@@ -78,6 +80,8 @@ public class AliPayController {
     private UserMoneyService userMoneyService;
     @Autowired
     private PayClassifyService payClassifyService;
+    @Resource
+    private EsOrderService esOrderService;
 
     private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
@@ -192,7 +196,7 @@ public class AliPayController {
         String name = one3.getValue();
         String url = one.getValue() + "/sqx_fast/app/aliPay/notifyApp";
         log.info("回调地址:" + url);
-        Orders orders = ordersDao.selectById(orderId);
+        Orders orders = esOrderService.selectById(orderId);
         PayDetails payDetails = payDetailsDao.selectByOrderId(orders.getOrdersNo());
         if (payDetails == null) {
             payDetails = new PayDetails();

src/main/java/com/sqx/modules/pay/controller/app/WuyouController.java

@@ -2,28 +2,15 @@ package com.sqx.modules.pay.controller.app;
 
 import cn.hutool.core.date.DateTime;
 import cn.hutool.core.date.DateUtil;
-import cn.hutool.core.thread.ThreadUtil;
 import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.sqx.common.annotation.Debounce;
-import com.sqx.common.utils.ApiAccessLimitUtil;
-import com.sqx.common.utils.DateUtils;
-import com.sqx.common.utils.RedisKeys;
 import com.sqx.common.utils.Result;
-import com.sqx.modules.app.entity.InviteAchievement;
-import com.sqx.modules.app.entity.UserEntity;
 import com.sqx.modules.app.entity.UserMoneyDetails;
-import com.sqx.modules.app.service.InviteAchievementService;
 import com.sqx.modules.app.service.UserMoneyDetailsService;
 import com.sqx.modules.app.service.UserMoneyService;
-import com.sqx.modules.app.service.UserService;
+import com.sqx.modules.es.service.EsOrderService;
-import com.sqx.modules.common.service.CommonInfoService;
-import com.sqx.modules.complet.entity.CompletAward;
-import com.sqx.modules.complet.service.CompletAwardService;
-import com.sqx.modules.invite.dao.InviteDao;
-import com.sqx.modules.invite.entity.Invite;
-import com.sqx.modules.invite.service.InviteService;
 import com.sqx.modules.job.task.TempOrdersTask;
 import com.sqx.modules.orders.entity.Orders;
 import com.sqx.modules.orders.service.OrdersService;
@@ -35,26 +22,22 @@ import com.sqx.modules.pay.wuyou.BaseResp;
 import com.sqx.modules.pay.wuyou.Encrypt;
 import com.sqx.modules.pay.wuyou.NotifyDto;
 import com.sqx.modules.pay.wuyou.WuyouPay;
-import com.sqx.modules.sys.entity.SysUserEntity;
-import com.sqx.modules.sys.service.SysUserService;
 import com.sqx.modules.utils.AliPayOrderUtil;
 import com.sqx.modules.utils.InvitationCodeUtil;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import org.springframework.context.ApplicationContext;
-import org.springframework.transaction.annotation.Transactional;
 import org.springframework.web.bind.annotation.*;
 
+import javax.annotation.Resource;
 import javax.servlet.http.HttpServletRequest;
 import java.math.BigDecimal;
-import java.text.SimpleDateFormat;
+import java.util.Date;
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.stream.Collectors;
 
 /**
  * @author GYJ
@@ -73,6 +56,9 @@ public class WuyouController {
     private final WuyouPay wuyouPay;
     private final TempOrdersTask ordersTask;
 
+    @Resource
+    private EsOrderService esOrderService;
+
     WuyouController(OrdersService ordersService, PayDetailsDao payDetailsDao, CashOutDao cashOutDao, UserMoneyService userMoneyService,
                     UserMoneyDetailsService userMoneyDetailsService, TempOrdersTask ordersTask,
                     WuyouPay wuyouPay) {
@@ -89,7 +75,7 @@ public class WuyouController {
     @ApiOperation("支付订单")
     @GetMapping("/payOrder/{orderId}")
     public Result payOrder(HttpServletRequest request, @PathVariable("orderId") Long orderId, @RequestParam(value = "payType", required = false) String payType) {
-        Orders order = ordersService.getById(orderId);
+        Orders order = esOrderService.selectById(orderId);
         if (order == null) {
             return Result.error("订单不存在");
         }
@@ -140,7 +126,7 @@ public class WuyouController {
     @ApiOperation("查询订单支付结果")
     @GetMapping("/queryOrder/{orderId}")
     public Result queryOrder(HttpServletRequest request, @PathVariable("orderId") Long orderId) {
-        Orders order = ordersService.getById(orderId);
+        Orders order = esOrderService.selectById(orderId);
         if (order == null) {
             return Result.error("订单不存在");
         }
@@ -228,10 +214,7 @@ public class WuyouController {
                 return "success";
             }
 
-            QueryWrapper<Orders> queryWrapper = new QueryWrapper<>();
+            Orders order = esOrderService.selectOrderByOrderIdAndUserId(payDetails.getOrderId(), payDetails.getUserId());
-            queryWrapper.eq("orders_no", payDetails.getOrderId()).eq("user_id", payDetails.getUserId());
-            queryWrapper.last("limit 1");
-            Orders order = ordersService.getOne(queryWrapper);
             if (order == null) {
                 log.error("无忧支付回调订单不存在, 参数: {}", JSONObject.toJSONString(notifyDto));
                 return "订单不存在";

src/main/java/com/sqx/modules/pay/entity/CashOut.java

@@ -15,12 +15,13 @@ import java.math.BigDecimal;
 
 /**
  * 提现申请
+ *
  * @author fang
  * @date 2020/7/8
  */
 @Data
 @TableName("cash_out")
-@IndexName("cash_out")
+@IndexName(value = "cash_out", keepGlobalPrefix = true)
 public class CashOut implements Serializable {
 
     private static final long serialVersionUID = 1L;
@@ -29,9 +30,9 @@ public class CashOut implements Serializable {
      * 申请提现id
      */
     @TableId(type = IdType.ID_WORKER)
-    @IndexId(type= org.dromara.easyes.annotation.rely.IdType.CUSTOMIZE)
+    @IndexId(type = org.dromara.easyes.annotation.rely.IdType.CUSTOMIZE)
     @JsonSerialize(using = ToStringSerializer.class)
-    private long id;
+    private Long id;
 
     /**
      * 申请时间

src/main/java/com/sqx/modules/pay/entity/PayDetails.java

@@ -14,12 +14,13 @@ import java.io.Serializable;
 import java.math.BigDecimal;
 
 /**
- *  充值记录
+ * 充值记录
+ *
  * @author fang 2020-05-14
  */
 @Data
 @TableName("pay_details")
-@IndexName("pay_details")
+@IndexName(value = "pay_details", keepGlobalPrefix = true)
 public class PayDetails implements Serializable {
 
     private static final long serialVersionUID = 1L;
@@ -28,7 +29,7 @@ public class PayDetails implements Serializable {
      * 充值记录id
      */
     @TableId(type = IdType.ID_WORKER)
-    @IndexId(type= org.dromara.easyes.annotation.rely.IdType.CUSTOMIZE)
+    @IndexId(type = org.dromara.easyes.annotation.rely.IdType.CUSTOMIZE)
     @JsonSerialize(using = ToStringSerializer.class)
     private Long id;
 

src/main/java/com/sqx/modules/pay/service/impl/KsServiceImpl.java

@@ -13,6 +13,7 @@ import com.sqx.modules.app.service.UserService;
 import com.sqx.modules.common.entity.CommonInfo;
 import com.sqx.modules.common.service.CommonInfoService;
 import com.sqx.modules.course.service.CourseService;
+import com.sqx.modules.es.service.EsOrderService;
 import com.sqx.modules.invite.service.InviteService;
 import com.sqx.modules.orders.entity.Orders;
 import com.sqx.modules.orders.service.OrdersService;
@@ -30,6 +31,7 @@ import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import javax.annotation.Resource;
 import java.math.BigDecimal;
 import java.text.SimpleDateFormat;
 import java.util.Date;
@@ -65,6 +67,8 @@ public class KsServiceImpl implements KsService {
     private CourseService courseService;
     @Autowired
     private PayClassifyService payClassifyService;
+    @Resource
+    private EsOrderService esOrderService;
 
     private SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");;
 
@@ -184,7 +188,7 @@ public class KsServiceImpl implements KsService {
                     String format = sdf.format(new Date());
                     payDetailsDao.updateState(payDetails.getId(),1,format,"","");
                     if(payDetails.getType()==1){
-                        Orders orders = ordersService.selectOrderByOrdersNo(payDetails.getOrderId());
+                        Orders orders = esOrderService.selectOrderByOrdersNo(payDetails.getOrderId());
                         orders.setPayWay(8);
                         orders.setStatus(1);
                         orders.setPayTime(DateUtils.format(new Date()));

src/main/java/com/sqx/modules/taskCenter/entity/TaskCenterRecord.java

@@ -18,7 +18,7 @@ import org.dromara.easyes.annotation.IndexName;
  */
 @Data
 @TableName("task_center_record")
-@IndexName("task_center_record")
+@IndexName(value = "task_center_record", keepGlobalPrefix = true)
 @ApiModel(value = "任务领取记录 实体类")
 public class TaskCenterRecord extends Model<TaskCenterRecord> {
 

src/test/java/com/sqx/es/SyncDataToEsScriptGenTest.java

@@ -0,0 +1,61 @@
+package com.sqx.es;
+
+import cn.hutool.core.convert.Convert;
+import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.json.JSONArray;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * @author tankaikai
+ * @since 2025-01-08 13:33
+ */
+public class SyncDataToEsScriptGenTest {
+
+    public static void main(String[] args) {
+        build("dev");
+    }
+
+    private static void build(String env) {
+        String absolutePath = FileUtil.getAbsolutePath("", SyncDataToEsScriptGenTest.class);
+        String basePath = FileUtil.getParent(absolutePath, 5) + "/src/test/java/com/sqx/es";
+        String configPath = basePath + "/" + env + "-config-data.json";
+        String config = FileUtil.readUtf8String(configPath);
+        System.out.println(config);
+        String outPath = basePath + "/" + "out";
+        FileUtil.del(outPath);
+        FileUtil.mkdir(outPath);
+        String templatePath = basePath + "/" + "template.yml";
+        String template = FileUtil.readUtf8String(templatePath);
+        JSONObject data = JSONUtil.parseObj(config);
+
+        JSONArray dbList = data.getJSONArray("dbList");
+        JSONArray tableList = data.getJSONArray("tableList");
+        Set<String> dbSet = dbList.stream().map(obj -> Convert.toStr(obj)).collect(Collectors.toSet());
+        for (String dbName : dbSet) {
+            for (int i = 0; i < tableList.size(); i++) {
+                Map<String, String> params = new HashMap();
+                params.put("env", data.getStr("env"));
+                params.put("dbName", dbName);
+                JSONObject table = tableList.getJSONObject(i);
+                String tableName = table.getStr("tableName");
+                params.put("tableName", tableName);
+                params.put("mappingSql", table.getStr("mappingSql"));
+                String content = StrUtil.format(template, params);
+                String fileName = tableName + "@" + env + "." + dbName + ".yml";
+                FileUtil.writeUtf8String(content, new File(outPath + "/" + fileName));
+            }
+        }
+
+
+        StrUtil.format(template, config);
+        //String template = fileReader.readString();
+    }
+}

src/test/java/com/sqx/es/dev-config-data.json

@@ -0,0 +1,44 @@
+{
+  "env": "dev",
+  "dbList": [
+    "duanju-0",
+    "duanju-1",
+    "duanju-2",
+    "duanju-3",
+    "duanju-4"
+  ],
+  "tableList": [
+    {
+      "tableName": "user_money_details",
+      "mappingSql": "select a.id as _id, a.user_id, a.sys_user_id, a.by_user_id, a.title, a.classify, a.type, a.state, a.money, a.content, a.money_type, a.create_time, a.source_id from user_money_details a"
+    },
+    {
+      "tableName": "orders",
+      "mappingSql": "select a.orders_id as _id, a.orders_no, a.trade_no, a.user_id, a.course_id, a.pay_money, a.pay_way, a.status, a.create_time, a.refund_content, a.orders_type, a.vip_name_type, a.course_details_id, a.course_details_ids, a.qd_money, a.sys_user_id, a.one_money, a.one_user_id, a.two_money, a.two_user_id, a.pay_time, a.pay_diamond, a.diamond from orders a"
+    },
+    {
+      "tableName": "course_collect",
+      "mappingSql": "select a.course_collect_id as _id, a.course_id, a.user_id, a.create_time, a.course_details_id, a.classify, a.update_time from course_collect a"
+    },
+    {
+      "tableName": "pay_details",
+      "mappingSql": "select a.id as _id, a.classify, a.trade_no, a.money, a.user_id, a.state, a.create_time, a.pay_time, a.type, a.remark, a.order_id, a.product_id, a.pay_diamond, a.diamond, a.third_order_no from pay_details a"
+    },
+    {
+      "tableName": "disc_spinning_record",
+      "mappingSql": "select a.id as _id, a.user_id, a.source_id, a.img_url, a.name, a.type, a.number, a.target, a.target_id, a.draw_day, a.create_time, a.source from disc_spinning_record a"
+    },
+    {
+      "tableName": "cash_out",
+      "mappingSql": "select a.id as _id, a.create_at, a.is_out, a.money, a.out_at, a.relation_id, a.user_id, a.zhifubao, a.zhifubao_name, a.order_number, a.state, a.refund, a.rate, a.sys_user_id, a.user_type, a.withdraw_type from cash_out a"
+    },
+    {
+      "tableName": "course_user",
+      "mappingSql": "select a.course_user_id as _id, a.course_id, a.order_id, a.user_id, a.create_time, a.update_time, a.course_details_id, a.classify from course_user a"
+    },
+    {
+      "tableName": "task_center_record",
+      "mappingSql": "select a.id as _id, a.user_id, a.task_id, a.source_id, a.name, a.type, a.target_id, a.number, a.create_time, a.update_time from task_center_record a"
+    }
+  ]
+}

src/test/java/com/sqx/es/out/cash_out@dev.duanju-0.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-0
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_cash_out
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.create_at, a.is_out, a.money, a.out_at, a.relation_id, a.user_id, a.zhifubao, a.zhifubao_name, a.order_number, a.state, a.refund, a.rate, a.sys_user_id, a.user_type, a.withdraw_type from cash_out a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/cash_out@dev.duanju-1.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-1
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_cash_out
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.create_at, a.is_out, a.money, a.out_at, a.relation_id, a.user_id, a.zhifubao, a.zhifubao_name, a.order_number, a.state, a.refund, a.rate, a.sys_user_id, a.user_type, a.withdraw_type from cash_out a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/cash_out@dev.duanju-2.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-2
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_cash_out
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.create_at, a.is_out, a.money, a.out_at, a.relation_id, a.user_id, a.zhifubao, a.zhifubao_name, a.order_number, a.state, a.refund, a.rate, a.sys_user_id, a.user_type, a.withdraw_type from cash_out a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/cash_out@dev.duanju-3.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-3
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_cash_out
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.create_at, a.is_out, a.money, a.out_at, a.relation_id, a.user_id, a.zhifubao, a.zhifubao_name, a.order_number, a.state, a.refund, a.rate, a.sys_user_id, a.user_type, a.withdraw_type from cash_out a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/cash_out@dev.duanju-4.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-4
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_cash_out
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.create_at, a.is_out, a.money, a.out_at, a.relation_id, a.user_id, a.zhifubao, a.zhifubao_name, a.order_number, a.state, a.refund, a.rate, a.sys_user_id, a.user_type, a.withdraw_type from cash_out a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_collect@dev.duanju-0.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-0
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_collect
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_collect_id as _id, a.course_id, a.user_id, a.create_time, a.course_details_id, a.classify, a.update_time from course_collect a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_collect@dev.duanju-1.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-1
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_collect
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_collect_id as _id, a.course_id, a.user_id, a.create_time, a.course_details_id, a.classify, a.update_time from course_collect a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_collect@dev.duanju-2.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-2
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_collect
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_collect_id as _id, a.course_id, a.user_id, a.create_time, a.course_details_id, a.classify, a.update_time from course_collect a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_collect@dev.duanju-3.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-3
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_collect
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_collect_id as _id, a.course_id, a.user_id, a.create_time, a.course_details_id, a.classify, a.update_time from course_collect a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_collect@dev.duanju-4.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-4
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_collect
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_collect_id as _id, a.course_id, a.user_id, a.create_time, a.course_details_id, a.classify, a.update_time from course_collect a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_user@dev.duanju-0.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-0
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_user
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_user_id as _id, a.course_id, a.order_id, a.user_id, a.create_time, a.update_time, a.course_details_id, a.classify from course_user a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_user@dev.duanju-1.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-1
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_user
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_user_id as _id, a.course_id, a.order_id, a.user_id, a.create_time, a.update_time, a.course_details_id, a.classify from course_user a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_user@dev.duanju-2.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-2
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_user
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_user_id as _id, a.course_id, a.order_id, a.user_id, a.create_time, a.update_time, a.course_details_id, a.classify from course_user a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_user@dev.duanju-3.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-3
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_user
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_user_id as _id, a.course_id, a.order_id, a.user_id, a.create_time, a.update_time, a.course_details_id, a.classify from course_user a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/course_user@dev.duanju-4.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-4
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_course_user
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.course_user_id as _id, a.course_id, a.order_id, a.user_id, a.create_time, a.update_time, a.course_details_id, a.classify from course_user a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/disc_spinning_record@dev.duanju-0.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-0
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_disc_spinning_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.source_id, a.img_url, a.name, a.type, a.number, a.target, a.target_id, a.draw_day, a.create_time, a.source from disc_spinning_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/disc_spinning_record@dev.duanju-1.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-1
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_disc_spinning_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.source_id, a.img_url, a.name, a.type, a.number, a.target, a.target_id, a.draw_day, a.create_time, a.source from disc_spinning_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/disc_spinning_record@dev.duanju-2.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-2
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_disc_spinning_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.source_id, a.img_url, a.name, a.type, a.number, a.target, a.target_id, a.draw_day, a.create_time, a.source from disc_spinning_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/disc_spinning_record@dev.duanju-3.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-3
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_disc_spinning_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.source_id, a.img_url, a.name, a.type, a.number, a.target, a.target_id, a.draw_day, a.create_time, a.source from disc_spinning_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/disc_spinning_record@dev.duanju-4.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-4
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_disc_spinning_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.source_id, a.img_url, a.name, a.type, a.number, a.target, a.target_id, a.draw_day, a.create_time, a.source from disc_spinning_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/orders@dev.duanju-0.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-0
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_orders
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.orders_id as _id, a.orders_no, a.trade_no, a.user_id, a.course_id, a.pay_money, a.pay_way, a.status, a.create_time, a.refund_content, a.orders_type, a.vip_name_type, a.course_details_id, a.course_details_ids, a.qd_money, a.sys_user_id, a.one_money, a.one_user_id, a.two_money, a.two_user_id, a.pay_time, a.pay_diamond, a.diamond from orders a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/orders@dev.duanju-1.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-1
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_orders
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.orders_id as _id, a.orders_no, a.trade_no, a.user_id, a.course_id, a.pay_money, a.pay_way, a.status, a.create_time, a.refund_content, a.orders_type, a.vip_name_type, a.course_details_id, a.course_details_ids, a.qd_money, a.sys_user_id, a.one_money, a.one_user_id, a.two_money, a.two_user_id, a.pay_time, a.pay_diamond, a.diamond from orders a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/orders@dev.duanju-2.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-2
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_orders
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.orders_id as _id, a.orders_no, a.trade_no, a.user_id, a.course_id, a.pay_money, a.pay_way, a.status, a.create_time, a.refund_content, a.orders_type, a.vip_name_type, a.course_details_id, a.course_details_ids, a.qd_money, a.sys_user_id, a.one_money, a.one_user_id, a.two_money, a.two_user_id, a.pay_time, a.pay_diamond, a.diamond from orders a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/orders@dev.duanju-3.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-3
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_orders
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.orders_id as _id, a.orders_no, a.trade_no, a.user_id, a.course_id, a.pay_money, a.pay_way, a.status, a.create_time, a.refund_content, a.orders_type, a.vip_name_type, a.course_details_id, a.course_details_ids, a.qd_money, a.sys_user_id, a.one_money, a.one_user_id, a.two_money, a.two_user_id, a.pay_time, a.pay_diamond, a.diamond from orders a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/orders@dev.duanju-4.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-4
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_orders
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.orders_id as _id, a.orders_no, a.trade_no, a.user_id, a.course_id, a.pay_money, a.pay_way, a.status, a.create_time, a.refund_content, a.orders_type, a.vip_name_type, a.course_details_id, a.course_details_ids, a.qd_money, a.sys_user_id, a.one_money, a.one_user_id, a.two_money, a.two_user_id, a.pay_time, a.pay_diamond, a.diamond from orders a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/pay_details@dev.duanju-0.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-0
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_pay_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.classify, a.trade_no, a.money, a.user_id, a.state, a.create_time, a.pay_time, a.type, a.remark, a.order_id, a.product_id, a.pay_diamond, a.diamond, a.third_order_no from pay_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/pay_details@dev.duanju-1.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-1
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_pay_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.classify, a.trade_no, a.money, a.user_id, a.state, a.create_time, a.pay_time, a.type, a.remark, a.order_id, a.product_id, a.pay_diamond, a.diamond, a.third_order_no from pay_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/pay_details@dev.duanju-2.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-2
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_pay_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.classify, a.trade_no, a.money, a.user_id, a.state, a.create_time, a.pay_time, a.type, a.remark, a.order_id, a.product_id, a.pay_diamond, a.diamond, a.third_order_no from pay_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/pay_details@dev.duanju-3.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-3
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_pay_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.classify, a.trade_no, a.money, a.user_id, a.state, a.create_time, a.pay_time, a.type, a.remark, a.order_id, a.product_id, a.pay_diamond, a.diamond, a.third_order_no from pay_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/pay_details@dev.duanju-4.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-4
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_pay_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.classify, a.trade_no, a.money, a.user_id, a.state, a.create_time, a.pay_time, a.type, a.remark, a.order_id, a.product_id, a.pay_diamond, a.diamond, a.third_order_no from pay_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/task_center_record@dev.duanju-0.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-0
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_task_center_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.task_id, a.source_id, a.name, a.type, a.target_id, a.number, a.create_time, a.update_time from task_center_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/task_center_record@dev.duanju-1.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-1
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_task_center_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.task_id, a.source_id, a.name, a.type, a.target_id, a.number, a.create_time, a.update_time from task_center_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/task_center_record@dev.duanju-2.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-2
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_task_center_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.task_id, a.source_id, a.name, a.type, a.target_id, a.number, a.create_time, a.update_time from task_center_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/task_center_record@dev.duanju-3.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-3
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_task_center_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.task_id, a.source_id, a.name, a.type, a.target_id, a.number, a.create_time, a.update_time from task_center_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/task_center_record@dev.duanju-4.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-4
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_task_center_record
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.task_id, a.source_id, a.name, a.type, a.target_id, a.number, a.create_time, a.update_time from task_center_record a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/user_money_details@dev.duanju-0.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-0
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_user_money_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.sys_user_id, a.by_user_id, a.title, a.classify, a.type, a.state, a.money, a.content, a.money_type, a.create_time, a.source_id from user_money_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/user_money_details@dev.duanju-1.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-1
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_user_money_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.sys_user_id, a.by_user_id, a.title, a.classify, a.type, a.state, a.money, a.content, a.money_type, a.create_time, a.source_id from user_money_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/user_money_details@dev.duanju-2.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-2
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_user_money_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.sys_user_id, a.by_user_id, a.title, a.classify, a.type, a.state, a.money, a.content, a.money_type, a.create_time, a.source_id from user_money_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/user_money_details@dev.duanju-3.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-3
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_user_money_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.sys_user_id, a.by_user_id, a.title, a.classify, a.type, a.state, a.money, a.content, a.money_type, a.create_time, a.source_id from user_money_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/out/user_money_details@dev.duanju-4.yml

@@ -0,0 +1,16 @@
+dataSourceKey: dev_duanju-4
+outerAdapterKey: esKey1
+destination: dev_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_user_money_details
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "select a.id as _id, a.user_id, a.sys_user_id, a.by_user_id, a.title, a.classify, a.type, a.state, a.money, a.content, a.money_type, a.create_time, a.source_id from user_money_details a"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000

src/test/java/com/sqx/es/prod-config-data.json

@@ -0,0 +1,44 @@
+{
+  "env": "prod",
+  "dbList": [
+    "duanju-0",
+    "duanju-1",
+    "duanju-2",
+    "duanju-3",
+    "duanju-4"
+  ],
+  "tableList": [
+    {
+      "tableName": "user_money_details",
+      "mappingSql": "select a.id as _id, a.user_id, a.sys_user_id, a.by_user_id, a.title, a.classify, a.type, a.state, a.money, a.content, a.money_type, a.create_time, a.source_id from user_money_details a"
+    },
+    {
+      "tableName": "orders",
+      "mappingSql": "select a.orders_id as _id, a.orders_no, a.trade_no, a.user_id, a.course_id, a.pay_money, a.pay_way, a.status, a.create_time, a.refund_content, a.orders_type, a.vip_name_type, a.course_details_id, a.course_details_ids, a.qd_money, a.sys_user_id, a.one_money, a.one_user_id, a.two_money, a.two_user_id, a.pay_time, a.pay_diamond, a.diamond from orders a"
+    },
+    {
+      "tableName": "course_collect",
+      "mappingSql": "select a.course_collect_id as _id, a.course_id, a.user_id, a.create_time, a.course_details_id, a.classify, a.update_time from course_collect a"
+    },
+    {
+      "tableName": "pay_details",
+      "mappingSql": "select a.id as _id, a.classify, a.trade_no, a.money, a.user_id, a.state, a.create_time, a.pay_time, a.type, a.remark, a.order_id, a.product_id, a.pay_diamond, a.diamond, a.third_order_no from pay_details a"
+    },
+    {
+      "tableName": "disc_spinning_record",
+      "mappingSql": "select a.id as _id, a.user_id, a.source_id, a.img_url, a.name, a.type, a.number, a.target, a.target_id, a.draw_day, a.create_time, a.source from disc_spinning_record a"
+    },
+    {
+      "tableName": "cash_out",
+      "mappingSql": "select a.id as _id, a.create_at, a.is_out, a.money, a.out_at, a.relation_id, a.user_id, a.zhifubao, a.zhifubao_name, a.order_number, a.state, a.refund, a.rate, a.sys_user_id, a.user_type, a.withdraw_type from cash_out a"
+    },
+    {
+      "tableName": "course_user",
+      "mappingSql": "select a.course_user_id as _id, a.course_id, a.order_id, a.user_id, a.create_time, a.update_time, a.course_details_id, a.classify from course_user a"
+    },
+    {
+      "tableName": "task_center_record",
+      "mappingSql": "select a.id as _id, a.user_id, a.task_id, a.source_id, a.name, a.type, a.target_id, a.number, a.create_time, a.update_time from task_center_record a"
+    }
+  ]
+}

src/test/java/com/sqx/es/template.yml

@@ -0,0 +1,16 @@
+dataSourceKey: {env}_{dbName}
+outerAdapterKey: esKey1
+destination: {env}_sync_data_to_es
+groupId: g1
+esMapping:
+  _index: dev_{tableName}
+  #  _type: doc
+  _id: _id
+  upsert: true
+  #  pk: "orders_id"
+  sql: "{mappingSql}"
+  #  objFields:
+  #    _labels: array:;
+  #    _obj: object
+  #  etlCondition: "where 1=1"
+  commitBatch: 3000