diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java index ed694799cb..7382587dcd 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java +++ b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java @@ -28,6 +28,9 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Consumer; import org.apache.rocketmq.common.ThreadFactoryImpl; import org.apache.rocketmq.common.constant.LoggerName; @@ -63,6 +66,8 @@ public class AutoSwitchHAService extends DefaultHAService { private EpochFileCache epochCache; private AutoSwitchHAClient haClient; + private ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); + public AutoSwitchHAService() { } @@ -287,8 +292,14 @@ public class AutoSwitchHAService extends DefaultHAService { } @Override - public synchronized int inSyncReplicasNums(final long masterPutWhere) { - return syncStateSet.size(); + public int inSyncReplicasNums(final long masterPutWhere) { + final Lock readLock = readWriteLock.readLock(); + try { + readLock.lock(); + return syncStateSet.size(); + } finally { + readLock.unlock(); + } } @Override @@ -344,16 +355,28 @@ public class AutoSwitchHAService extends DefaultHAService { return confirmOffset; } - public synchronized void setSyncStateSet(final Set syncStateSet) { - this.syncStateSet.clear(); - this.syncStateSet.addAll(syncStateSet); - this.confirmOffset = computeConfirmOffset(); + public void setSyncStateSet(final Set syncStateSet) { + final Lock writeLock = readWriteLock.writeLock(); + try { + writeLock.lock(); + this.syncStateSet.clear(); + this.syncStateSet.addAll(syncStateSet); + this.confirmOffset = computeConfirmOffset(); + } finally { + writeLock.unlock(); + } } - public synchronized Set getSyncStateSet() { - HashSet set = new HashSet<>(this.syncStateSet.size()); - set.addAll(this.syncStateSet); - return set; + public Set getSyncStateSet() { + final Lock readLock = readWriteLock.readLock(); + try { + readLock.lock(); + HashSet set = new HashSet<>(this.syncStateSet.size()); + set.addAll(this.syncStateSet); + return set; + } finally { + readLock.unlock(); + } } public void truncateEpochFilePrefix(final long offset) {