Skip to content

Commit

Permalink
fixed #166 concurrency execution error for druid ds pool
Browse files Browse the repository at this point in the history
  • Loading branch information
@hanahmily
hanahmily authored and gaohongtao committed Nov 1, 2016
1 parent 9360049 commit 905b94a
Show file tree
Hide file tree
Showing 6 changed files with 203 additions and 116 deletions.
84 changes: 84 additions & 0 deletions ...ding-jdbc-core/src/main/java/com/dangdang/ddframe/rdb/sharding/executor/EventPostman.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
/*
* Copyright 1999-2015 dangdang.com.
* <p>
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
* </p>
*/

package com.dangdang.ddframe.rdb.sharding.executor;

import com.dangdang.ddframe.rdb.sharding.executor.event.DMLExecutionEvent;
import com.dangdang.ddframe.rdb.sharding.executor.event.DMLExecutionEventBus;
import com.dangdang.ddframe.rdb.sharding.executor.event.DQLExecutionEvent;
import com.dangdang.ddframe.rdb.sharding.executor.event.DQLExecutionEventBus;
import com.dangdang.ddframe.rdb.sharding.executor.event.EventExecutionType;
import com.dangdang.ddframe.rdb.sharding.executor.wrapper.AbstractExecutorWrapper;
import com.dangdang.ddframe.rdb.sharding.executor.wrapper.BatchPreparedStatementExecutorWrapper;
import com.google.common.base.Optional;
import lombok.RequiredArgsConstructor;

import java.sql.SQLException;
import java.util.Collection;

/**
* 消息投递员.
* 负责SQL执行消息的投递
*
* @author gaohongtao.
*/
@RequiredArgsConstructor
class EventPostman {

private final Collection<? extends AbstractExecutorWrapper> statementExecutorWrappers;

void postExecutionEvents() {
for (AbstractExecutorWrapper each : statementExecutorWrappers) {
if (each.getDMLExecutionEvent().isPresent()) {
DMLExecutionEventBus.post(each.getDMLExecutionEvent().get());
} else if (each.getDQLExecutionEvent().isPresent()) {
DQLExecutionEventBus.post(each.getDQLExecutionEvent().get());
}
}
}

void postExecutionEventsAfterExecution(final AbstractExecutorWrapper statementExecutorWrapper) {
postExecutionEventsAfterExecution(statementExecutorWrapper, EventExecutionType.EXECUTE_SUCCESS, Optional.<SQLException>absent());
}

void postExecutionEventsAfterExecution(final AbstractExecutorWrapper statementExecutorWrapper, final EventExecutionType eventExecutionType, final Optional<SQLException> exp) {
if (statementExecutorWrapper.getDMLExecutionEvent().isPresent()) {
DMLExecutionEvent event = statementExecutorWrapper.getDMLExecutionEvent().get();
event.setEventExecutionType(eventExecutionType);
event.setExp(exp);
DMLExecutionEventBus.post(event);
} else if (statementExecutorWrapper.getDQLExecutionEvent().isPresent()) {
DQLExecutionEvent event = statementExecutorWrapper.getDQLExecutionEvent().get();
event.setEventExecutionType(eventExecutionType);
event.setExp(exp);
DQLExecutionEventBus.post(event);
}
}

void postBatchExecutionEventsAfterExecution(final BatchPreparedStatementExecutorWrapper batchPreparedStatementExecutorWrapper) {
postBatchExecutionEventsAfterExecution(batchPreparedStatementExecutorWrapper, EventExecutionType.EXECUTE_SUCCESS, Optional.<SQLException>absent());
}

void postBatchExecutionEventsAfterExecution(
final BatchPreparedStatementExecutorWrapper batchPreparedStatementExecutorWrapper, final EventExecutionType eventExecutionType, final Optional<SQLException> exp) {
for (DMLExecutionEvent each : batchPreparedStatementExecutorWrapper.getDmlExecutionEvents()) {
each.setEventExecutionType(eventExecutionType);
each.setExp(exp);
DMLExecutionEventBus.post(each);
}
}
}
106 changes: 34 additions & 72 deletions ...e/src/main/java/com/dangdang/ddframe/rdb/sharding/executor/PreparedStatementExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,16 +18,11 @@
package com.dangdang.ddframe.rdb.sharding.executor;

