提交 0b39fcad 编写于 作者: Y yukon

[ROCKETMQ-145][HOTFIX] Resolve concureent issue in HAService and GroupCommitService

上级 11653ce2
...@@ -1101,12 +1101,12 @@ public class CommitLog { ...@@ -1101,12 +1101,12 @@ public class CommitLog {
private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>(); private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>();
private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>(); private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>();
public void putRequest(final GroupCommitRequest request) { public synchronized void putRequest(final GroupCommitRequest request) {
synchronized (this) { synchronized (this.requestsWrite) {
this.requestsWrite.add(request); this.requestsWrite.add(request);
if (hasNotified.compareAndSet(false, true)) { }
waitPoint.countDown(); // notify if (hasNotified.compareAndSet(false, true)) {
} waitPoint.countDown(); // notify
} }
} }
...@@ -1117,32 +1117,34 @@ public class CommitLog { ...@@ -1117,32 +1117,34 @@ public class CommitLog {
} }
private void doCommit() { private void doCommit() {
if (!this.requestsRead.isEmpty()) { synchronized (this.requestsRead) {
for (GroupCommitRequest req : this.requestsRead) { if (!this.requestsRead.isEmpty()) {
// There may be a message in the next file, so a maximum of for (GroupCommitRequest req : this.requestsRead) {
// two times the flush // There may be a message in the next file, so a maximum of
boolean flushOK = false; // two times the flush
for (int i = 0; i < 2 && !flushOK; i++) { boolean flushOK = false;
flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset(); for (int i = 0; i < 2 && !flushOK; i++) {
flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
if (!flushOK) {
CommitLog.this.mappedFileQueue.flush(0); if (!flushOK) {
CommitLog.this.mappedFileQueue.flush(0);
}
} }
req.wakeupCustomer(flushOK);
} }
req.wakeupCustomer(flushOK); long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
} if (storeTimestamp > 0) {
CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
}
long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp(); this.requestsRead.clear();
if (storeTimestamp > 0) { } else {
CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp); // Because of individual messages is set to not sync flush, it
// will come to this process
CommitLog.this.mappedFileQueue.flush(0);
} }
this.requestsRead.clear();
} else {
// Because of individual messages is set to not sync flush, it
// will come to this process
CommitLog.this.mappedFileQueue.flush(0);
} }
} }
......
...@@ -253,12 +253,12 @@ public class HAService { ...@@ -253,12 +253,12 @@ public class HAService {
private volatile List<CommitLog.GroupCommitRequest> requestsWrite = new ArrayList<>(); private volatile List<CommitLog.GroupCommitRequest> requestsWrite = new ArrayList<>();
private volatile List<CommitLog.GroupCommitRequest> requestsRead = new ArrayList<>(); private volatile List<CommitLog.GroupCommitRequest> requestsRead = new ArrayList<>();
public void putRequest(final CommitLog.GroupCommitRequest request) { public synchronized void putRequest(final CommitLog.GroupCommitRequest request) {
synchronized (this) { synchronized (this.requestsWrite) {
this.requestsWrite.add(request); this.requestsWrite.add(request);
if (hasNotified.compareAndSet(false, true)) { }
waitPoint.countDown(); // notify if (hasNotified.compareAndSet(false, true)) {
} waitPoint.countDown(); // notify
} }
} }
...@@ -273,22 +273,24 @@ public class HAService { ...@@ -273,22 +273,24 @@ public class HAService {
} }
private void doWaitTransfer() { private void doWaitTransfer() {
if (!this.requestsRead.isEmpty()) { synchronized (this.requestsRead) {
for (CommitLog.GroupCommitRequest req : this.requestsRead) { if (!this.requestsRead.isEmpty()) {
boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset(); for (CommitLog.GroupCommitRequest req : this.requestsRead) {
for (int i = 0; !transferOK && i < 5; i++) { boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
this.notifyTransferObject.waitForRunning(1000); for (int i = 0; !transferOK && i < 5; i++) {
transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset(); this.notifyTransferObject.waitForRunning(1000);
} transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
}
if (!transferOK) { if (!transferOK) {
log.warn("transfer messsage to slave timeout, " + req.getNextOffset()); log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
}
req.wakeupCustomer(transferOK);
} }
req.wakeupCustomer(transferOK); this.requestsRead.clear();
} }
this.requestsRead.clear();
} }
} }
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册