Skip to content

W
wecloud_im_server
Commit fc359073 by Future
Options

保存消息逻辑

parent 81c41b05
Showing with 118 additions and 126 deletions
core/src/main/java/com/wecloud/im/service/ImMessageService.java
......@@ -180,10 +180,4 @@ public interface ImMessageService extends BaseService<ImMessage> {
List<WeCloudMessageVo> syncListMessage(SyncListMessageParam param);
/**
* 保存消息到数据库
* @param imMessage
*/
void saveMessageToDb(ImMessage imMessage);
}
core/src/main/java/com/wecloud/im/service/SaveMessageService.java 0 → 100644
package com.wecloud.im.service;
import com.baomidou.lock.annotation.Lock4j;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.wecloud.im.entity.ImClient;
import com.wecloud.im.entity.ImConversationMembers;
import com.wecloud.im.entity.ImMessage;
import com.wecloud.im.param.ImConversationQueryVo;
import com.wecloud.im.ws.utils.RedisUtils;
import com.wecloud.utils.SnowflakeUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.util.Date;
import java.util.List;
/**
* @Author Future
* @Date 2022/8/24 17:18
* @Description 本地缓存服务
*/
@Slf4j
@Service
public class SaveMessageService {
private static final String LAST_MSG_PREFIX = "last_msg_";
@Autowired
private ImMessageService imMessageService;
@Autowired
private RedisUtils redisUtils;
/**
* 持久化消息到数据库并更新相关缓存
* expire 锁过期时间 防止死锁
*
* @param imMessage
*/
@Lock4j(keys = {"#imMessage.fkConversationId"}, expire = 3000, acquireTimeout = 1000)
public void saveMessageToDb(ImMessage imMessage) {
imMessage.setId(SnowflakeUtil.getId());
imMessage.setCreateTime(new Date());
this.setLastMessageIdAndFreshCash(imMessage);
imMessageService.save(imMessage);
}
/**
* 查找会话的最新一条消息
*
* @param message
* @return
*/
private void setLastMessageIdAndFreshCash(ImMessage message) {
try {
// 执行业务方法
String key = LAST_MSG_PREFIX + message.getFkConversationId();
String messageId = redisUtils.getKey(key);
if (StringUtils.isNotBlank(messageId)) {
// 将Redis中最新的messageId存入
message.setPreMessageId(Long.valueOf(messageId));
} else {
// 落库查
ImMessage preMessage = imMessageService.getOne(new QueryWrapper<ImMessage>().lambda()
.eq(ImMessage::getFkConversationId, message.getFkConversationId())
.orderByDesc(ImMessage::getId)
.last("limit 1")
);
if (preMessage != null) {
message.setPreMessageId(preMessage.getId());
}
}
redisUtils.addKey(key, message.getId().toString(), Duration.ofMinutes(30));
} catch (Exception e) {
log.info("获取会话最后一条消息异常 ", e);
}
}
}
core/src/main/java/com/wecloud/im/service/impl/ImConversationServiceImpl.java
......@@ -62,6 +62,7 @@ import com.wecloud.im.service.ImConversationMembersService;
import com.wecloud.im.service.ImConversationService;
import com.wecloud.im.service.ImInboxService;
import com.wecloud.im.service.ImMessageService;
import com.wecloud.im.service.SaveMessageService;
import com.wecloud.im.vo.ChatRoomMemberVo;
import com.wecloud.im.vo.ConversationCountVo;
import com.wecloud.im.vo.ConversationMemberVo;
......@@ -162,6 +163,9 @@ public class ImConversationServiceImpl extends BaseServiceImpl<ImConversationMap
@Resource
private EhcacheService ehcacheService;
@Autowired
private SaveMessageService saveMessageService;
@Override
@Transactional(rollbackFor = Exception.class)
public ImConversationCreateVo createImConversation(ImConversationCreate imConversationCreate) {
......@@ -341,7 +345,7 @@ public class ImConversationServiceImpl extends BaseServiceImpl<ImConversationMap
content.put("operator", imClientSender.getClientId()); //操作的client ID
content.put("passivityOperator", clientToConversation.getClientId()); //被操作的client ID
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.INVITE_CLIENT_JOIN_CONVERSATION, imApplication, createClient, imConversation, JsonUtils.encodeJson(content));
boolean save = imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
// 发送给在群内的成员
sendMsgToMembers(imConversation, membersList, createClient, imMessage, content);
// 发送给被邀请人
......@@ -403,7 +407,7 @@ public class ImConversationServiceImpl extends BaseServiceImpl<ImConversationMap
content.put("passivityOperator", clientToBeRemove.getClientId()); //被操作的client ID
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.REMOVE_CLIENT_CONVERSATION, imApplication, createClient, imConversation, JsonUtils.encodeJson(content));
imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
membersList.removeIf(e -> e.getId().equals(members.getId()));
......@@ -451,7 +455,7 @@ public class ImConversationServiceImpl extends BaseServiceImpl<ImConversationMap
Map<String, Object> content = new HashMap<>();
content.put("operator", currentClient.getClientId());
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.CONVERSATION_DISBAND, imApplication, currentClient, imConversation, JsonUtils.encodeJson(content));
imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
sendMsgToMembers(imConversation, membersList, currentClient, imMessage, content);
}
......@@ -527,11 +531,7 @@ public class ImConversationServiceImpl extends BaseServiceImpl<ImConversationMap
// 生成消息id
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.LEAVE_CONVERSATION, imApplication, currentClient, imConversation, "");
// 保存消息至消息表
boolean save = imMessageService.save(imMessage);
if (!save) {
throw new BusinessException("退出群聊错误");
}
saveMessageService.saveMessageToDb(imMessage);
sendMsgToMembers(imConversation, membersList, currentClient, imMessage, null);
// 群主退出 转移给下一个人
if (GroupRoleEnum.OWNER.getCode().equals(members.getRole())) {
......@@ -1022,7 +1022,7 @@ public class ImConversationServiceImpl extends BaseServiceImpl<ImConversationMap
if (!membersList.isEmpty()) {
// 保存事件消息
ImMessage imMessage = MessageBuilder.buildEventMessage(msgType, application, operatorClient, group, JsonUtils.encodeJson(content));
imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
// 发送消息至群成员
sendMsgToMembers(group, membersList, operatorClient, imMessage, content);
}
......@@ -1541,7 +1541,7 @@ public class ImConversationServiceImpl extends BaseServiceImpl<ImConversationMap
content.put("operator", creator.getClientId()); //操作的client ID
content.put("passivityOperator", member.getClientId()); //被操作的client ID
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.CLIENT_JOIN_NEW_CONVERSATION, application, creator, conversation, JsonUtils.encodeJson(content));
imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
// 投递消息
couriers.deliver(imMessage, content, creator, member, WsResponseCmdEnum.CONVERSATION_EVENT_MSG);
}
......
core/src/main/java/com/wecloud/im/service/impl/ImGroupServiceImpl.java
package com.wecloud.im.service.impl;
import com.wecloud.im.service.SaveMessageService;
import io.geekidea.springbootplus.framework.common.exception.BusinessException;
import io.geekidea.springbootplus.framework.shiro.util.SecurityUtils;
import lombok.AllArgsConstructor;
......@@ -12,6 +13,7 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
......@@ -86,10 +88,8 @@ public class ImGroupServiceImpl implements ImGroupService {
*/
private final ImConversationMapper imConversationMapper;
/**
* 消息服务
*/
private final ImMessageService imMessageService;
@Autowired
private SaveMessageService saveMessageService;
@Override
public Long createGroup(String creatorClientId, String groupName, List<String> memberIds) {
......@@ -163,7 +163,7 @@ public class ImGroupServiceImpl implements ImGroupService {
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.CONVERSATION_DISBAND, application,
operator, conversation, JsonUtils.encodeJson(content));
// 保存消息至消息表
imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
conversationService.sendMsgToMembers(conversation, membersList, operator, imMessage, content);
return true;
......@@ -236,7 +236,7 @@ public class ImGroupServiceImpl implements ImGroupService {
content.put("passivityOperator", newMember.getClientId());
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.INVITE_CLIENT_JOIN_CONVERSATION,
imApplication, inviter, conversation, JsonUtils.encodeJson(content));
boolean save = imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
// 发送给在群内的成员
conversationService.sendMsgToMembers(conversation, oldMembers, inviter, imMessage, content);
// 发送给被邀请人
......@@ -286,7 +286,7 @@ public class ImGroupServiceImpl implements ImGroupService {
content.put("operator", operator.getClientId());
// 自己主动退出
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.LEAVE_CONVERSATION, imApplication, operator, conversation, JsonUtils.encodeJson(content));
imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
// 发送给在群内的成员
conversationService.sendMsgToMembers(conversation, membersList, operator, imMessage, content);
} else {
......@@ -294,7 +294,7 @@ public class ImGroupServiceImpl implements ImGroupService {
// 被操作的client ID
content.put("passivityOperator", members.getClientId());
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.REMOVE_CLIENT_CONVERSATION, imApplication, operator, conversation, JsonUtils.encodeJson(content));
imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
// 发送给在群内的成员
conversationService.sendMsgToMembers(conversation, membersList, operator, imMessage, content);
}
......@@ -394,7 +394,7 @@ public class ImGroupServiceImpl implements ImGroupService {
Map<String, Object> content = new HashMap<>();
content.put("operator", newGroupOwner.getClientId());
ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.CONVERSATION_NEW_CREATOR, imApplication, newGroupOwnerClient, conversation, JsonUtils.encodeJson(content));
imMessageService.save(imMessage);
saveMessageService.saveMessageToDb(imMessage);
// 发送给在群内的成员
List<ImConversationMembers> existMemberList =
conversationMembersService.list(Wrappers.<ImConversationMembers>lambdaQuery()
......
core/src/main/java/com/wecloud/im/service/impl/ImMessageServiceImpl.java
package com.wecloud.im.service.impl;
import com.wecloud.im.entity.ImClientBlacklist;
import com.wecloud.im.param.ImConversationQueryVo;
import com.wecloud.im.param.SyncListMessageParam;
import com.wecloud.im.param.WeCloudMessageVo;
import com.wecloud.im.post.MessageBuilder;
import io.geekidea.springbootplus.framework.common.api.ApiCode;
import io.geekidea.springbootplus.framework.common.api.ApiResult;
import io.geekidea.springbootplus.framework.common.exception.BusinessException;
import io.geekidea.springbootplus.framework.common.service.impl.BaseServiceImpl;
import io.geekidea.springbootplus.framework.core.pagination.PageInfo;
import io.geekidea.springbootplus.framework.core.pagination.Paging;
import io.geekidea.springbootplus.framework.shiro.util.SecurityUtils;
import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.lock.annotation.Lock4j;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.metadata.OrderItem;
......@@ -69,6 +32,8 @@ import com.wecloud.im.param.PrivateChatMessageParam;
import com.wecloud.im.param.PrivateChatStatusMessageParam;
import com.wecloud.im.param.PushExtParam;
import com.wecloud.im.param.SingleUserNotificationParam;
import com.wecloud.im.param.SyncListMessageParam;
import com.wecloud.im.param.WeCloudMessageVo;
import com.wecloud.im.param.add.ImMsgRecall;
import com.wecloud.im.param.add.ImMsgSendToOnlineClient;
import com.wecloud.im.param.add.ImMsgUpdate;
......@@ -82,6 +47,7 @@ import com.wecloud.im.service.ImConversationMembersService;
import com.wecloud.im.service.ImConversationService;
import com.wecloud.im.service.ImInboxService;
import com.wecloud.im.service.ImMessageService;
import com.wecloud.im.service.SaveMessageService;
import com.wecloud.im.thousandchat.service.ThousandChatService;
import com.wecloud.im.vo.ImMessageOfflineListVo;
import com.wecloud.im.vo.OfflineMsgDto;
......@@ -91,8 +57,6 @@ import com.wecloud.im.ws.enums.MsgTypeEnum;
import com.wecloud.im.ws.enums.WsResponseCmdEnum;
import com.wecloud.im.ws.model.WsResponse;
import com.wecloud.im.ws.model.request.PushVO;
import com.wecloud.im.ws.model.request.ReceiveDataVO;
import com.wecloud.im.ws.model.request.ReceiveVO;
import com.wecloud.im.ws.sender.ChannelSender;
import com.wecloud.im.ws.utils.RedisUtils;
import com.wecloud.pushserver.client.model.constant.MqConstant;
......@@ -115,7 +79,6 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
......@@ -124,6 +87,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
......@@ -139,8 +103,6 @@ import java.util.stream.Collectors;
@Service
public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMessage> implements ImMessageService {
private static final String LAST_MSG_PREFIX = "last_msg_";
@Autowired
private ImMessageMapper imMessageMapper;
......@@ -175,7 +137,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
private ApplicationEventPublisher eventPublisher;
@Autowired
private RedisUtils redisUtils;
private SaveMessageService saveMessageService;
@Override
@Transactional(rollbackFor = Exception.class)
......@@ -248,10 +210,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
imMessage.setMsgType(data.getType());
imMessage.setAt(data.getAt());
imMessage.setFkConversationId(data.getToConversation());
this.saveMessageToDb(imMessage);
// eventPublisher.publishEvent(new ClientSendMessageEvent(client.getFkAppid(), imMessage));
saveMessageService.saveMessageToDb(imMessage);
return imMessage;
}
......@@ -290,7 +249,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
ImMessage eventMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.MSG_WITHDRAW, imApplication,
imClientSender, new ImConversation().setId(beOperatedMsg.getFkConversationId()), JsonUtils.encodeJson(content));
this.saveMessageToDb(eventMessage);
saveMessageService.saveMessageToDb(eventMessage);
if (saveOk) {
ImConversation imConversation = imConversationService.getById(beOperatedMsg.getFkConversationId());
......@@ -362,7 +321,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
ImMessage eventMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.MSG_DELETE, imApplication,
imClientSender, new ImConversation().setId(beOperatedMsg.getFkConversationId()), JsonUtils.encodeJson(content));
this.saveMessageToDb(eventMessage);
saveMessageService.saveMessageToDb(eventMessage);
if (saveOk) {
ImConversation imConversation = imConversationService.getById(beOperatedMsg.getFkConversationId());
......@@ -650,7 +609,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
ImMessage message = assembleImMessage(appId, sender, conversation.getId(), param.getMessageType(), false,
JSONObject.toJSONString(param.getContent()));
// 持久化
this.saveMessageToDb(message);
saveMessageService.saveMessageToDb(message);
// 拼装发送消息体
ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, appId);
// 入库 保存收件箱
......@@ -748,7 +707,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
}
}
// 持久化
this.saveMessageToDb(message);
saveMessageService.saveMessageToDb(message);
// 拼装发送消息体
ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, appId);
......@@ -789,7 +748,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
// 组装消息
ImMessage message = assembleImMessage(appId, sender, conversation.getId(), param.getMessageType(), false, JSONObject.toJSONString(param.getContent()));
// 持久化
this.saveMessageToDb(message);
saveMessageService.saveMessageToDb(message);
// 拼装发送消息体
ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, appId);
......@@ -839,7 +798,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
ImMessage message = assembleImMessage(appId, sender, conversation.getId(), param.getMessageType(), true,
JSONObject.toJSONString(param.getContent()));
// 持久化
this.saveMessageToDb(message);
saveMessageService.saveMessageToDb(message);
// 拼装发送消息体
ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, appId);
......@@ -890,7 +849,7 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
ImMessage message = assembleImMessage(application.getId(), sender, conversation.getId(), param.getMessageType(),
false, param.getContent());
// 持久化
this.saveMessageToDb(message);
saveMessageService.saveMessageToDb(message);
// 拼装发送消息体
ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, application.getId());
......@@ -917,53 +876,6 @@ public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMes
return true;
}
/**
* 持久化消息到数据库并更新相关缓存
* expire 锁过期时间 防止死锁
*
* @param imMessage
*/
@Lock4j(keys = {"#imMessage.fkConversationId"}, expire = 3000, acquireTimeout = 1000)
@Override
public void saveMessageToDb(ImMessage imMessage) {
imMessage.setId(SnowflakeUtil.getId());
imMessage.setCreateTime(new Date());
this.setLastMessageIdAndFreshCash(imMessage);
this.save(imMessage);
}
/**
* 查找会话的最新一条消息
*
* @param message
* @return
*/
private void setLastMessageIdAndFreshCash(ImMessage message) {
try {
// 执行业务方法
String key = LAST_MSG_PREFIX + message.getFkConversationId();
String messageId = redisUtils.getKey(key);
if (StringUtils.isNotBlank(messageId)) {
// 将Redis中最新的messageId存入
message.setPreMessageId(Long.valueOf(messageId));
} else {
// 落库查
ImMessage preMessage = imMessageMapper.selectOne(new QueryWrapper<ImMessage>().lambda()
.eq(ImMessage::getFkConversationId, message.getFkConversationId())
.orderByDesc(ImMessage::getId)
.last("limit 1")
);
if (preMessage != null) {
message.setPreMessageId(preMessage.getId());
}
}
redisUtils.addKey(key, message.getId().toString(), Duration.ofMinutes(30));
} catch (Exception e) {
log.info("获取会话最后一条消息异常 ", e);
}
}
private WsResponse buildEventResponse(Long fkConversationId, MsgTypeEnum msgTypeEnum, ImMessage eventMessage, String senderClientId, String msgOwnerClientId, Long beOperatedMsgId) {
ImMessageOnlineSend imMessageOnlineSend = new ImMessageOnlineSend();
BeanUtils.copyProperties(eventMessage, imMessageOnlineSend);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment