每五秒从redis中读取已读的消息id 并将数据库设为已读

alien-store/src/main/java/shop/alien/store/task/ScheduledTask.java

@@ -1,15 +1,20 @@
 package shop.alien.store.task;
 
+import cn.hutool.core.collection.CollectionUtil;
+import com.alibaba.fastjson2.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import shop.alien.entity.second.SecondTradeRecord;
 import shop.alien.entity.store.*;
-import shop.alien.mapper.LifeCouponMapper;
-import shop.alien.mapper.LifeMessageMapper;
-import shop.alien.mapper.LifeUserOrderMapper;
+import shop.alien.entity.store.vo.WebsocketVo;
+import shop.alien.mapper.*;
+import shop.alien.mapper.second.SecondTradeRecordMapper;
+import shop.alien.store.config.BaseRedisService;
+import shop.alien.store.config.WebSocketProcess;
 import shop.alien.store.service.*;
 import shop.alien.util.common.AlipayTradeRefund;
 import shop.alien.util.common.DateUtils;
@@ -17,6 +22,8 @@ import shop.alien.store.util.ali.AliApi;
 
 import java.math.BigDecimal;
 import java.math.RoundingMode;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.util.*;
 import java.util.stream.Collectors;
 
@@ -45,6 +52,16 @@ public class ScheduledTask {
 
     private final LifeUserService lifeUserService;
 
+    private final SecondTradeRecordMapper secondTradeRecordMapper;
+
+    private final LifeNoticeMapper lifeNoticeMapper;
+
+    private final LifeUserMapper lifeUserMapper;
+
+    private final WebSocketProcess webSocketProcess;
+
+    private final BaseRedisService baseRedisService;
+
     @Scheduled(cron = "0 */10 * * * ?")
     public void sendDiscountCouponRemind() {
         //查询
@@ -176,7 +193,7 @@ public class ScheduledTask {
      * 自动转账
      * 一分钟执行一次
      */
-//    @Scheduled(cron = "0 */1 * * * ?")
+    @Scheduled(cron = "0 */1 * * * ?")
     public void autoTransferAccounts() {
         //获取已到账期
         LambdaQueryWrapper<StoreIncomeDetailsRecord> wrapper = new LambdaQueryWrapper<>();
@@ -274,4 +291,103 @@ public class ScheduledTask {
             }
         }
     }
+
+    /**
+     * 二手交易平台 - 交易时间前的十分钟和十五分钟之间  给买家和卖家发送通知提醒
+     */
+    @Scheduled(cron = "0 */5 * * * ?")
+    public void secondTradeRemind() throws Exception {
+        LocalDateTime now = LocalDateTime.now();
+        // 五分钟后
+        LocalDateTime fiveMinutesLater = now.plusMinutes(5);
+
+        // 查询所有待交易
+        LambdaQueryWrapper<SecondTradeRecord> wrapper = new LambdaQueryWrapper<>();
+        wrapper.eq(SecondTradeRecord::getTradeStatus, 0);
+        List<SecondTradeRecord> tradeRecordList = secondTradeRecordMapper.selectList(wrapper);
+
+        for (SecondTradeRecord tradeRecord : tradeRecordList) {
+            // 十分钟前
+            LocalDateTime tenMinutesAgo = tradeRecord.getTransactionTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime().minusMinutes(10);
+            // 交易时间前的十分钟和十五分钟之间
+            if ((now.isBefore(tenMinutesAgo) && fiveMinutesLater.isAfter(tenMinutesAgo))
+                    || now.isEqual(tenMinutesAgo) || fiveMinutesLater.isEqual(tenMinutesAgo)) {
+
+                // 给买家发送消息
+                LifeMessage lifeMessage = new LifeMessage();
+                lifeMessage.setSenderId(String.valueOf(tradeRecord.getSellerId()));
+                lifeMessage.setReceiverId(String.valueOf(tradeRecord.getBuyerId()));
+                lifeMessage.setContent("签到");
+                lifeMessage.setType("5");
+                lifeMessageMapper.insert(lifeMessage);
+
+                // 给卖家发送消息
+                lifeMessage = new LifeMessage();
+                lifeMessage.setSenderId(String.valueOf(tradeRecord.getBuyerId()));
+                lifeMessage.setReceiverId(String.valueOf(tradeRecord.getSellerId()));
+                lifeMessage.setContent("签到");
+                lifeMessage.setType("5");
+                lifeMessageMapper.insert(lifeMessage);
+
+                // 给买家发送通知
+                LifeNotice lifeNotice = new LifeNotice();
+                lifeNotice.setSenderId("system");
+                lifeNotice.setReceiverId(String.valueOf(tradeRecord.getBuyerId()));
+                lifeNotice.setBusinessId(tradeRecord.getId());
+                lifeNotice.setTitle("交易提醒");
+                lifeNotice.setContext("您有一笔交易即将开始, 请及时前往查看");
+                lifeNotice.setNoticeType(1);
+                lifeNoticeMapper.insert(lifeNotice);
+
+                // 给卖家发送通知
+                lifeNotice = new LifeNotice();
+                lifeNotice.setSenderId("system");
+                lifeNotice.setReceiverId(String.valueOf(tradeRecord.getSellerId()));
+                lifeNotice.setBusinessId(tradeRecord.getId());
+                lifeNotice.setTitle("交易提醒");
+                lifeNotice.setContext("您有一笔交易即将开始, 请及时前往查看");
+                lifeNotice.setNoticeType(1);
+                lifeNoticeMapper.insert(lifeNotice);
+
+                // 给买家推送通知
+                LifeUser user = lifeUserMapper.selectById(tradeRecord.getBuyerId());
+                WebsocketVo websocketVo = new WebsocketVo();
+                websocketVo.setSenderId("system");
+                websocketVo.setCategory("notice");
+                websocketVo.setType("5");
+                websocketVo.setText("签到");
+                if (user != null) {
+                    websocketVo.setReceiverId("user_" + user.getUserPhone());
+                    webSocketProcess.sendMessage("user_" + user.getUserPhone(), JSONObject.from(websocketVo).toJSONString());
+                }
+
+                // 给卖家推送通知
+                user = lifeUserMapper.selectById(tradeRecord.getSellerId());
+                if (user != null) {
+                    websocketVo.setReceiverId("user_" + user.getUserPhone());
+                    webSocketProcess.sendMessage("user_" + user.getUserPhone(), JSONObject.from(websocketVo).toJSONString());
+                }
+            }
+        }
+    }
+
+    /**
+     * 每五秒从redis中读取已读的消息id 并将数据库设为已读
+     */
+    @Scheduled(cron = "*/5 * * * * ?")
+    public void readMessage() {
+        try {
+            if (CollectionUtil.isEmpty(baseRedisService.getList("readMessageIdKey"))) return;
+
+            List<String> dataList = baseRedisService.popBatchFromList("readMessageIdKey");
+            if (CollectionUtil.isNotEmpty(dataList)) {
+                LambdaUpdateWrapper<LifeMessage> wrapper = new LambdaUpdateWrapper<>();
+                wrapper.in(LifeMessage::getId, dataList)
+                        .set(LifeMessage::getIsRead, 1);
+                lifeMessageMapper.update(null, wrapper);
+            }
+        } catch (Exception e) {
+            log.error("ScheduledTask.readMessage Error Mgs={}", e.getMessage());
+        }
+    }
 }