add cost

src/main/scala/com.gmei.up/Main.scala

@@ -47,7 +47,7 @@ object Main extends LazyLogging {
 
       // TODO remove test
       val kafkaConsumer = new FlinkKafkaConsumer[UserInfo](
-        "gm-portrait-update-device-test",
+        "gm-portrait-update-device",
         new UserInfoDeserializationSchema,
         kafkaConsumerProperties
       )
@@ -110,20 +110,21 @@ object Main extends LazyLogging {
           // current time
           logger.info(s"${user.eventCn} ${deviceId}")
 
-          if (deviceId == "64695DE0-B926-4188-9C62-D987DC20BEDF") {
+          val cost = streamTimeBegin - user.sendTimestamp
+          if (cost > 1000) {
+            deviceId == "64695DE0-B926-4188-9C62-D987DC20BEDF"
             DingTalk.send(
               Map(
                 "deviceId" -> deviceId,
                 "eventCn" -> user.eventCn,
                 "action" -> user.action,
                 "cityId" -> user.cityId.toString,
-                "logTime" -> user.logTime.toString,
                 "projects" -> user.projects.mkString(" "),
                 "diaryRead" -> diaryRead.size.toString,
                 "tractateRead" -> tractateRead.size.toString,
                 "answerRead" -> answerRead.size.toString,
-                "cost" -> s"${streamTimeBegin - user.sendTimestamp}ms",
-                "description" -> "收到请求到开始处理的时间"
+                "cost" -> s"${cost}ms",
+                "description" -> "收到请求到开始处理的耗时"
               )
             )
           }

src/main/scala/com.gmei.up/utils/Redis.scala

@@ -50,7 +50,7 @@ object Redis {
                     "method" -> "Redis.save",
                     "deviceId" -> deviceId,
                     "contentType" -> contentType,
-                    "description" -> "该设备从流开始处理到存入结果的总时间",
+                    "description" -> "从流开始处理到存入结果的总耗时",
                     "streamTotalSeconds" -> s"${total}ms"
                   )
                 )