股权交易消息推送

ruoyi-equity/src/main/java/com/ruoyi/equity/controller/EquityMessageRecordController.java

@@ -42,15 +42,8 @@ public class EquityMessageRecordController extends BaseController
 
 
     /**
-     * 获取当前用户未读消息数量
+     * 获取当前用户未读消息数量,之前采用轮询，已经废弃
      */
-    @ApiOperation("获取当前用户未读消息数量")
-    @GetMapping("/getUnreadMessageCount")
-    public AjaxResult getUnreadMessageCount()
-    {
-        Integer unreadMessageCount = equityMessageRecordService.getUnreadMessageCount();
-        return success(unreadMessageCount);
-    }
 
     /**
      * 设置已读

ruoyi-equity/src/main/java/com/ruoyi/equity/event/WebsocketEventListener.java

@@ -4,8 +4,10 @@ import com.ruoyi.equity.service.IEquityMessageRecordService;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import org.springframework.context.ApplicationListener;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
+@Async
 @Component
 @RequiredArgsConstructor
 public class WebsocketEventListener implements ApplicationListener<WebsocketEvent> {

ruoyi-equity/src/main/java/com/ruoyi/equity/service/IEquityMessageRecordService.java

@@ -66,14 +66,13 @@ public interface IEquityMessageRecordService extends IService<EquityMessageRecor
     public int deleteEquityMessageRecordByEquityMessageId(Long equityMessageId);
 
     /**
-     * 获取未读消息
-     * @return
+     * 获取未读消息 登录时获取个人
+     * @param vo
      */
-    Integer getUnreadMessageCount();
+    void getUnreadMessageCount(UserInfoVO vo);
 
     /**
-     * 获取未读消息
-     * @param vo
+     * 获取未读消息 股权变化时，获取所有变动人
      */
-    void getUnreadMessageCount(UserInfoVO vo);
+    void getUnreadMessageCountChanged(List<Long> ids);
 }

ruoyi-equity/src/main/java/com/ruoyi/equity/service/impl/BecomeShareholderRecordServiceImpl.java

@@ -16,10 +16,13 @@ import com.ruoyi.common.utils.DateUtils;
 import com.ruoyi.equity.domain.*;
 import com.ruoyi.equity.domain.dto.BecomeShareholderRecordDTO;
 import com.ruoyi.equity.domain.vo.BecomeShareholderRecordVO;
+import com.ruoyi.equity.domain.vo.UserInfoVO;
 import com.ruoyi.equity.service.IEquityChangeRecordService;
 import com.ruoyi.equity.service.IEquityMessageRecordService;
 import com.ruoyi.equity.service.IShareholderInfoService;
+import com.ruoyi.equity.socket.WebSocketService;
 import com.ruoyi.system.service.ISysUserService;
+import lombok.RequiredArgsConstructor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import com.ruoyi.equity.mapper.BecomeShareholderRecordMapper;
@@ -33,22 +36,20 @@ import org.springframework.transaction.annotation.Transactional;
  * @date 2025-03-04
  */
 @Service
