Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-19686 Use KeyLocker instead of ReentrantLock in PeerProcedureHandlerImpl

This commit is contained in:
zhangduo 2018-01-02 16:13:55 +08:00
parent 5d922a5e0c
commit b0de0ec439
1 changed files with 29 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/PeerProcedureHandlerImpl.java
View File

@ -19,10 +19,10 @@
package org.apache.hadoop.hbase.replication.regionserver; package org.apache.hadoop.hbase.replication.regionserver;
import java.io.IOException; import java.io.IOException;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.Lock;
import org.apache.hadoop.hbase.replication.ReplicationException; import org.apache.hadoop.hbase.replication.ReplicationException;
import org.apache.hadoop.hbase.replication.ReplicationPeer.PeerState; import org.apache.hadoop.hbase.replication.ReplicationPeer.PeerState;
import org.apache.hadoop.hbase.util.KeyLocker;
import org.apache.yetus.audience.InterfaceAudience; import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -32,7 +32,7 @@ public class PeerProcedureHandlerImpl implements PeerProcedureHandler {
private static final Logger LOG = LoggerFactory.getLogger(PeerProcedureHandlerImpl.class); private static final Logger LOG = LoggerFactory.getLogger(PeerProcedureHandlerImpl.class);
private final ReplicationSourceManager replicationSourceManager; private final ReplicationSourceManager replicationSourceManager;
private final ReentrantLock peersLock = new ReentrantLock(); private final KeyLocker<String> peersLock = new KeyLocker<>();
public PeerProcedureHandlerImpl(ReplicationSourceManager replicationSourceManager) { public PeerProcedureHandlerImpl(ReplicationSourceManager replicationSourceManager) {
this.replicationSourceManager = replicationSourceManager; this.replicationSourceManager = replicationSourceManager;
@ -40,40 +40,57 @@ public class PeerProcedureHandlerImpl implements PeerProcedureHandler {
@Override @Override
public void addPeer(String peerId) throws ReplicationException, IOException { public void addPeer(String peerId) throws ReplicationException, IOException {
peersLock.lock(); Lock peerLock = peersLock.acquireLock(peerId);
try { try {
replicationSourceManager.addPeer(peerId); replicationSourceManager.addPeer(peerId);
} finally { } finally {
peersLock.unlock(); peerLock.unlock();
} }
} }
@Override @Override
public void removePeer(String peerId) throws ReplicationException, IOException { public void removePeer(String peerId) throws ReplicationException, IOException {
peersLock.lock(); Lock peerLock = peersLock.acquireLock(peerId);
try { try {
if (replicationSourceManager.getReplicationPeers().getPeer(peerId) != null) { if (replicationSourceManager.getReplicationPeers().getPeer(peerId) != null) {
replicationSourceManager.removePeer(peerId); replicationSourceManager.removePeer(peerId);
} }
} finally { } finally {
peersLock.unlock(); peerLock.unlock();
} }
} }
@Override @Override
public void disablePeer(String peerId) throws ReplicationException, IOException { public void disablePeer(String peerId) throws ReplicationException, IOException {
PeerState newState = replicationSourceManager.getReplicationPeers().refreshPeerState(peerId); PeerState newState;
LOG.info("disable replication peer, id: " + peerId + ", new state: " + newState); Lock peerLock = peersLock.acquireLock(peerId);
try {
newState = replicationSourceManager.getReplicationPeers().refreshPeerState(peerId);
} finally {
peerLock.unlock();
}
LOG.info("disable replication peer, id: {}, new state: {}", peerId, newState);
} }
@Override @Override
public void enablePeer(String peerId) throws ReplicationException, IOException { public void enablePeer(String peerId) throws ReplicationException, IOException {
PeerState newState = replicationSourceManager.getReplicationPeers().refreshPeerState(peerId); PeerState newState;
LOG.info("enable replication peer, id: " + peerId + ", new state: " + newState); Lock peerLock = peersLock.acquireLock(peerId);
try {
newState = replicationSourceManager.getReplicationPeers().refreshPeerState(peerId);
} finally {
peerLock.unlock();
}
LOG.info("enable replication peer, id: {}, new state: {}", peerId, newState);
} }
@Override @Override
public void updatePeerConfig(String peerId) throws ReplicationException, IOException { public void updatePeerConfig(String peerId) throws ReplicationException, IOException {
replicationSourceManager.getReplicationPeers().refreshPeerConfig(peerId); Lock peerLock = peersLock.acquireLock(peerId);
try {
replicationSourceManager.getReplicationPeers().refreshPeerConfig(peerId);
} finally {
peerLock.unlock();
}
} }
} }