From 3ccfd9a29b6d2bc7267829556004b18fa84d22d6 Mon Sep 17 00:00:00 2001 From: heavyrian2012 Date: Tue, 15 Nov 2022 17:12:51 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E5=86=B3user=20setting=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E9=97=AE=E9=A2=98=EF=BC=8Cfix=20#566?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../persistence/MemoryMessagesStore.java | 136 +++++++++++------- 1 file changed, 88 insertions(+), 48 deletions(-) diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 673eb471..b384a70f 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -49,6 +49,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Consumer; @@ -121,6 +122,8 @@ public class MemoryMessagesStore implements IMessagesStore { private Lock mReadLock = mLock.readLock(); private Lock mWriteLock = mLock.writeLock(); + private Lock mUserSettingLock = new ReentrantLock(); + private final DatabaseStore databaseStore; private ConcurrentHashMap userMaxPullSeq = new ConcurrentHashMap<>(); @@ -1422,12 +1425,16 @@ public class MemoryMessagesStore implements IMessagesStore { void removeFavGroup(String groupId, List memberIds) { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); - MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); databaseStore.removeFavGroup(groupId, memberIds); - - for (String member : memberIds) { - userSettingMap.remove(member); + try { + mUserSettingLock.lock(); + MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); + for (String member : memberIds) { + userSettingMap.remove(member); + } + } finally { + mUserSettingLock.unlock(); } } @@ -1507,13 +1514,17 @@ public class MemoryMessagesStore implements IMessagesStore { private void removeGroupUserSettings(String groupId, List users) { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); - MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); databaseStore.removeGroupUserSettings(groupId, users); - for (String userId:users) { - userSettingMap.remove(userId); + try { + mUserSettingLock.lock(); + MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); + for (String userId:users) { + userSettingMap.remove(userId); + } + } finally { + mUserSettingLock.unlock(); } - } @Override @@ -3587,9 +3598,15 @@ public class MemoryMessagesStore implements IMessagesStore { IMap mGroups = hzInstance.getMap(GROUPS_MAP); IMap mChannels = hzInstance.getMap(CHANNELS); - Collection entries = userSettingMap.get(userId); - if (entries == null || entries.size() == 0) { - entries = loadPersistedUserSettings(userId, userSettingMap); + Collection entries; + try { + mUserSettingLock.lock(); + entries = userSettingMap.get(userId); + if (entries == null || entries.size() == 0) { + entries = loadPersistedUserSettings(userId, userSettingMap); + } + } finally { + mUserSettingLock.unlock(); } ErrorCode ec = ErrorCode.ERROR_CODE_NOT_MODIFIED; @@ -3668,17 +3685,22 @@ public class MemoryMessagesStore implements IMessagesStore { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); - Collection entries = userSettingMap.get(userId); - if (entries == null || entries.size() == 0) { - entries = loadPersistedUserSettings(userId, userSettingMap); - } + try { + mUserSettingLock.lock(); + Collection entries = userSettingMap.get(userId); + if (entries == null || entries.size() == 0) { + entries = loadPersistedUserSettings(userId, userSettingMap); + } - if (entries != null) { - for (WFCMessage.UserSettingEntry entry : entries) { - if (entry.getScope() == scope && (key== null || key.equals(entry.getKey()))) { - return entry; + if (entries != null) { + for (WFCMessage.UserSettingEntry entry : entries) { + if (entry.getScope() == scope && (key== null || key.equals(entry.getKey()))) { + return entry; + } } } + } finally { + mUserSettingLock.unlock(); } return null; @@ -3689,19 +3711,25 @@ public class MemoryMessagesStore implements IMessagesStore { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); - Collection entries = userSettingMap.get(userId); - if (entries == null || entries.size() == 0) { - entries = loadPersistedUserSettings(userId, userSettingMap); - } + List result; + try { + mUserSettingLock.lock(); + Collection entries = userSettingMap.get(userId); + if (entries == null || entries.size() == 0) { + entries = loadPersistedUserSettings(userId, userSettingMap); + } - List result = new ArrayList<>(); - if (entries != null) { - for (WFCMessage.UserSettingEntry entry : entries - ) { - if (entry.getScope() == scope) { - result.add(entry); + result = new ArrayList<>(); + if (entries != null) { + for (WFCMessage.UserSettingEntry entry : entries + ) { + if (entry.getScope() == scope) { + result.add(entry); + } } } + } finally { + mUserSettingLock.unlock(); } return result; @@ -3712,28 +3740,34 @@ public class MemoryMessagesStore implements IMessagesStore { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); - Collection entries = userSettingMap.get(userId); - if (entries == null || entries.size() == 0) { - entries = loadPersistedUserSettings(userId, userSettingMap); - if (entries == null) { - entries = new ArrayList<>(); + long updateDt = 0; + try { + mUserSettingLock.lock(); + Collection entries = userSettingMap.get(userId); + if (entries == null || entries.size() == 0) { + entries = loadPersistedUserSettings(userId, userSettingMap); + if (entries == null) { + entries = new ArrayList<>(); + } } - } - long updateDt = System.currentTimeMillis(); - WFCMessage.UserSettingEntry settingEntry = WFCMessage.UserSettingEntry.newBuilder().setScope(request.getScope()).setKey(request.getKey()).setValue(request.getValue()).setUpdateDt(updateDt).build(); - databaseStore.persistUserSetting(userId, settingEntry); + updateDt = System.currentTimeMillis(); + WFCMessage.UserSettingEntry settingEntry = WFCMessage.UserSettingEntry.newBuilder().setScope(request.getScope()).setKey(request.getKey()).setValue(request.getValue()).setUpdateDt(updateDt).build(); + databaseStore.persistUserSetting(userId, settingEntry); - for (WFCMessage.UserSettingEntry entry : entries - ) { - if (entry.getScope() == request.getScope() && entry.getKey().equals(request.getKey())) { - userSettingMap.remove(userId, entry); - userSettingMap.put(userId, settingEntry); - break; + for (WFCMessage.UserSettingEntry entry : entries + ) { + if (entry.getScope() == request.getScope() && entry.getKey().equals(request.getKey())) { + userSettingMap.remove(userId, entry); + userSettingMap.put(userId, settingEntry); + break; + } } - } - userSettingMap.put(userId, settingEntry); + userSettingMap.put(userId, settingEntry); + } finally { + mUserSettingLock.unlock(); + } if(request.getScope() == UserSettingScope.kUserSettingConversationSilent) { int firstSplit = request.getKey().indexOf("-"); @@ -3758,8 +3792,14 @@ public class MemoryMessagesStore implements IMessagesStore { public void clearUserSettings(String userId) { databaseStore.clearUserSetting(userId); HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); - MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); - userSettingMap.remove(userId); + + try { + mUserSettingLock.lock(); + MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); + userSettingMap.remove(userId); + } finally { + mUserSettingLock.unlock(); + } } @Override