+@RequiredArgsConstructor
 public class BecomeShareholderRecordServiceImpl extends ServiceImpl<BecomeShareholderRecordMapper, BecomeShareholderRecord> implements IBecomeShareholderRecordService
 {
-    @Autowired
-    private BecomeShareholderRecordMapper becomeShareholderRecordMapper;
+    private final BecomeShareholderRecordMapper becomeShareholderRecordMapper;
 
-    @Autowired
-    private IShareholderInfoService shareholderInfoService;
+    private final IShareholderInfoService shareholderInfoService;
 
-    @Autowired
-    private IEquityChangeRecordService equityChangeRecordService;
+    private final IEquityChangeRecordService equityChangeRecordService;
 
-    @Autowired
-    private ISysUserService sysUserService;
+    private final ISysUserService sysUserService;
 
-    @Autowired
-    private IEquityMessageRecordService equityMessageRecordService;
+    private final IEquityMessageRecordService equityMessageRecordService;
+
+    private final WebSocketService webSocketService;
 
     /**
      * 查询股东入股记录
@@ -125,10 +126,24 @@ public class BecomeShareholderRecordServiceImpl extends ServiceImpl<BecomeShareh
                 .changeDate(DateUtils.getNowDate())
                 .changeAfterRatio(become.getShareholdingRatio())
                 .build());
+        //消息通知集合
+        List<EquityMessageRecord> messages = new ArrayList<>();
+        if (CollectionUtils.isNotEmpty(map.get(become.getShareholderId()))) {
+            List<Long> ids = map.get(become.getShareholderId());
+            for (Long id : ids) {
+                EquityMessageRecord message = new EquityMessageRecord();
+                message.setMsgType("1");
+                message.setMsgTime(DateUtils.getNowDate());
+                message.setUserId(id);
+                message.setMsgTitle(String.format("股东【%s】入股公司【%s】通知",shareholderInfo.getShareholderName(),company.getShareholderName()));
+                message.setMsgContent(String.format("股东【%s】的股份占比已由0增长为【%.2f】",
+                        shareholderInfo.getShareholderName(),become.getShareholdingRatio().setScale(2, RoundingMode.HALF_UP)));
+                messages.add(message);
+            }
+
+        }
         //空余股份不够划分，则进行稀释股份
         if (totalRatio.add(become.getShareholdingRatio()).compareTo(new BigDecimal(100)) > 0){
-            //消息通知集合
-            List<EquityMessageRecord> messages = new ArrayList<>();
             //待稀释的股份
             BigDecimal divideShare = totalRatio.add(become.getShareholdingRatio()).subtract(new BigDecimal(100));
             // 计算稀释比例
@@ -152,14 +167,11 @@ public class BecomeShareholderRecordServiceImpl extends ServiceImpl<BecomeShareh
                         message.setMsgType("1");
                         message.setMsgTime(DateUtils.getNowDate());
                         message.setUserId(userId);
-                        message.setMsgTitle(String.format("股东【%s】入股公司【%s】",shareholderInfo.getShareholderName(),company.getShareholderName()));
-//                        message.setMsgContent("股东【" + info.getShareholderName() + "】的股份占比已由【" + info.getShareholdingRatio().setScale(2, RoundingMode.HALF_UP) + "】降低为【" + afterRatio.setScale(2, RoundingMode.HALF_UP) + "】");
+                        message.setMsgTitle(String.format("股东【%s】入股公司【%s】通知",shareholderInfo.getShareholderName(),company.getShareholderName()));
                         message.setMsgContent(String.format("股东【%s】的股份占比已由【%.2f】降低为【%.2f】",
                                 info.getShareholderName(),
                                 info.getShareholdingRatio().setScale(2, RoundingMode.HALF_UP).doubleValue(),
                                 afterRatio.setScale(2, RoundingMode.HALF_UP).doubleValue()));
-
-                        message.setCreateTime(DateUtils.getNowDate());
                         messages.add(message);
                     }
                 }
@@ -172,9 +184,9 @@ public class BecomeShareholderRecordServiceImpl extends ServiceImpl<BecomeShareh
             }
             //批量根据list的主键修改
             shareholderInfoService.updateBatchById(list);
-            //批量保存消息通知
-            equityMessageRecordService.saveBatch(messages);
         }
+        //批量保存消息通知
+        equityMessageRecordService.saveBatch(messages);
         //批量保存变更记录
         equityChangeRecordService.saveBatch(records);
         shareholderInfo.setParentId(company.getShareholderId());
@@ -183,7 +195,12 @@ public class BecomeShareholderRecordServiceImpl extends ServiceImpl<BecomeShareh
         shareholderInfo.setControlledCompany(company.getShareholderId());
         shareholderInfo.setNodeType(company.getNodeType());
         shareholderInfoService.updateShareholderInfo(shareholderInfo);
-        return this.save(become);
+        boolean save = this.save(become);
+        //消息推送
+        if (!messages.isEmpty()) {
+            equityMessageRecordService.getUnreadMessageCountChanged(messages.stream().map(EquityMessageRecord::getUserId).collect(Collectors.toList()));
+        }
+        return save;
     }
 
     /**

ruoyi-equity/src/main/java/com/ruoyi/equity/service/impl/EquityMessageRecordServiceImpl.java

@@ -3,6 +3,8 @@ package com.ruoyi.equity.service.impl;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 import cn.hutool.json.JSONUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
@@ -16,6 +18,7 @@ import com.ruoyi.equity.domain.vo.UserInfoVO;
 import com.ruoyi.equity.socket.WebSocketService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import com.ruoyi.equity.mapper.EquityMessageRecordMapper;
 import com.ruoyi.equity.domain.EquityMessageRecord;
@@ -109,14 +112,6 @@ public class EquityMessageRecordServiceImpl extends ServiceImpl<EquityMessageRec
         return equityMessageRecordMapper.deleteEquityMessageRecordByEquityMessageId(equityMessageId);
     }
 
-    @Override
-    public Integer getUnreadMessageCount() {
-        //获取当前用户id
-        Long userId = SecurityUtils.getUserId();
-        return Math.toIntExact(this.count(new LambdaQueryWrapper<EquityMessageRecord>()
-                .eq(EquityMessageRecord::getUserId, userId)
-                .eq(EquityMessageRecord::getIsRead, "0")));
-    }
     @Override
     public void getUnreadMessageCount(UserInfoVO vo) {
         //获取当前用户未读消息数量
@@ -129,4 +124,25 @@ public class EquityMessageRecordServiceImpl extends ServiceImpl<EquityMessageRec
         webSocketService.sendMessage(vo);
     }
 
+    @Async
+    @Override
+    public void getUnreadMessageCountChanged(List<Long> ids) {
+        //查询出所有未读消息数量，根据用户id
+        List<EquityMessageRecord> records = this.list(new LambdaQueryWrapper<EquityMessageRecord>()
+                .eq(EquityMessageRecord::getIsRead, "0")
+                .in(EquityMessageRecord::getUserId, ids));
+        //按照用户id分组，获取每个用户未读消息数量
+        Map<Long, Long> idMap = records.stream().collect(Collectors.groupingBy(EquityMessageRecord::getUserId, Collectors.counting()));
+        //stream流循环集合，封装为List<UserInfoVO>
+        List<UserInfoVO> list = ids.stream().map(id -> {
+            UserInfoVO vo = new UserInfoVO();
+            vo.setUserId(id);
+            vo.setMsg(JSONUtil.toJsonStr(MsgVO.<Integer>builder().type("msgUnreadCount").data(Math.toIntExact(idMap.get(id))).build()));
+            return vo;
+        }).collect(Collectors.toList());
+        //调用批量消息推送方法
+        webSocketService.sendMessage(list);
+
+    }
+
 }

ruoyi-equity/src/main/java/com/ruoyi/equity/service/impl/EquityTransferRecordServiceImpl.java

@@ -18,10 +18,12 @@ import com.ruoyi.equity.domain.EquityMessageRecord;
 import com.ruoyi.equity.domain.ShareholderInfo;
 import com.ruoyi.equity.domain.dto.EquityTransferRecordDTO;
 import com.ruoyi.equity.domain.vo.EquityTransferRecordVO;
+import com.ruoyi.equity.domain.vo.UserInfoVO;
 import com.ruoyi.equity.mapper.ShareholderInfoMapper;
 import com.ruoyi.equity.service.IEquityChangeRecordService;
 import com.ruoyi.equity.service.IEquityMessageRecordService;
 import com.ruoyi.equity.service.IShareholderInfoService;
+import com.ruoyi.equity.socket.WebSocketService;
 import com.ruoyi.system.service.ISysUserService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -126,7 +128,7 @@ public class EquityTransferRecordServiceImpl extends ServiceImpl<EquityTransferR
                 message.setMsgType("2");
                 message.setMsgTime(DateUtils.getNowDate());
                 message.setUserId(userId);
-                message.setMsgTitle(String.format("股东【%s】与股东【%s】股权交易",seller.getShareholderName(),buyer.getShareholderName()));
+                message.setMsgTitle(String.format("股东【%s】与股东【%s】股权交易通知",seller.getShareholderName(),buyer.getShareholderName()));
                 message.setMsgContent("股东【" + seller.getShareholderName() + "】的股份占比已由【" + seller.getShareholdingRatio().setScale(2, RoundingMode.HALF_UP) + "】降低为【" + sellerAfterRatio.setScale(2, RoundingMode.HALF_UP) + "】");
                 message.setCreateTime(DateUtils.getNowDate());
                 messages.add(message);
@@ -138,7 +140,7 @@ public class EquityTransferRecordServiceImpl extends ServiceImpl<EquityTransferR
                 message.setMsgType("2");
                 message.setMsgTime(DateUtils.getNowDate());
                 message.setUserId(userId);
-                message.setMsgTitle(String.format("股东【%s】与股东【%s】股权交易",seller.getShareholderName(),buyer.getShareholderName()));
+                message.setMsgTitle(String.format("股东【%s】与股东【%s】股权交易通知",seller.getShareholderName(),buyer.getShareholderName()));
                 message.setMsgContent("股东【" + buyer.getShareholderName() + "】的股份占比已由【" + buyer.getShareholdingRatio().setScale(2, RoundingMode.HALF_UP) + "】提高为【" + buyerAfterRatio.setScale(2, RoundingMode.HALF_UP) + "】");
                 message.setCreateTime(DateUtils.getNowDate());
                 messages.add(message);
@@ -176,9 +178,12 @@ public class EquityTransferRecordServiceImpl extends ServiceImpl<EquityTransferR
         //修改数据库股东信息
         shareholderInfoService.updateById(seller);
         shareholderInfoService.updateById(buyer);
-
-        equityTransferRecord.setCreateTime(DateUtils.getNowDate());
-        return this.save(equityTransferRecord);
+        boolean save = this.save(equityTransferRecord);
+        //消息推送
+        if (!messages.isEmpty()) {
+            equityMessageRecordService.getUnreadMessageCountChanged(messages.stream().map(EquityMessageRecord::getUserId).collect(Collectors.toList()));
+        }
+        return save;
     }
 
     /**

ruoyi-equity/src/main/java/com/ruoyi/equity/socket/WebSocketService.java

@@ -134,9 +134,13 @@ public class WebSocketService implements ApplicationContextAware {
         error.printStackTrace();
     }
 
+    /**
+     * 单个消息推送
+     * @param vo
+     */
     public void sendMessage(UserInfoVO vo){
         List<Session> sessions = userSessionMap.get(vo.getUserId());
-        if(sessions != null && sessions.size() > 0){
+        if(sessions != null && !sessions.isEmpty()){
             /** 此处不能使用普通集合，否则在推送消息时，当前用户又开启一个页面，集合发生变化迭代器会报错ConcurrentModificationException*/
             for (Session s : sessions) {
                 if (s == null){
@@ -156,6 +160,30 @@ public class WebSocketService implements ApplicationContextAware {
         }
     }
 
+    /**
+     * 批量消息推送
+     */
+    public void sendMessage(List<UserInfoVO> vos){
+        for (UserInfoVO vo : vos) {
+            List<Session> sessions = userSessionMap.get(vo.getUserId());
+            if (sessions != null && !sessions.isEmpty()) {
+                for (Session s : sessions) {
+                    if (s == null) {
+                        continue;
+                    }
+                    RemoteEndpoint.Basic basicRemote = s.getBasicRemote();
+                    if (basicRemote != null) {
+                        try {
+                            basicRemote.sendText(vo.getMsg());
+                        } catch (IOException e) {
+                            log.error(vo.getUserId() + "推送消息失败,消息内容[" + vo.getMsg() + "],原因:" + e.getMessage());
+                        }
+                    }
+                }
+            }
+        }
+    }
+
     public static synchronized void addOnlineUserCount() {
         WebSocketService.onlineUserCount++;
     }