xie.bx
3 years ago
66 changed files with 827 additions and 331 deletions
@ -0,0 +1,24 @@ |
|||
package com.bx.common.enums; |
|||
|
|||
public enum ListenerType { |
|||
|
|||
PRIVATE_MESSAGE(0,"私聊消息"), |
|||
GROUP_MESSAGE(1,"群聊消息"); |
|||
|
|||
private Integer code; |
|||
|
|||
private String desc; |
|||
|
|||
ListenerType(Integer index, String desc) { |
|||
this.code =index; |
|||
this.desc=desc; |
|||
} |
|||
|
|||
public String getDesc() { |
|||
return desc; |
|||
} |
|||
|
|||
public Integer getCode(){ |
|||
return this.code; |
|||
} |
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
package com.bx.common.enums; |
|||
|
|||
|
|||
public enum SendResultType { |
|||
|
|||
SUCCESS(0,"发送成功"), |
|||
FAIL(1,"发送失败"); |
|||
|
|||
private int code; |
|||
private String msg; |
|||
|
|||
// 构造方法
|
|||
SendResultType(int code, String msg) { |
|||
this.code = code; |
|||
this.msg = msg; |
|||
} |
|||
public int getCode() { |
|||
return code; |
|||
} |
|||
|
|||
public void setCode(int code) { |
|||
this.code = code; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,17 @@ |
|||
package com.bx.common.model.im; |
|||
|
|||
import lombok.Data; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Data |
|||
public class IMRecvInfo<T> { |
|||
|
|||
private Integer cmd; |
|||
|
|||
private List<Long> recvIds; |
|||
|
|||
private T data; |
|||
} |
|||
|
|||
|
|||
@ -0,0 +1,17 @@ |
|||
package com.bx.common.model.im; |
|||
|
|||
import com.bx.common.enums.SendResultType; |
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
public class SendResult<T> { |
|||
|
|||
private Long recvId; |
|||
|
|||
private SendResultType result; |
|||
|
|||
private String failReason=""; |
|||
|
|||
private T messageInfo; |
|||
|
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
<?xml version="1.0" encoding="UTF-8"?> |
|||
<project xmlns="http://maven.apache.org/POM/4.0.0" |
|||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" |
|||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> |
|||
<parent> |
|||
<artifactId>box-im</artifactId> |
|||
<groupId>com.bx</groupId> |
|||
<version>1.0.0</version> |
|||
</parent> |
|||
<modelVersion>4.0.0</modelVersion> |
|||
|
|||
<artifactId>im-client</artifactId> |
|||
|
|||
<dependencies> |
|||
<dependency> |
|||
<groupId>com.bx</groupId> |
|||
<artifactId>commom</artifactId> |
|||
<version>1.0.0</version> |
|||
</dependency> |
|||
<!-- 引入redis --> |
|||
<dependency> |
|||
<groupId>org.springframework.boot</groupId> |
|||
<artifactId>spring-boot-starter-data-redis</artifactId> |
|||
</dependency> |
|||
</dependencies> |
|||
</project> |
|||
@ -0,0 +1,12 @@ |
|||
package com.bx.imclient; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.context.annotation.ComponentScan; |
|||
import org.springframework.context.annotation.Configuration; |
|||
|
|||
|
|||
@Slf4j |
|||
@Configuration |
|||
@ComponentScan("com.bx.imclient") |
|||
public class IMAutoConfiguration { |
|||
} |
|||
@ -0,0 +1,30 @@ |
|||
package com.bx.imclient; |
|||
|
|||
import com.bx.common.model.im.GroupMessageInfo; |
|||
import com.bx.common.model.im.PrivateMessageInfo; |
|||
import com.bx.imclient.listener.MessageListenerMulticaster; |
|||
import com.bx.imclient.sender.IMSender; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.context.annotation.Configuration; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Configuration |
|||
public class IMClient { |
|||
|
|||
@Autowired |
|||
private MessageListenerMulticaster listenerMulticaster; |
|||
|
|||
@Autowired |
|||
private IMSender imSender; |
|||
|
|||
public void sendPrivateMessage(Long userId, PrivateMessageInfo... messageInfo){ |
|||
imSender.sendPrivateMessage(userId,messageInfo); |
|||
} |
|||
|
|||
public void sendGroupMessage(List<Long> userTokens, GroupMessageInfo... messageInfo){ |
|||
imSender.sendGroupMessage(userTokens,messageInfo); |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -0,0 +1,18 @@ |
|||
package com.bx.imclient.annotation; |
|||
|
|||
import com.bx.common.enums.ListenerType; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.lang.annotation.ElementType; |
|||
import java.lang.annotation.Retention; |
|||
import java.lang.annotation.RetentionPolicy; |
|||
import java.lang.annotation.Target; |
|||
|
|||
@Target({ElementType.TYPE,ElementType.FIELD}) |
|||
@Retention(RetentionPolicy.RUNTIME) |
|||
@Component |
|||
public @interface IMListener { |
|||
|
|||
ListenerType type(); |
|||
|
|||
} |
|||
@ -0,0 +1,49 @@ |
|||
package com.bx.imclient.config; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonAutoDetect; |
|||
import com.fasterxml.jackson.annotation.JsonTypeInfo; |
|||
import com.fasterxml.jackson.annotation.PropertyAccessor; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import com.fasterxml.jackson.databind.SerializationFeature; |
|||
import org.springframework.context.annotation.Bean; |
|||
import org.springframework.context.annotation.Configuration; |
|||
import org.springframework.data.redis.connection.RedisConnectionFactory; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; |
|||
import org.springframework.data.redis.serializer.StringRedisSerializer; |
|||
|
|||
import javax.annotation.Resource; |
|||
|
|||
@Configuration("IMRedisConfig") |
|||
public class RedisConfig { |
|||
|
|||
@Resource |
|||
private RedisConnectionFactory factory; |
|||
|
|||
@Bean("IMRedisTemplate") |
|||
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) { |
|||
RedisTemplate<String, Object> redisTemplate = new RedisTemplate(); |
|||
redisTemplate.setConnectionFactory(redisConnectionFactory); |
|||
// 设置值（value）的序列化采用jackson2JsonRedisSerializer
|
|||
redisTemplate.setValueSerializer(jackson2JsonRedisSerializer()); |
|||
redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer()); |
|||
// 设置键（key）的序列化采用StringRedisSerializer。
|
|||
redisTemplate.setKeySerializer(new StringRedisSerializer()); |
|||
redisTemplate.setHashKeySerializer(new StringRedisSerializer()); |
|||
redisTemplate.afterPropertiesSet(); |
|||
return redisTemplate; |
|||
} |
|||
|
|||
@Bean |
|||
public Jackson2JsonRedisSerializer jackson2JsonRedisSerializer(){ |
|||
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class); |
|||
ObjectMapper om = new ObjectMapper(); |
|||
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY); |
|||
// 解决jackson2无法反序列化LocalDateTime的问题
|
|||
om.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); |
|||
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY); |
|||
jackson2JsonRedisSerializer.setObjectMapper(om); |
|||
return jackson2JsonRedisSerializer; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
package com.bx.imclient.listener; |
|||
|
|||
|
|||
import com.bx.common.model.im.SendResult; |
|||
|
|||
public interface MessageListener { |
|||
|
|||
void process(SendResult result); |
|||
|
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
package com.bx.imclient.listener; |
|||
|
|||
|
|||
import com.bx.common.enums.ListenerType; |
|||
import com.bx.common.model.im.SendResult; |
|||
import com.bx.imclient.annotation.IMListener; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
|
|||
@Component |
|||
public class MessageListenerMulticaster { |
|||
|
|||
|
|||
@Autowired(required = false) |
|||
private List<MessageListener> messageListeners = Collections.emptyList(); |
|||
|
|||
public void multicast(ListenerType type, SendResult result){ |
|||
for(MessageListener listener:messageListeners){ |
|||
IMListener annotation = listener.getClass().getAnnotation(IMListener.class); |
|||
if(annotation.type().equals(type)){ |
|||
listener.process(result); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,112 @@ |
|||
package com.bx.imclient.sender; |
|||
|
|||
import com.bx.common.contant.RedisKey; |
|||
import com.bx.common.enums.IMCmdType; |
|||
import com.bx.common.enums.ListenerType; |
|||
import com.bx.common.enums.SendResultType; |
|||
import com.bx.common.model.im.GroupMessageInfo; |
|||
import com.bx.common.model.im.IMRecvInfo; |
|||
import com.bx.common.model.im.PrivateMessageInfo; |
|||
import com.bx.common.model.im.SendResult; |
|||
import com.bx.imclient.listener.MessageListenerMulticaster; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Qualifier; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.stereotype.Service; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.LinkedList; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
|
|||
@Service |
|||
public class IMSender { |
|||
|
|||
@Autowired |
|||
@Qualifier("IMRedisTemplate") |
|||
private RedisTemplate redisTemplate; |
|||
|
|||
@Autowired |
|||
private MessageListenerMulticaster listenerMulticaster; |
|||
|
|||
public void sendPrivateMessage(Long recvId, PrivateMessageInfo... messageInfos){ |
|||
// 获取对方连接的channelId
|
|||
String key = RedisKey.IM_USER_SERVER_ID + recvId; |
|||
Integer serverId = (Integer) redisTemplate.opsForValue().get(key); |
|||
// 如果对方在线，将数据存储至redis，等待拉取推送
|
|||
if (serverId != null) { |
|||
String sendKey = RedisKey.IM_UNREAD_PRIVATE_QUEUE + serverId; |
|||
IMRecvInfo[] recvInfos = new IMRecvInfo[messageInfos.length]; |
|||
for (int i=0;i<messageInfos.length;i++){ |
|||
IMRecvInfo<PrivateMessageInfo> recvInfo = new IMRecvInfo<>(); |
|||
recvInfo.setCmd(IMCmdType.PRIVATE_MESSAGE.getCode()); |
|||
List recvIds = new LinkedList(); |
|||
recvIds.add(recvId); |
|||
recvInfo.setRecvIds(recvIds); |
|||
recvInfo.setData(messageInfos[i]); |
|||
recvInfos[i] = recvInfo; |
|||
} |
|||
redisTemplate.opsForList().rightPushAll(sendKey, recvInfos); |
|||
}else{ |
|||
// 回复消息状态
|
|||
for(PrivateMessageInfo messageInfo : messageInfos ) { |
|||
SendResult result = new SendResult(); |
|||
result.setMessageInfo(messageInfo); |
|||
result.setRecvId(recvId); |
|||
result.setResult(SendResultType.FAIL); |
|||
result.setFailReason("用户不在线"); |
|||
listenerMulticaster.multicast(ListenerType.PRIVATE_MESSAGE, result); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public void sendGroupMessage(List<Long> recvIds, GroupMessageInfo... messageInfos){ |
|||
// 根据群聊每个成员所连的IM-server，进行分组
|
|||
List<Long> offLineIds = Collections.synchronizedList(new LinkedList<Long>()); |
|||
Map<Integer, List<Long>> serverMap = new ConcurrentHashMap<>(); |
|||
recvIds.parallelStream().forEach(id->{ |
|||
String key = RedisKey.IM_USER_SERVER_ID + id; |
|||
Integer serverId = (Integer)redisTemplate.opsForValue().get(key); |
|||
if(serverId != null){ |
|||
if(serverMap.containsKey(serverId)){ |
|||
serverMap.get(serverId).add(id); |
|||
}else { |
|||
// 此处需要加锁，否则list可以会被覆盖
|
|||
synchronized(serverMap){ |
|||
List<Long> list = Collections.synchronizedList(new LinkedList<Long>()); |
|||
list.add(id); |
|||
serverMap.put(serverId,list); |
|||
} |
|||
} |
|||
}else{ |
|||
offLineIds.add(id); |
|||
} |
|||
}); |
|||
// 逐个server发送
|
|||
for (Map.Entry<Integer,List<Long>> entry : serverMap.entrySet()) { |
|||
IMRecvInfo[] recvInfos = new IMRecvInfo[messageInfos.length]; |
|||
for (int i=0;i<messageInfos.length;i++){ |
|||
IMRecvInfo<GroupMessageInfo> recvInfo = new IMRecvInfo<>(); |
|||
recvInfo.setCmd(IMCmdType.GROUP_MESSAGE.getCode()); |
|||
recvInfo.setRecvIds(new LinkedList<>(entry.getValue())); |
|||
recvInfo.setData(messageInfos[i]); |
|||
recvInfos[i] = recvInfo; |
|||
} |
|||
String key = RedisKey.IM_UNREAD_GROUP_QUEUE +entry.getKey(); |
|||
redisTemplate.opsForList().rightPushAll(key,recvInfos); |
|||
} |
|||
// 不在线的用户，回复消息状态
|
|||
for(GroupMessageInfo messageInfo:messageInfos ){ |
|||
for(Long id : offLineIds){ |
|||
// 回复消息状态
|
|||
SendResult result = new SendResult(); |
|||
result.setMessageInfo(messageInfo); |
|||
result.setRecvId(id); |
|||
result.setResult(SendResultType.FAIL); |
|||
result.setFailReason("用户不在线"); |
|||
listenerMulticaster.multicast(ListenerType.GROUP_MESSAGE,result); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
package com.bx.imclient.task; |
|||
|
|||
import lombok.SneakyThrows; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
@Slf4j |
|||
public abstract class AbstractPullMessageTask { |
|||
|
|||
private int threadNum = 8; |
|||
|
|||
private ExecutorService executorService = Executors.newFixedThreadPool(threadNum); |
|||
|
|||
@PostConstruct |
|||
public void init(){ |
|||
// 初始化定时器
|
|||
for(int i=0;i<threadNum;i++){ |
|||
executorService.execute(new Runnable() { |
|||
@SneakyThrows |
|||
@Override |
|||
public void run() { |
|||
try{ |
|||
pullMessage(); |
|||
}catch (Exception e){ |
|||
log.error("任务调度异常",e); |
|||
Thread.sleep(200); |
|||
} |
|||
if(!executorService.isShutdown()){ |
|||
executorService.execute(this); |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy(){ |
|||
log.info("{}线程任务关闭",this.getClass().getSimpleName()); |
|||
executorService.shutdown(); |
|||
} |
|||
|
|||
public abstract void pullMessage(); |
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
package com.bx.imclient.task; |
|||
|
|||
import com.bx.common.contant.RedisKey; |
|||
import com.bx.common.enums.ListenerType; |
|||
import com.bx.common.model.im.SendResult; |
|||
import com.bx.imclient.listener.MessageListenerMulticaster; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Qualifier; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Component |
|||
public class PullSendResultGroupMessageTask extends AbstractPullMessageTask{ |
|||
|
|||
@Qualifier("IMRedisTemplate") |
|||
@Autowired |
|||
private RedisTemplate<String,Object> redisTemplate; |
|||
|
|||
@Autowired |
|||
private MessageListenerMulticaster listenerMulticaster; |
|||
|
|||
@Override |
|||
public void pullMessage() { |
|||
String key = RedisKey.IM_RESULT_GROUP_QUEUE; |
|||
SendResult result = (SendResult)redisTemplate.opsForList().leftPop(key,10, TimeUnit.SECONDS); |
|||
if(result != null) { |
|||
listenerMulticaster.multicast(ListenerType.GROUP_MESSAGE,result); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
package com.bx.imclient.task; |
|||
|
|||
import com.bx.common.contant.RedisKey; |
|||
import com.bx.common.enums.ListenerType; |
|||
import com.bx.common.model.im.SendResult; |
|||
import com.bx.imclient.listener.MessageListenerMulticaster; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Qualifier; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
public class PullSendResultPrivateMessageTask extends AbstractPullMessageTask{ |
|||
|
|||
|
|||
@Qualifier("IMRedisTemplate") |
|||
@Autowired |
|||
private RedisTemplate<String,Object> redisTemplate; |
|||
|
|||
@Autowired |
|||
private MessageListenerMulticaster listenerMulticaster; |
|||
|
|||
@Override |
|||
public void pullMessage() { |
|||
String key = RedisKey.IM_RESULT_PRIVATE_QUEUE; |
|||
SendResult result = (SendResult)redisTemplate.opsForList().leftPop(key,10, TimeUnit.SECONDS); |
|||
if(result != null) { |
|||
listenerMulticaster.multicast(ListenerType.PRIVATE_MESSAGE, result); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,2 @@ |
|||
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ |
|||
com.bx.imclient.IMAutoConfiguration |
|||
@ -0,0 +1,12 @@ |
|||
package com.bx.implatform.contant; |
|||
|
|||
|
|||
public class Constant { |
|||
// 最大图片上传大小
|
|||
public static final long MAX_IMAGE_SIZE = 5*1024*1024; |
|||
// 最大上传文件大小
|
|||
public static final long MAX_FILE_SIZE = 10*1024*1024; |
|||
// 群聊最大人数
|
|||
public static final long MAX_GROUP_MEMBER = 500; |
|||
|
|||
} |
|||
@ -0,0 +1,16 @@ |
|||
package com.bx.implatform.contant; |
|||
|
|||
public class RedisKey { |
|||
|
|||
// 已读群聊消息位置(已读最大id)
|
|||
public final static String IM_GROUP_READED_POSITION = "im:readed:group:position:"; |
|||
// 缓存前缀
|
|||
public final static String IM_CACHE = "im:cache:"; |
|||
// 缓存是否好友：bool
|
|||
public final static String IM_CACHE_FRIEND = IM_CACHE+"friend"; |
|||
// 缓存群聊信息
|
|||
public final static String IM_CACHE_GROUP = IM_CACHE+"group"; |
|||
// 缓存群聊成员id
|
|||
public final static String IM_CACHE_GROUP_MEMBER_ID = IM_CACHE+"group_member_ids"; |
|||
|
|||
} |
|||
@ -1,4 +1,4 @@ |
|||
package com.bx.common.enums; |
|||
package com.bx.implatform.enums; |
|||
|
|||
/** |
|||
* 响应码枚举 |
|||
@ -1,4 +1,4 @@ |
|||
package com.bx.common.generator; |
|||
package com.bx.implatform.generator; |
|||
|
|||
import com.baomidou.mybatisplus.annotation.IdType; |
|||
import com.baomidou.mybatisplus.core.toolkit.StringPool; |
|||
@ -0,0 +1,34 @@ |
|||
package com.bx.implatform.listener; |
|||
|
|||
import com.bx.common.enums.ListenerType; |
|||
import com.bx.common.enums.MessageType; |
|||
import com.bx.common.model.im.GroupMessageInfo; |
|||
import com.bx.common.model.im.SendResult; |
|||
import com.bx.imclient.annotation.IMListener; |
|||
import com.bx.imclient.listener.MessageListener; |
|||
import com.bx.implatform.contant.RedisKey; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
|
|||
|
|||
@Slf4j |
|||
@IMListener(type = ListenerType.GROUP_MESSAGE) |
|||
public class GroupMessageListener implements MessageListener { |
|||
|
|||
@Autowired |
|||
private RedisTemplate<String,Object> redisTemplate; |
|||
|
|||
@Override |
|||
public void process(SendResult result){ |
|||
GroupMessageInfo messageInfo = (GroupMessageInfo) result.getMessageInfo(); |
|||
if(messageInfo.getType().equals(MessageType.TIP)){ |
|||
// 提示类数据不记录
|
|||
return; |
|||
} |
|||
// 保存该用户已拉取的最大消息id
|
|||
String key = RedisKey.IM_GROUP_READED_POSITION + messageInfo.getGroupId()+":"+result.getRecvId(); |
|||
redisTemplate.opsForValue().set(key,messageInfo.getId()); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
package com.bx.implatform.listener; |
|||
|
|||
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper; |
|||
import com.bx.common.enums.ListenerType; |
|||
import com.bx.common.enums.MessageStatus; |
|||
import com.bx.common.enums.MessageType; |
|||
import com.bx.common.model.im.PrivateMessageInfo; |
|||
import com.bx.common.model.im.SendResult; |
|||
import com.bx.imclient.annotation.IMListener; |
|||
import com.bx.imclient.listener.MessageListener; |
|||
import com.bx.implatform.entity.PrivateMessage; |
|||
import com.bx.implatform.service.IPrivateMessageService; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
|
|||
|
|||
@Slf4j |
|||
@IMListener(type = ListenerType.PRIVATE_MESSAGE) |
|||
public class PrivateMessageListener implements MessageListener { |
|||
|
|||
@Autowired |
|||
private IPrivateMessageService privateMessageService; |
|||
|
|||
@Override |
|||
public void process(SendResult result){ |
|||
PrivateMessageInfo messageInfo = (PrivateMessageInfo) result.getMessageInfo(); |
|||
if(messageInfo.getType().equals(MessageType.TIP)){ |
|||
// 提示类数据不记录
|
|||
return; |
|||
} |
|||
// 更新消息状态
|
|||
UpdateWrapper<PrivateMessage> updateWrapper = new UpdateWrapper<>(); |
|||
updateWrapper.lambda().eq(PrivateMessage::getId,messageInfo.getId()) |
|||
.eq(PrivateMessage::getStatus, MessageStatus.UNREAD.getCode()) |
|||
.set(PrivateMessage::getStatus, MessageStatus.ALREADY_READ.getCode()); |
|||
privateMessageService.update(updateWrapper); |
|||
log.info("消息已读，消息id:{}，发送者:{},接收者:{}",messageInfo.getId(),messageInfo.getSendId(),messageInfo.getRecvId()); |
|||
} |
|||
|
|||
} |
|||
@ -1,4 +1,4 @@ |
|||
package com.bx.common.result; |
|||
package com.bx.implatform.result; |
|||
|
|||
import lombok.Data; |
|||
|
|||
@ -1,74 +0,0 @@ |
|||
package com.bx.implatform.task; |
|||
|
|||
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper; |
|||
import com.bx.common.contant.RedisKey; |
|||
import com.bx.common.enums.MessageStatusEnum; |
|||
import com.bx.implatform.entity.PrivateMessage; |
|||
import com.bx.implatform.service.IPrivateMessageService; |
|||
import lombok.SneakyThrows; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
public class PullAlreadyReadMessageTask { |
|||
|
|||
private int threadNum = 8; |
|||
|
|||
private ExecutorService executorService = Executors.newFixedThreadPool(threadNum); |
|||
|
|||
@Autowired |
|||
private RedisTemplate<String,Object> redisTemplate; |
|||
|
|||
@Autowired |
|||
private IPrivateMessageService privateMessageService; |
|||
|
|||
@PostConstruct |
|||
public void init(){ |
|||
for(int i=0;i<threadNum;i++){ |
|||
executorService.execute(new Task()); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy(){ |
|||
log.info("{}线程任务关闭",this.getClass().getSimpleName()); |
|||
executorService.shutdown(); |
|||
} |
|||
|
|||
|
|||
protected class Task implements Runnable{ |
|||
@SneakyThrows |
|||
@Override |
|||
public void run() { |
|||
try { |
|||
String key = RedisKey.IM_READED_PRIVATE_MESSAGE_ID; |
|||
Integer msgId = (Integer)redisTemplate.opsForList().leftPop(key,10, TimeUnit.SECONDS); |
|||
if(msgId!=null){ |
|||
UpdateWrapper<PrivateMessage> updateWrapper = new UpdateWrapper<>(); |
|||
updateWrapper.lambda().eq(PrivateMessage::getId,msgId) |
|||
.eq(PrivateMessage::getStatus,MessageStatusEnum.UNREAD.getCode()) |
|||
.set(PrivateMessage::getStatus, MessageStatusEnum.ALREADY_READ.getCode()); |
|||
privateMessageService.update(updateWrapper); |
|||
log.info("消息已读，id:{}",msgId); |
|||
} |
|||
}catch (Exception e){ |
|||
log.error(e.getMessage()); |
|||
Thread.sleep(200); |
|||
}finally { |
|||
// 下一次循环
|
|||
if(!executorService.isShutdown()){ |
|||
executorService.submit(this); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -1,4 +1,4 @@ |
|||
package com.bx.common.util; |
|||
package com.bx.implatform.util; |
|||
|
|||
import cn.hutool.core.date.DateTime; |
|||
import cn.hutool.core.date.DateUtil; |
|||
Loading…
Reference in new issue