chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

调整etl锁的粒度

Browse Source
This commit is contained in:
mcy
2018-11-07 09:05:47 +08:00
parent c27d5fa162
commit 202557630c
1 changed files with 6 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java
View File

@@ -56,8 +56,11 @@ public class CommonRest {
@PostMapping("/etl/{type}/{task}") @PostMapping("/etl/{type}/{task}")
public EtlResult etl(@PathVariable String type, @PathVariable String task, public EtlResult etl(@PathVariable String type, @PathVariable String task,
@RequestParam(name = "params", required = false) String params) { @RequestParam(name = "params", required = false) String params) {
OuterAdapter adapter = loader.getExtension(type);
String destination = adapter.getDestination(task);
String lockKey = destination == null ? task : destination;
boolean locked = etlLock.tryLock(ETL_LOCK_ZK_NODE + type + "-" + task); boolean locked = etlLock.tryLock(ETL_LOCK_ZK_NODE + type + "-" + lockKey);
if (!locked) { if (!locked) {
EtlResult result = new EtlResult(); EtlResult result = new EtlResult();
result.setSucceeded(false); result.setSucceeded(false);
@@ -65,8 +68,7 @@ public class CommonRest {
return result; return result;
} }
try { try {
OuterAdapter adapter = loader.getExtension(type);
String destination = adapter.getDestination(task);
Boolean oriSwitchStatus; Boolean oriSwitchStatus;
if (destination != null) { if (destination != null) {
oriSwitchStatus = syncSwitch.status(destination); oriSwitchStatus = syncSwitch.status(destination);
@@ -95,7 +97,7 @@ public class CommonRest {
} }
} }
} finally { } finally {
etlLock.unlock(ETL_LOCK_ZK_NODE + type + "-" + task); etlLock.unlock(ETL_LOCK_ZK_NODE + type + "-" + lockKey);
} }
} }