From e41ae9a9e74c1204d078bf4635ef7dba5dcbde1c Mon Sep 17 00:00:00 2001 From: imhao183 Date: Mon, 24 Aug 2020 09:30:59 +0800 Subject: [PATCH] =?UTF-8?q?=E6=97=B6=E9=97=B4=E5=9B=9E=E6=8B=A8=E9=97=AE?= =?UTF-8?q?=E9=A2=98=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../imhandler/BroadcastMessageHandler.java | 8 +- .../io/moquette/imhandler/GroupHandler.java | 10 ++- .../imhandler/HandleFriendRequestHandler.java | 88 ++++++++++--------- .../imhandler/MultiCastMessageHandler.java | 8 +- .../imhandler/SendMessageHandler.java | 8 +- .../win/liyufan/im/MessageShardingUtil.java | 31 ++++++- 6 files changed, 104 insertions(+), 49 deletions(-) diff --git a/broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java b/broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java index df98c0fe..22d13b59 100644 --- a/broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java @@ -33,7 +33,13 @@ public class BroadcastMessageHandler extends IMHandler { } 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); diff --git a/broker/src/main/java/io/moquette/imhandler/GroupHandler.java b/broker/src/main/java/io/moquette/imhandler/GroupHandler.java index 62670b33..2f603b25 100644 --- a/broker/src/main/java/io/moquette/imhandler/GroupHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/GroupHandler.java @@ -32,9 +32,13 @@ abstract public class GroupHandler extends IMHandler { 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(); + } } } diff --git a/broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java b/broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java index c6cb5bab..8e5980cb 100644 --- a/broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java @@ -32,50 +32,54 @@ public class HandleFriendRequestHandler extends IMHandler { } 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) { diff --git a/broker/src/main/java/win/liyufan/im/MessageShardingUtil.java b/broker/src/main/java/win/liyufan/im/MessageShardingUtil.java index 65b97b82..6cd733aa 100644 --- a/broker/src/main/java/win/liyufan/im/MessageShardingUtil.java +++ b/broker/src/main/java/win/liyufan/im/MessageShardingUtil.java @@ -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);