存储课程流量日志数据的转移

fs-admin/src/main/java/com/fs/course/task/VideoTask.java

@@ -5,6 +5,7 @@ import com.fs.common.core.redis.RedisCache;
 import com.fs.course.config.CourseConfig;
 import com.fs.course.domain.*;
 import com.fs.course.mapper.*;
+import com.fs.course.service.IFsCourseTrafficLogService;
 import com.fs.course.service.IFsUserVideoCommentService;
 import com.fs.his.domain.FsUser;
 import com.fs.his.mapper.FsUserMapper;
@@ -18,6 +19,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 import java.time.LocalDate;
@@ -55,6 +57,9 @@ public class VideoTask {
     @Autowired
     private FsCourseRedPacketLogMapper redPacketLogMapper;
 
+    @Autowired
+    private IFsCourseTrafficLogService iFsCourseTrafficLogService;
+
     public void autoNotShowVideo() {
         List<Long> list = videoMapper.selectNotAuditVideo();
         for (Long videoId : list){
@@ -238,4 +243,15 @@ public class VideoTask {
 
         }
     }
+
+    /**
+     * 存储课程流量日志
+     * @throws Exception
+     */
+    @Scheduled(fixedRate = 1000)
+    public void saveCourseTrafficLog() throws Exception
+    {
+        iFsCourseTrafficLogService.saveCourseTrafficLog();
+    }
+
 }

fs-framework/src/main/java/com/fs/framework/config/DataSourceConfig.java

@@ -33,13 +33,18 @@ public class DataSourceConfig {
         return new DruidDataSource();
     }
 
-
+    @Bean
+    @ConfigurationProperties(prefix = "spring.datasource.clickhouse")
+    public DataSource clickhouseDataSource() {
+        return new DruidDataSource();
+    }
 
     @Bean
     @Primary
-    public DynamicDataSource dataSource(@Qualifier("masterDataSource") DataSource masterDataSource, @Qualifier("sopDataSource") DataSource sopDataSource) {
+    public DynamicDataSource dataSource(@Qualifier("clickhouseDataSource") DataSource clickhouseDataSource,@Qualifier("masterDataSource") DataSource masterDataSource, @Qualifier("sopDataSource") DataSource sopDataSource) {
         Map<Object, Object> targetDataSources = new HashMap<>();
         targetDataSources.put(DataSourceType.SOP.name(), sopDataSource);
+        targetDataSources.put(DataSourceType.CLICKHOUSE.name(), clickhouseDataSource);
         return new DynamicDataSource(masterDataSource, targetDataSources);
     }
 

fs-service/src/main/java/com/fs/course/domain/FsCourseTrafficLog.java

@@ -56,6 +56,18 @@ public class FsCourseTrafficLog extends BaseEntity
     @Excel(name = "课程id")
     private Long courseId;
 
+    /**
+     * 项目
+     */
+    private Long project;
+
+    /**
+     * 营期id
+     */
+    private Long periodId;
+
+
+
 //    @JsonFormat(pattern = "yyyy-MM-dd")
 //    private Date time;
 

fs-service/src/main/java/com/fs/course/mapper/FsCourseTrafficLogMapper.java

@@ -1,11 +1,15 @@
 package com.fs.course.mapper;
 
 import java.util.List;
+
+import com.fs.common.annotation.DataSource;
+import com.fs.common.enums.DataSourceType;
 import com.fs.course.domain.FsCourseTrafficLog;
 import com.fs.course.param.FsCourseTrafficLogParam;
 import com.fs.course.vo.FsCourseTrafficLogListVO;
 import org.apache.ibatis.annotations.Param;
 import org.apache.ibatis.annotations.Select;
+import org.springframework.stereotype.Repository;
 
 /**
  * 短链课程流量记录Mapper接口
@@ -13,6 +17,7 @@ import org.apache.ibatis.annotations.Select;
  * @author fs
  * @date 2024-10-31
  */
+@Repository
 public interface FsCourseTrafficLogMapper
 {
     /**
@@ -97,5 +102,10 @@ public interface FsCourseTrafficLogMapper
     Long getYesterdayTrafficLog();
 
     Long getMonthTrafficLog();
+    @Select("select count(1) from fs_course_traffic_log l WHERE l.create_time < DATE_SUB(NOW(), INTERVAL 2 DAY)")
+    Integer saveCourseTrafficLogByTwoDaysLaterCount();
 
+    List<FsCourseTrafficLog> selectCourseTrafficLogByTwoDaysLater(@Param("offset")Integer offset,@Param("limit")Integer limit);
+    @DataSource(DataSourceType.CLICKHOUSE)
+    void insertCourseTrafficLogByTwoDaysLaterBatch(@Param("list") List<FsCourseTrafficLog> redPacketLogs);
 }

fs-service/src/main/java/com/fs/course/service/IFsCourseTrafficLogService.java

@@ -63,4 +63,9 @@ public interface IFsCourseTrafficLogService
 
 
     List<FsCourseTrafficLogListVO> selectTrafficByCompany(FsCourseTrafficLogParam param);
+
+    /**
+     * 存储课程流量日志
+     */
+    void saveCourseTrafficLog();
 }

fs-service/src/main/java/com/fs/course/service/impl/FsCourseTrafficLogServiceImpl.java

@@ -1,6 +1,8 @@
 package com.fs.course.service.impl;
 
+import java.text.SimpleDateFormat;
 import java.util.Collections;
+import java.util.Date;
 import java.util.List;
 import com.fs.common.utils.DateUtils;
 import com.fs.course.param.FsCourseTrafficLogParam;
@@ -100,4 +102,68 @@ public class FsCourseTrafficLogServiceImpl implements IFsCourseTrafficLogService
     public List<FsCourseTrafficLogListVO> selectTrafficByCompany(FsCourseTrafficLogParam param) {
         return fsCourseTrafficLogMapper.selectTrafficByCompany(param);
     }
+
+
+    @Override
+    public void saveCourseTrafficLog() {
+        Integer count = fsCourseTrafficLogMapper.saveCourseTrafficLogByTwoDaysLaterCount();
+        Integer limit = 1000;
+        System.out.println("总数:" + count);
+
+        // Start timing
+        long startTime = System.currentTimeMillis();
+        System.out.println("开始处理，时间: " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
+
+        // 计算需要查询的次数
+        int totalPages = (count + limit - 1) / limit; // 向上取整
+        for (int page = 0; page < totalPages; page++) {
+            int offset = page * limit;
+
+            System.out.println("开始查询");
+            List<FsCourseTrafficLog> redPacketLogs = fsCourseTrafficLogMapper.selectCourseTrafficLogByTwoDaysLater(0,limit);
+            System.out.println(redPacketLogs);
+            // 处理当前批次的1000条数据
+            if (redPacketLogs==null ||redPacketLogs.isEmpty()) {
+                System.out.println("没有数据");
+                continue;
+            }
+            fsCourseTrafficLogMapper.insertCourseTrafficLogByTwoDaysLaterBatch(redPacketLogs);
+            Long[] delArray = redPacketLogs.stream()
+                    .map(FsCourseTrafficLog::getLogId)
+                    .toArray(Long[]::new);
+            fsCourseTrafficLogMapper.deleteFsCourseTrafficLogByLogIds(delArray);
+
+            // 打印进度
+            System.out.printf("已处理: %d/%d (%.2f%%)%n",
+                    Math.min(offset + limit, count),
+                    count,
+                    (double)(offset + limit) / count * 100);
+
+            // Calculate estimated remaining time
+            long currentTime = System.currentTimeMillis();
+            long elapsedTime = currentTime - startTime;
+            double timePerBatch = (double)elapsedTime / (page + 1);
+            long estimatedRemaining = (long)(timePerBatch * (totalPages - page - 1));
+
+            System.out.printf("当前批次耗时: %.2fs, 预计剩余时间: %s%n",
+                    timePerBatch / 1000,
+                    formatDuration(estimatedRemaining));
+        }
+
+// End timing
+        long endTime = System.currentTimeMillis();
+        System.out.println("处理完成，时间: " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
+        System.out.printf("总耗时: %s%n", formatDuration(endTime - startTime));
+    }
+
+    private static String formatDuration(long millis) {
+        long seconds = millis / 1000;
+        long minutes = seconds / 60;
+        long hours = minutes / 60;
+
+        return String.format("%02d:%02d:%02d",
+                hours,
+                minutes % 60,
+                seconds % 60);
+    }
 }

fs-service/src/main/resources/mapper/course/FsCourseTrafficLogMapper.xml

@@ -16,10 +16,24 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         <result property="companyId"    column="company_id"    />
         <result property="courseId"    column="course_id"    />
         <result property="uuId"    column="uu_id"    />
+        <result property="periodId"    column="period_id"    />
+        <result property="project"    column="project"    />
     </resultMap>
 
     <sql id="selectFsCourseTrafficLogVo">
-        select log_id, uu_id,user_id, video_id, create_time, qw_external_contact_id, internet_traffic, qw_user_id, company_user_id, company_id, course_id from fs_course_traffic_log
+        select log_id,
+               user_id,
+               video_id,
+               create_time,
+               qw_external_contact_id,
+               internet_traffic,
+               qw_user_id,
+               company_user_id,
+               company_id,
+               course_id,
+               uu_id,
+               project,
+               period_id from fs_course_traffic_log
     </sql>
 
     <select id="selectFsCourseTrafficLogList" parameterType="FsCourseTrafficLog" resultMap="FsCourseTrafficLogResult">
@@ -190,4 +204,38 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
             #{logId}
         </foreach>
     </delete>
+
+
+
+    <select id="selectCourseTrafficLogByTwoDaysLater" resultMap="FsCourseTrafficLogResult">
+        <![CDATA[
+        SELECT
+            l.log_id,
+            l.user_id,
+            l.video_id,
+            l.create_time,
+            l.qw_external_contact_id,
+            l.internet_traffic,
+            l.qw_user_id,
+            l.company_user_id,
+            l.company_id,
+            l.course_id,
+            l.uu_id,
+            l.project, l.period_id
+        FROM
+            fs_course_traffic_log l
+        WHERE
+            l.create_time < DATE_SUB( NOW( ), INTERVAL 2 DAY )
+        ORDER BY l.log_id  limit #{offset},#{limit} ]]>
+    </select>
+
+
+    <insert id="insertCourseTrafficLogByTwoDaysLaterBatch" useGeneratedKeys="false">
+        insert into fs_course_traffic_log (
+        log_id, user_id, video_id, create_time, qw_external_contact_id, internet_traffic, qw_user_id, company_user_id, company_id, course_id, uu_id, project, period_id ) values
+        <foreach collection="list" item="item" separator=",">
+            (#{item.logId},#{item.userId},#{item.videoId},#{item.createTime},#{item.qwExternalContactId},#{item.internetTraffic},#{item.qwUserId}
+            ,#{item.companyUserId},#{item.companyId},#{item.courseId},#{item.uuId},#{item.project},#{item.periodId})
+        </foreach>
+    </insert>
 </mapper>