import com.codahale.metrics.Timer.Context;
import com.dangdang.ddframe.rdb.sharding.executor.event.DMLExecutionEvent;
import com.dangdang.ddframe.rdb.sharding.executor.event.DMLExecutionEventBus;
import com.dangdang.ddframe.rdb.sharding.executor.event.DQLExecutionEvent;
import com.dangdang.ddframe.rdb.sharding.executor.event.DQLExecutionEventBus;
import com.dangdang.ddframe.rdb.sharding.executor.event.EventExecutionType;
import com.dangdang.ddframe.rdb.sharding.executor.wrapper.BatchPreparedStatementExecutorWrapper;
import com.dangdang.ddframe.rdb.sharding.executor.wrapper.PreparedStatementExecutorWrapper;
import com.dangdang.ddframe.rdb.sharding.metrics.MetricsContext;
import com.google.common.base.Optional;
import lombok.RequiredArgsConstructor;

import java.sql.ResultSet;
import java.sql.SQLException;
Expand All @@ -42,21 +37,28 @@
* @author zhangliang
* @author caohao
*/
@RequiredArgsConstructor
public final class PreparedStatementExecutor {

private final ExecutorEngine executorEngine;

private final Collection<PreparedStatementExecutorWrapper> preparedStatementExecutorWrappers;

private final EventPostman eventPostman;

public PreparedStatementExecutor(final ExecutorEngine executorEngine, final Collection<PreparedStatementExecutorWrapper> preparedStatementExecutorWrappers) {
this.executorEngine = executorEngine;
this.preparedStatementExecutorWrappers = preparedStatementExecutorWrappers;
this.eventPostman = new EventPostman(preparedStatementExecutorWrappers);
}

/**
* 执行SQL查询.
*
* @return 结果集列表
*/
public List<ResultSet> executeQuery() {
Context context = MetricsContext.start("ShardingPreparedStatement-executeQuery");
postExecutionEvents();
eventPostman.postExecutionEvents();
List<ResultSet> result;
final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
Expand All @@ -68,7 +70,9 @@ public List<ResultSet> executeQuery() {

@Override
public ResultSet execute(final PreparedStatementExecutorWrapper input) throws Exception {
return executeQueryInternal(input, isExceptionThrown, dataMap);
synchronized (input.getPreparedStatement().getConnection()) {
return executeQueryInternal(input, isExceptionThrown, dataMap);
}
}
});
} finally {
Expand All @@ -85,11 +89,11 @@ private ResultSet executeQueryInternal(final PreparedStatementExecutorWrapper pr
try {
result = preparedStatementExecutorWrapper.getPreparedStatement().executeQuery();
} catch (final SQLException ex) {
postExecutionEventsAfterExecution(preparedStatementExecutorWrapper, EventExecutionType.EXECUTE_FAILURE, Optional.of(ex));
eventPostman.postExecutionEventsAfterExecution(preparedStatementExecutorWrapper, EventExecutionType.EXECUTE_FAILURE, Optional.of(ex));
ExecutorExceptionHandler.handleException(ex);
return null;
}
postExecutionEventsAfterExecution(preparedStatementExecutorWrapper);
eventPostman.postExecutionEventsAfterExecution(preparedStatementExecutorWrapper);
return result;
}

Expand All @@ -100,7 +104,7 @@ private ResultSet executeQueryInternal(final PreparedStatementExecutorWrapper pr
*/
public int executeUpdate() {
Context context = MetricsContext.start("ShardingPreparedStatement-executeUpdate");
postExecutionEvents();
eventPostman.postExecutionEvents();
final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
try {
Expand All @@ -111,7 +115,9 @@ public int executeUpdate() {

@Override
public Integer execute(final PreparedStatementExecutorWrapper input) throws Exception {
return executeUpdateInternal(input, isExceptionThrown, dataMap);
synchronized (input.getPreparedStatement().getConnection()) {
return executeUpdateInternal(input, isExceptionThrown, dataMap);
}
}
}, new MergeUnit<Integer, Integer>() {

Expand Down Expand Up @@ -140,11 +146,11 @@ private int executeUpdateInternal(final PreparedStatementExecutorWrapper prepare
try {
result = preparedStatementExecutorWrapper.getPreparedStatement().executeUpdate();
} catch (final SQLException ex) {
postExecutionEventsAfterExecution(preparedStatementExecutorWrapper, EventExecutionType.EXECUTE_FAILURE, Optional.of(ex));
eventPostman.postExecutionEventsAfterExecution(preparedStatementExecutorWrapper, EventExecutionType.EXECUTE_FAILURE, Optional.of(ex));
ExecutorExceptionHandler.handleException(ex);
return 0;
}
postExecutionEventsAfterExecution(preparedStatementExecutorWrapper);
eventPostman.postExecutionEventsAfterExecution(preparedStatementExecutorWrapper);
return result;
}

Expand All @@ -155,19 +161,21 @@ private int executeUpdateInternal(final PreparedStatementExecutorWrapper prepare
*/
public boolean execute() {
Context context = MetricsContext.start("ShardingPreparedStatement-execute");
postExecutionEvents();
eventPostman.postExecutionEvents();
final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
try {
if (1 == preparedStatementExecutorWrappers.size()) {
PreparedStatementExecutorWrapper preparedStatementExecutorWrapper = preparedStatementExecutorWrappers.iterator().next();
return executeInternal(preparedStatementExecutorWrapper, isExceptionThrown, dataMap, Optional.fromNullable(context));
return executeInternal(preparedStatementExecutorWrapper, isExceptionThrown, dataMap);
}
List<Boolean> result = executorEngine.execute(preparedStatementExecutorWrappers, new ExecuteUnit<PreparedStatementExecutorWrapper, Boolean>() {

@Override
public Boolean execute(final PreparedStatementExecutorWrapper input) throws Exception {
return executeInternal(input, isExceptionThrown, dataMap, Optional.<Context>absent());
synchronized (input.getPreparedStatement().getConnection()) {
return executeInternal(input, isExceptionThrown, dataMap);
}
}
});
return (null == result || result.isEmpty()) ? false : result.get(0);
Expand All @@ -177,22 +185,18 @@ public Boolean execute(final PreparedStatementExecutorWrapper input) throws Exce
}

private boolean executeInternal(final PreparedStatementExecutorWrapper preparedStatementExecutorWrapper,
final boolean isExceptionThrown, final Map<String, Object> dataMap, final Optional<Context> context) {
final boolean isExceptionThrown, final Map<String, Object> dataMap) {
boolean result;
ExecutorExceptionHandler.setExceptionThrown(isExceptionThrown);
ExecutorDataMap.setDataMap(dataMap);
try {
result = preparedStatementExecutorWrapper.getPreparedStatement().execute();
} catch (final SQLException ex) {
postExecutionEventsAfterExecution(preparedStatementExecutorWrapper, EventExecutionType.EXECUTE_FAILURE, Optional.of(ex));
eventPostman.postExecutionEventsAfterExecution(preparedStatementExecutorWrapper, EventExecutionType.EXECUTE_FAILURE, Optional.of(ex));
ExecutorExceptionHandler.handleException(ex);
return false;
} finally {
if (context.isPresent()) {
MetricsContext.stop(context.get());
}
}
postExecutionEventsAfterExecution(preparedStatementExecutorWrapper);
eventPostman.postExecutionEventsAfterExecution(preparedStatementExecutorWrapper);
return result;
}

Expand All @@ -204,7 +208,7 @@ private boolean executeInternal(final PreparedStatementExecutorWrapper preparedS
*/
public int[] executeBatch() {
Context context = MetricsContext.start("ShardingPreparedStatement-executeUpdate");
postExecutionEvents();
eventPostman.postExecutionEvents();
final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
try {
Expand All @@ -215,7 +219,9 @@ public int[] executeBatch() {

@Override
public int[] execute(final PreparedStatementExecutorWrapper input) throws Exception {
return executeBatchInternal((BatchPreparedStatementExecutorWrapper) input, isExceptionThrown, dataMap);
synchronized (input.getPreparedStatement().getConnection()) {
return executeBatchInternal((BatchPreparedStatementExecutorWrapper) input, isExceptionThrown, dataMap);
}
}
}, new MergeUnit<int[], int[]>() {

Expand Down Expand Up @@ -249,55 +255,11 @@ private int[] executeBatchInternal(final BatchPreparedStatementExecutorWrapper b
try {
result = batchPreparedStatementExecutorWrapper.getPreparedStatement().executeBatch();
} catch (final SQLException ex) {
postBatchExecutionEventsAfterExecution(batchPreparedStatementExecutorWrapper, EventExecutionType.EXECUTE_FAILURE, Optional.of(ex));
eventPostman.postBatchExecutionEventsAfterExecution(batchPreparedStatementExecutorWrapper, EventExecutionType.EXECUTE_FAILURE, Optional.of(ex));
ExecutorExceptionHandler.handleException(ex);
return null;
}
postBatchExecutionEventsAfterExecution(batchPreparedStatementExecutorWrapper);
eventPostman.postBatchExecutionEventsAfterExecution(batchPreparedStatementExecutorWrapper);
return result;
}

private void postBatchExecutionEventsAfterExecution(final BatchPreparedStatementExecutorWrapper batchPreparedStatementExecutorWrapper) {
postBatchExecutionEventsAfterExecution(batchPreparedStatementExecutorWrapper, EventExecutionType.EXECUTE_SUCCESS, Optional.<SQLException>absent());
}

private void postBatchExecutionEventsAfterExecution(
final BatchPreparedStatementExecutorWrapper batchPreparedStatementExecutorWrapper, final EventExecutionType eventExecutionType, final Optional<SQLException> exp) {
for (DMLExecutionEvent each : batchPreparedStatementExecutorWrapper.getDmlExecutionEvents()) {
each.setEventExecutionType(eventExecutionType);
each.setExp(exp);
DMLExecutionEventBus.post(each);
}
}

private void postExecutionEvents() {
for (PreparedStatementExecutorWrapper each : preparedStatementExecutorWrappers) {
if (each.getDMLExecutionEvent().isPresent()) {
DMLExecutionEventBus.post(each.getDMLExecutionEvent().get());
}
if (each.getDQLExecutionEvent().isPresent()) {
DQLExecutionEventBus.post(each.getDQLExecutionEvent().get());
}
}
}

private void postExecutionEventsAfterExecution(final PreparedStatementExecutorWrapper preparedStatementExecutorWrapper) {
postExecutionEventsAfterExecution(preparedStatementExecutorWrapper, EventExecutionType.EXECUTE_SUCCESS, Optional.<SQLException>absent());
}

private void postExecutionEventsAfterExecution(final PreparedStatementExecutorWrapper preparedStatementExecutorWrapper,
final EventExecutionType eventExecutionType, final Optional<SQLException> exp) {
if (preparedStatementExecutorWrapper.getDMLExecutionEvent().isPresent()) {
DMLExecutionEvent event = preparedStatementExecutorWrapper.getDMLExecutionEvent().get();
event.setEventExecutionType(eventExecutionType);
event.setExp(exp);
DMLExecutionEventBus.post(event);
}
if (preparedStatementExecutorWrapper.getDQLExecutionEvent().isPresent()) {
DQLExecutionEvent event = preparedStatementExecutorWrapper.getDQLExecutionEvent().get();
event.setEventExecutionType(eventExecutionType);
event.setExp(exp);
DQLExecutionEventBus.post(event);
}
}
}
Loading

0 comments on commit 905b94a

Please sign in to comment.