opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #5809] Replace synchronized with ReentrantReadWriteLock in AutoSwitchHAService (#5810)

Browse Source
This commit is contained in:
mxsm
2023-01-03 15:43:02 +08:00
committed by GitHub
parent d070d467a6
commit 24c9a52c3a
1 changed files with 33 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
+33 -10
View File
@@ -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<String> syncStateSet) {
this.syncStateSet.clear();
this.syncStateSet.addAll(syncStateSet);
this.confirmOffset = computeConfirmOffset();
public void setSyncStateSet(final Set<String> 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<String> getSyncStateSet() {
HashSet<String> set = new HashSet<>(this.syncStateSet.size());
set.addAll(this.syncStateSet);
return set;
public Set<String> getSyncStateSet() {
final Lock readLock = readWriteLock.readLock();
try {
readLock.lock();
HashSet<String> set = new HashSet<>(this.syncStateSet.size());
set.addAll(this.syncStateSet);
return set;
} finally {
readLock.unlock();
}
}
public void truncateEpochFilePrefix(final long offset) {