增加定时任务

api-module/src/main/java/com/tzld/piaoquan/api/mq/MessageCallbackCustomer.java

@@ -7,10 +7,10 @@ import com.aliyun.openservices.ons.api.Action;
 import com.aliyun.openservices.ons.api.ConsumeContext;
 import com.aliyun.openservices.ons.api.Message;
 import com.aliyun.openservices.ons.api.MessageListener;
-import com.tzld.piaoquan.api.dao.mapper.NewPushMessageCallbackMapper;
 import com.tzld.piaoquan.api.model.bo.ReplyInfo;
-import com.tzld.piaoquan.api.model.po.PushMessageCallback;
 import com.tzld.piaoquan.api.model.vo.CallbackParam;
+import com.tzld.piaoquan.growth.common.dao.mapper.PushMessageCallbackMapper;
+import com.tzld.piaoquan.growth.common.model.po.PushMessageCallback;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -26,7 +26,7 @@ public class MessageCallbackCustomer implements MessageListener {
 
 
     @Autowired
-    private NewPushMessageCallbackMapper newPushMessageCallbackMapper;
+    private PushMessageCallbackMapper pushMessageCallbackMapper;
 
     @Override
     public Action consume(Message message, ConsumeContext consumeContext) {
@@ -48,7 +48,7 @@ public class MessageCallbackCustomer implements MessageListener {
             insertList.add(insertPushMessageCallback);
         }
         try {
-            newPushMessageCallbackMapper.insertList(insertList);
+            pushMessageCallbackMapper.insertList(insertList);
         } catch (Exception e) {
             log.error("PushMessageCallback newPushMessageCallbackMapper insert pushMessageCallback={}, error={}", pushMessageCallback, e.getMessage());
             return Action.ReconsumeLater;

common-module/src/main/java/com/tzld/piaoquan/growth/common/common/constant/TimeConstant.java

@@ -6,6 +6,8 @@ public interface TimeConstant {
     Integer HOUR = 60 * 60;
     Integer DAY = 60 * 60 * 24;
 
+    Long MILLISECOND_HOUR = (long) HOUR * 1000;
+
     Long MILLISECOND_DAY = (long) DAY * 1000;
 
 }

api-module/src/main/java/com/tzld/piaoquan/api/dao/mapper/NewPushMessageCallbackMapper.java → common-module/src/main/java/com/tzld/piaoquan/growth/common/dao/mapper/PushMessageCallbackMapper.java

@@ -1,15 +1,17 @@
-package com.tzld.piaoquan.api.dao.mapper;
+package com.tzld.piaoquan.growth.common.dao.mapper;
 
 
-import com.tzld.piaoquan.api.model.po.PushMessageCallback;
-import com.tzld.piaoquan.api.model.po.PushMessageCallbackExample;
+import com.tzld.piaoquan.growth.common.model.po.PushMessageCallback;
+import com.tzld.piaoquan.growth.common.model.po.PushMessageCallbackExample;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
+import org.springframework.stereotype.Repository;
 
 import java.util.List;
 
 @Mapper
-public interface NewPushMessageCallbackMapper {
+@Repository
+public interface PushMessageCallbackMapper {
     long countByExample(PushMessageCallbackExample example);
 
     int deleteByExample(PushMessageCallbackExample example);

api-module/src/main/java/com/tzld/piaoquan/api/model/po/PushMessageCallback.java → common-module/src/main/java/com/tzld/piaoquan/growth/common/model/po/PushMessageCallback.java

@@ -1,4 +1,4 @@
-package com.tzld.piaoquan.api.model.po;
+package com.tzld.piaoquan.growth.common.model.po;
 
 import java.util.Date;
 

api-module/src/main/java/com/tzld/piaoquan/api/model/po/PushMessageCallbackExample.java → common-module/src/main/java/com/tzld/piaoquan/growth/common/model/po/PushMessageCallbackExample.java

@@ -1,4 +1,4 @@
-package com.tzld.piaoquan.api.model.po;
+package com.tzld.piaoquan.growth.common.model.po;
 
 import java.util.ArrayList;
 import java.util.Date;

common-module/src/main/java/com/tzld/piaoquan/growth/common/utils/DateUtil.java

@@ -9,6 +9,7 @@ import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeParseException;
 import java.util.Calendar;
 import java.util.Date;
+import java.util.TimeZone;
 
 public class DateUtil {
 
@@ -115,4 +116,22 @@ public class DateUtil {
         return dateTime.toEpochSecond(zone.getRules().getOffset(dateTime));
     }
 
+    public static Long getBeforeDayStart(int days) {
+        Calendar calendar = Calendar.getInstance();
+        calendar.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
+        calendar.set(Calendar.HOUR_OF_DAY, 0);
+        calendar.set(Calendar.MINUTE, 0);
+        calendar.set(Calendar.SECOND, 0);
+        calendar.set(Calendar.MILLISECOND, 0);
+        calendar.add(Calendar.DAY_OF_MONTH, -days);
+        return calendar.getTime().getTime() / 1000;
+    }
+
+    public static Date getDate(String dateString) {
+        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
+        LocalDate localDate = LocalDate.parse(dateString, formatter);
+        ZoneId zone = ZoneId.of("Asia/Shanghai");
+        return Date.from(localDate.atStartOfDay(zone).toInstant());
+    }
+
 }

api-module/src/main/resources/mapper/NewPushMessageCallbackMapper.xml → common-module/src/main/resources/mapper/PushMessageCallbackMapper.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
-<mapper namespace="com.tzld.piaoquan.api.dao.mapper.NewPushMessageCallbackMapper">
-    <resultMap id="BaseResultMap" type="com.tzld.piaoquan.api.model.po.PushMessageCallback">
+<mapper namespace="com.tzld.piaoquan.growth.common.dao.mapper.PushMessageCallbackMapper">
+    <resultMap id="BaseResultMap" type="com.tzld.piaoquan.growth.common.model.po.PushMessageCallback">
         <id column="id" jdbcType="BIGINT" property="id"/>
         <result column="gh_id" jdbcType="VARCHAR" property="ghId"/>
         <result column="open_id" jdbcType="VARCHAR" property="openId"/>
@@ -75,7 +75,7 @@
         , gh_id, open_id, timestamp, msg_type, video_id, create_time
     </sql>
     <select id="selectByExample"
-            parameterType="com.tzld.piaoquan.api.model.po.PushMessageCallbackExample"
+            parameterType="com.tzld.piaoquan.growth.common.model.po.PushMessageCallbackExample"
             resultMap="BaseResultMap">
         select
         <if test="distinct">
@@ -102,13 +102,13 @@
         where id = #{id,jdbcType=BIGINT}
     </delete>
     <delete id="deleteByExample"
-            parameterType="com.tzld.piaoquan.api.model.po.PushMessageCallbackExample">
+            parameterType="com.tzld.piaoquan.growth.common.model.po.PushMessageCallbackExample">
         delete from push_message_callback
         <if test="_parameter != null">
             <include refid="Example_Where_Clause"/>
         </if>
     </delete>
-    <insert id="insert" parameterType="com.tzld.piaoquan.api.model.po.PushMessageCallback">
+    <insert id="insert" parameterType="com.tzld.piaoquan.growth.common.model.po.PushMessageCallback">
         <selectKey keyProperty="id" order="AFTER" resultType="java.lang.Long">
             SELECT LAST_INSERT_ID()
         </selectKey>
@@ -120,7 +120,7 @@
         )
     </insert>
     <insert id="insertSelective"
-            parameterType="com.tzld.piaoquan.api.model.po.PushMessageCallback">
+            parameterType="com.tzld.piaoquan.growth.common.model.po.PushMessageCallback">
         <selectKey keyProperty="id" order="AFTER" resultType="java.lang.Long">
             SELECT LAST_INSERT_ID()
         </selectKey>
@@ -167,7 +167,7 @@
         </trim>
     </insert>
     <select id="countByExample"
-            parameterType="com.tzld.piaoquan.api.model.po.PushMessageCallbackExample"
+            parameterType="com.tzld.piaoquan.growth.common.model.po.PushMessageCallbackExample"
             resultType="java.lang.Long">
         select count(*) from push_message_callback
         <if test="_parameter != null">
@@ -217,7 +217,7 @@
         </if>
     </update>
     <update id="updateByPrimaryKeySelective"
-            parameterType="com.tzld.piaoquan.api.model.po.PushMessageCallback">
+            parameterType="com.tzld.piaoquan.growth.common.model.po.PushMessageCallback">
         update push_message_callback
         <set>
             <if test="ghId != null">
@@ -242,7 +242,7 @@
         where id = #{id,jdbcType=BIGINT}
     </update>
     <update id="updateByPrimaryKey"
-            parameterType="com.tzld.piaoquan.api.model.po.PushMessageCallback">
+            parameterType="com.tzld.piaoquan.growth.common.model.po.PushMessageCallback">
         update push_message_callback
         set gh_id       = #{ghId,jdbcType=VARCHAR},
             open_id     = #{openId,jdbcType=VARCHAR},

offline-module/src/main/java/com/tzld/piaoquan/offline/job/PushMessageDataJob.java

@@ -0,0 +1,104 @@
+package com.tzld.piaoquan.offline.job;
+
+import com.aliyun.odps.data.Record;
+import com.tzld.piaoquan.growth.common.component.ODPSManager;
+import com.tzld.piaoquan.growth.common.dao.mapper.PushMessageCallbackMapper;
+import com.tzld.piaoquan.growth.common.model.po.PushMessageCallbackExample;
+import com.tzld.piaoquan.growth.common.utils.DateUtil;
+import com.tzld.piaoquan.growth.common.utils.LarkRobotUtil;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+import java.util.Calendar;
+import java.util.Date;
+import java.util.List;
+
+import static com.tzld.piaoquan.growth.common.common.constant.TimeConstant.MILLISECOND_DAY;
+import static com.tzld.piaoquan.growth.common.common.constant.TimeConstant.MILLISECOND_HOUR;
+
+@Slf4j
+@Component
+public class PushMessageDataJob {
+
+    @Autowired
+    private PushMessageCallbackMapper pushMessageCallbackMapper;
+
+    @Autowired
+    private ODPSManager odpsManager;
+
+    @XxlJob("alertPushMessageJob")
+    public ReturnT<String> alertPushMessageJob(String param) {
+        long endTimestamp = System.currentTimeMillis();
+        long startTimestamp = endTimestamp - MILLISECOND_HOUR;
+        long ytdStartTimestamp = startTimestamp - MILLISECOND_DAY;
+        long ytdEndTimestamp = endTimestamp - MILLISECOND_DAY;
+        PushMessageCallbackExample example = new PushMessageCallbackExample();
+        example.createCriteria().andCreateTimeBetween(new Date(startTimestamp), new Date(endTimestamp));
+        long nowCount = pushMessageCallbackMapper.countByExample(example);
+        if (nowCount == 0) {
+            LarkRobotUtil.sendMessage("最近一小时回调消息为0");
+        }
+        PushMessageCallbackExample ytdExample = new PushMessageCallbackExample();
+        ytdExample.createCriteria().andCreateTimeBetween(new Date(ytdStartTimestamp), new Date(ytdEndTimestamp));
+        long ydtCount = pushMessageCallbackMapper.countByExample(example);
+        if (nowCount + 1000 < ydtCount) {
+            if (ydtCount == 0) {
+                ydtCount = 1L;
+            }
+            double res = ((double) nowCount - ydtCount) * 100 / ydtCount;
+            LarkRobotUtil.sendMessage("最近一小时回调消息量：" + nowCount + "\n"
+                    + "昨日消息回调量：" + ydtCount + "\n"
+                    + "百分比为：" + String.format("%.2f", res) + "%");
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    @XxlJob("delPushMessageCallbackJob")
+    public ReturnT<String> delPushMessageCallbackJob(String param) {
+        Calendar calendar = Calendar.getInstance();
+        Long beforeDayStart = DateUtil.getBeforeDayStart(7);
+        Date startDay = new Date(beforeDayStart * 1000);
+        if (StringUtils.hasText(param)) {
+            startDay = DateUtil.getDate(param);
+        }
+        for (int i = 0; i < 30; i++) {
+            // 计算当前遍历日期
+            Date currentStartDate = (Date) startDay.clone();
+            calendar.setTime(currentStartDate);
+            calendar.add(Calendar.DAY_OF_MONTH, -i); // 向前推i天
+
+            // 生成开始时间和结束时间
+            Date startTime = calendar.getTime(); // 开始时间
+            calendar.set(Calendar.HOUR_OF_DAY, 23);
+            calendar.set(Calendar.MINUTE, 59);
+            calendar.set(Calendar.SECOND, 59);
+            Date endTime = calendar.getTime(); // 结束时间
+
+            PushMessageCallbackExample example = new PushMessageCallbackExample();
+            example.createCriteria().andCreateTimeBetween(startTime, endTime);
+            long mysqlCount = pushMessageCallbackMapper.countByExample(example);
+            if (mysqlCount == 0) {
+                break;
+            }
+            String pt = DateUtil.getDateString(startTime.getTime());
+            String sql = String.format("SELECT count(*) FROM push_message_callback WHERE pt = %s;", pt);
+            List<Record> recordList = odpsManager.query(sql);
+            if (CollectionUtils.isEmpty(recordList)) {
+                LarkRobotUtil.sendMessage("查询hive失败" + pt);
+                return ReturnT.FAIL;
+            }
+            Long hiveCount = recordList.get(0).getBigint(0);
+            if (Math.abs(mysqlCount - hiveCount) > 0) {
+                LarkRobotUtil.sendMessage("数量异常" + pt + "\n mysql数量：" + mysqlCount + "\n hive数量：" + hiveCount);
+                return ReturnT.FAIL;
+            }
+            pushMessageCallbackMapper.deleteByExample(example);
+        }
+        return ReturnT.SUCCESS;
+    }
+}