anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

时间回拨问题优化

Browse Source
This commit is contained in:
imhao183
2020-08-24 09:30:59 +08:00
parent 989f5aa190
commit e41ae9a9e7
6 changed files with 104 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java
+7 -1
View File
@@ -33,7 +33,13 @@ public class BroadcastMessageHandler extends IMHandler<WFCMessage.Message> {
}
long timestamp = System.currentTimeMillis();
long messageId = MessageShardingUtil.generateId();
long messageId = 0;
try {
messageId = MessageShardingUtil.generateId();
} catch (Exception e) {
e.printStackTrace();
return ErrorCode.ERROR_CODE_SERVER_ERROR;
}
message = message.toBuilder().setFromUser(fromUser).setMessageId(messageId).setServerTimestamp(timestamp).setConversation(WFCMessage.Conversation.newBuilder().setTarget(fromUser).setLine(message.getConversation().getLine()).setType(ProtoConstants.ConversationType.ConversationType_Private).build()).build();
long count = saveAndBroadcast(fromUser, clientID, message);
broker/src/main/java/io/moquette/imhandler/GroupHandler.java
+7 -3
View File
@@ -32,9 +32,13 @@ abstract public class GroupHandler<T> extends IMHandler<T> {
WFCMessage.Message.Builder builder = WFCMessage.Message.newBuilder().setContent(content).setServerTimestamp(timestamp);
builder.setConversation(builder.getConversationBuilder().setType(ProtoConstants.ConversationType.ConversationType_Group).setTarget(targetId).setLine(line));
builder.setFromUser(fromUser);
long messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
saveAndPublish(fromUser, null, builder.build(), false);
try {
long messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
saveAndPublish(fromUser, null, builder.build(), false);
} catch (Exception e) {
e.printStackTrace();
}
}
}
broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java
+46 -42
View File
@@ -32,50 +32,54 @@ public class HandleFriendRequestHandler extends IMHandler<WFCMessage.HandleFrien
if (errorCode == ERROR_CODE_SUCCESS) {
if (!isAdmin && builder.getConversation() != null && request.getStatus() == ProtoConstants.FriendRequestStatus.RequestStatus_Accepted) {
long messageId = MessageShardingUtil.generateId();
long timestamp = System.currentTimeMillis();
builder.setMessageId(messageId);
builder.setServerTimestamp(timestamp);
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
try {
long messageId = MessageShardingUtil.generateId();
long timestamp = System.currentTimeMillis();
builder.setMessageId(messageId);
builder.setServerTimestamp(timestamp);
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
MemorySessionStore.Session session = m_sessionsStore.getSession(clientID);
String language = "zh_CN";
if (session != null && !StringUtil.isNullOrEmpty(session.getLanguage())) {
language = session.getLanguage();
MemorySessionStore.Session session = m_sessionsStore.getSession(clientID);
String language = "zh_CN";
if (session != null && !StringUtil.isNullOrEmpty(session.getLanguage())) {
language = session.getLanguage();
}
WFCMessage.MessageContent.Builder contentBuilder = WFCMessage.MessageContent.newBuilder();
if (m_messagesStore.isNewFriendWelcomeMessage()) {
contentBuilder.setType(92);
} else {
contentBuilder.setType(90).setContent(I18n.getString(language, "Above_Greeting_Message"));
}
builder = WFCMessage.Message.newBuilder();
builder.setFromUser(request.getTargetUid());
builder.setConversation(WFCMessage.Conversation.newBuilder().setTarget(fromUser).setLine(0).setType(ProtoConstants.ConversationType.ConversationType_Private).build());
builder.setContent(contentBuilder);
timestamp = System.currentTimeMillis();
builder.setServerTimestamp(timestamp);
messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
if (m_messagesStore.isNewFriendWelcomeMessage()) {
contentBuilder.setType(93);
} else {
contentBuilder.setContent(I18n.getString(language, "Friend_Can_Start_Chat"));
}
builder.setContent(contentBuilder);
messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
timestamp = System.currentTimeMillis();
builder.setServerTimestamp(timestamp);
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
publisher.publishNotification(IMTopic.NotifyFriendTopic, request.getTargetUid(), heads[0]);
publisher.publishNotification(IMTopic.NotifyFriendTopic, fromUser, heads[1]);
} catch (Exception e) {
e.printStackTrace();
}
WFCMessage.MessageContent.Builder contentBuilder = WFCMessage.MessageContent.newBuilder();
if (m_messagesStore.isNewFriendWelcomeMessage()) {
contentBuilder.setType(92);
} else {
contentBuilder.setType(90).setContent(I18n.getString(language, "Above_Greeting_Message"));
}
builder = WFCMessage.Message.newBuilder();
builder.setFromUser(request.getTargetUid());
builder.setConversation(WFCMessage.Conversation.newBuilder().setTarget(fromUser).setLine(0).setType(ProtoConstants.ConversationType.ConversationType_Private).build());
builder.setContent(contentBuilder);
timestamp = System.currentTimeMillis();
builder.setServerTimestamp(timestamp);
messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
if (m_messagesStore.isNewFriendWelcomeMessage()) {
contentBuilder.setType(93);
} else {
contentBuilder.setContent(I18n.getString(language, "Friend_Can_Start_Chat"));
}
builder.setContent(contentBuilder);
messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
timestamp = System.currentTimeMillis();
builder.setServerTimestamp(timestamp);
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
publisher.publishNotification(IMTopic.NotifyFriendTopic, request.getTargetUid(), heads[0]);
publisher.publishNotification(IMTopic.NotifyFriendTopic, fromUser, heads[1]);
}
}
return errorCode;
broker/src/main/java/io/moquette/imhandler/MultiCastMessageHandler.java
+7 -1
View File
@@ -36,7 +36,13 @@ public class MultiCastMessageHandler extends IMHandler<WFCMessage.MultiCastMessa
}
long timestamp = System.currentTimeMillis();
long messageId = MessageShardingUtil.generateId();
long messageId = 0;
try {
messageId = MessageShardingUtil.generateId();
} catch (Exception e) {
e.printStackTrace();
return ErrorCode.ERROR_CODE_SERVER_ERROR;
}
WFCMessage.Message message = WFCMessage.Message.newBuilder()
.setContent(multiCastMessage.getContent())
.setConversation(WFCMessage.Conversation.newBuilder().setTarget(fromUser).setType(ConversationType_Private).setLine(multiCastMessage.getLine()).build())
broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java
+7 -1
View File
@@ -100,7 +100,13 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
}
long timestamp = System.currentTimeMillis();
long messageId = MessageShardingUtil.generateId();
long messageId = 0;
try {
messageId = MessageShardingUtil.generateId();
} catch (Exception e) {
e.printStackTrace();
return ErrorCode.ERROR_CODE_SERVER_ERROR;
}
message = message.toBuilder().setFromUser(fromUser).setMessageId(messageId).setServerTimestamp(timestamp).build();
if (mForwardUrl != null) {
broker/src/main/java/win/liyufan/im/MessageShardingUtil.java
+30 -1
View File
@@ -14,6 +14,7 @@ import java.util.Date;
public class MessageShardingUtil {
private static SpinLock mLock = new SpinLock();
private static volatile int rotateId = 0;
private static volatile long timeId = 0;
private static int nodeId = 0;
private static int rotateIdWidth = 15;
@@ -33,13 +34,41 @@ public class MessageShardingUtil {
* 所以时间限制是到2157/5/15（2的42次幂代表的时间 + (2018-1970)）。节点数限制是小于64，每台服务器每毫秒最多发送32768条消息
* @return
*/
public static long generateId() {
public static long generateId() throws Exception {
mLock.lock();
rotateId = (rotateId + 1)&rotateIdMask;
mLock.unLock();
long id = System.currentTimeMillis() - T201801010000;
if (id > timeId) {
timeId = id;
rotateId = 1;
} else if(id == timeId) {
if (rotateId == (rotateIdMask - 1)) { //当前空间已经用完，等待下一个空间打开
while (id <= timeId) {
id = System.currentTimeMillis() - T201801010000;
}
mLock.unLock();
return generateId();
}
} else { //id < timeId;
if (rotateId > (rotateIdMask -1)*9/10) { //空间已经接近用完
if (timeId - id < 3000) { //时间回拨小于3秒，等待时间赶上回拨之前记录
while (id < timeId) {
id = System.currentTimeMillis() - T201801010000;
}
mLock.unLock();
return generateId();
} else { //时间回拨大于3秒，抛出异常，这段时间消息将不可用。
mLock.unLock();
throw new Exception("Time turn back " + (timeId - id) + " ms, it too long!!!");
}
} else {
id = timeId;
}
}
id <<= nodeIdWidth;
id += (nodeId & nodeIdMask);