From 4fdc883f25433d8d270db445378f202b7267fc5a Mon Sep 17 00:00:00 2001 From: heavyrain2012 Date: Sat, 20 Nov 2021 10:52:35 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E6=9C=8D=E5=8A=A1=E5=9C=B0?= =?UTF-8?q?=E5=9D=80=E9=94=99=E8=AF=AF=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/io/moquette/BrokerConstants.java | 7 --- .../io/moquette/imhandler/RouteHandler.java | 11 ++--- .../persistence/MemoryMessagesStore.java | 7 ++- .../main/java/io/moquette/server/Server.java | 46 ++++++++----------- 4 files changed, 26 insertions(+), 45 deletions(-) diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index cd2be18b..1ced4d77 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -30,8 +30,6 @@ public final class BrokerConstants { public static final String HTTP_LOCAL_PORT = "local_port"; public static final String HTTP_ADMIN_PORT = "http.admin.port"; - public static final String NODE_ID = "node_id"; - public static final String NODE_IDS = "node_ids"; public static final String DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME = "moquette_store.mapdb"; public static final String DEFAULT_PERSISTENT_PATH = System.getProperty("user.dir") + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; @@ -106,11 +104,6 @@ public final class BrokerConstants { public static final String CHATROOM_INFO_UPDATE_CALLBACK = "chatroom.chatroom_info_update_callback"; public static final String CHATROOM_MEMBER_UPDATE_CALLBACK = "chatroom.chatroom_member_update_callback"; - - public static final String HZ_Cluster_Node_External_IP = "node_external_ip"; - public static final String HZ_Cluster_Node_External_Long_Port = "node_external_long_port"; - public static final String HZ_Cluster_Node_External_Short_Port = "node_external_short_port"; - public static final String HZ_Cluster_Node_ID = "node_id"; public static final String HTTP_SERVER_SECRET_KEY = "http.admin.secret_key"; public static final String HTTP_SERVER_API_NO_CHECK_TIME = "http.admin.no_check_time"; diff --git a/broker/src/main/java/io/moquette/imhandler/RouteHandler.java b/broker/src/main/java/io/moquette/imhandler/RouteHandler.java index 5bda5e6a..ae420cdf 100644 --- a/broker/src/main/java/io/moquette/imhandler/RouteHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/RouteHandler.java @@ -17,10 +17,6 @@ import io.netty.buffer.ByteBuf; import cn.wildfirechat.common.ErrorCode; import win.liyufan.im.IMTopic; -import static io.moquette.BrokerConstants.HZ_Cluster_Node_External_IP; -import static io.moquette.BrokerConstants.HZ_Cluster_Node_External_Long_Port; -import static io.moquette.BrokerConstants.HZ_Cluster_Node_External_Short_Port; - @Handler(IMTopic.RouteTopic) public class RouteHandler extends IMHandler { @Override @@ -38,10 +34,9 @@ public class RouteHandler extends IMHandler { return ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH; } - Member member = mServer.getHazelcastInstance().getCluster().getLocalMember(); - String serverIp = member.getStringAttribute(HZ_Cluster_Node_External_IP); - String longPort = member.getStringAttribute(HZ_Cluster_Node_External_Long_Port); - String shortPort = member.getStringAttribute(HZ_Cluster_Node_External_Short_Port); + String serverIp = mServer.getServerIp(); + String longPort = mServer.getLongPort(); + String shortPort = mServer.getShortPort(); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); boolean isSessionAlreadyStored = clientSession != null; diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index a51f9fb9..dea3fadf 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -2597,10 +2597,9 @@ public class MemoryMessagesStore implements IMessagesStore { if (tokenUserId != null) { if (tokenUserId.equals(userId)) { - Member member = m_Server.getHazelcastInstance().getCluster().getLocalMember(); - String serverIp = member.getStringAttribute(HZ_Cluster_Node_External_IP); - String longPort = member.getStringAttribute(HZ_Cluster_Node_External_Long_Port); - String shortPort = member.getStringAttribute(HZ_Cluster_Node_External_Short_Port); + String serverIp = m_Server.getServerIp(); + String longPort = m_Server.getLongPort(); + String shortPort = m_Server.getShortPort(); if (!StringUtil.isNullOrEmpty(serverIp)) { serverIPs.add(serverIp); } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index f2462612..004b8ab1 100755 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -367,12 +367,9 @@ public class Server { } private boolean configureCluster(IConfig config) throws FileNotFoundException { LOG.info("Configuring embedded Hazelcast instance"); - String serverIp = getServerIp(config); + serverIp = getServerIp(config); String hzConfigPath = "config/hazelcast.xml"; - String hzClientIp = config.getProperty(BrokerConstants.HAZELCAST_CLIENT_IP, "localhost"); - String hzClientPort = config.getProperty(BrokerConstants.HAZELCAST_CLIENT_PORT, "5703"); - if (hzConfigPath != null) { boolean isHzConfigOnClasspath = this.getClass().getClassLoader().getResource(hzConfigPath) != null; Config hzconfig = isHzConfigOnClasspath @@ -385,29 +382,10 @@ public class Server { hazelcastInstance = Hazelcast.newHazelcastInstance(); } + longPort = config.getProperty(BrokerConstants.PORT_PROPERTY_NAME); + shortPort = config.getProperty(BrokerConstants.HTTP_SERVER_PORT); - String longPort = config.getProperty(BrokerConstants.PORT_PROPERTY_NAME); - String shortPort = config.getProperty(BrokerConstants.HTTP_SERVER_PORT); - String nodeIdStr = config.getProperty(BrokerConstants.NODE_ID); - ISet nodeIdSet = hazelcastInstance.getSet(BrokerConstants.NODE_IDS); - int nodeId; - try { - nodeId = Integer.parseInt(nodeIdStr); - }catch (Exception e){ - throw new IllegalArgumentException("nodeId error: " + nodeIdStr); - } - if (nodeIdSet != null && nodeIdSet.contains(nodeId)){ - LOG.error("只允许一个实例运行，多个实例会引起冲突，进程终止"); - System.exit(-1); - } - - MessageShardingUtil.setNodeId(nodeId); - nodeIdSet.add(nodeId); - - hazelcastInstance.getCluster().getLocalMember().setStringAttribute(HZ_Cluster_Node_External_Long_Port, longPort); - hazelcastInstance.getCluster().getLocalMember().setStringAttribute(HZ_Cluster_Node_External_Short_Port, shortPort); - hazelcastInstance.getCluster().getLocalMember().setIntAttribute(HZ_Cluster_Node_ID, nodeId); - hazelcastInstance.getCluster().getLocalMember().setStringAttribute(HZ_Cluster_Node_External_IP, serverIp); + MessageShardingUtil.setNodeId(1); Tokenor.setKey(config.getProperty(BrokerConstants.TOKEN_SECRET_KEY)); String expirTimeStr = config.getProperty(TOKEN_EXPIRE_TIME); if (!StringUtil.isNullOrEmpty(expirTimeStr)) { @@ -421,6 +399,22 @@ public class Server { return true; } + private String serverIp; + private String longPort; + private String shortPort; + + public String getServerIp() { + return serverIp; + } + + public String getLongPort() { + return longPort; + } + + public String getShortPort() { + return shortPort; + } + public HazelcastInstance getHazelcastInstance() { return hazelcastInstance; }