Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Proxy handle heart beat event #control-panel-cluster (#5795)
* proxy detect heart beat * revise log * revise class name * optimize code * for checkstyle
- Loading branch information
1 parent
61e8068
commit 4b192eb
Showing
9 changed files
with
308 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
79 changes: 79 additions & 0 deletions
79
...rc/main/java/org/apache/shardingsphere/proxy/backend/cluster/AbstractHeartbeatDetect.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You 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. | ||
| */ | ||
|
|
||
| package org.apache.shardingsphere.proxy.backend.cluster; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.shardingsphere.cluster.heartbeat.response.HeartbeatResult; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.concurrent.Callable; | ||
|
|
||
| /** | ||
| * Abstract heart beat detect. | ||
| */ | ||
| @Slf4j | ||
| public abstract class AbstractHeartbeatDetect implements Callable<Map<String, HeartbeatResult>> { | ||
|
|
||
| private Boolean retryEnable; | ||
|
|
||
| private Integer retryMaximum; | ||
|
|
||
| private Integer retryInterval; | ||
|
|
||
| public AbstractHeartbeatDetect(final Boolean retryEnable, final Integer retryMaximum, final Integer retryInterval) { | ||
| this.retryEnable = retryEnable; | ||
| this.retryMaximum = retryMaximum; | ||
| this.retryInterval = retryInterval; | ||
| } | ||
|
|
||
| /** | ||
| * Detect heart beat. | ||
| * | ||
| * @return heart beat result. | ||
| */ | ||
| protected abstract Boolean detect(); | ||
|
|
||
| /** | ||
| * Build heart beat result. | ||
| * | ||
| * @param result heart beat result | ||
| * @return heart beat result | ||
| */ | ||
| protected abstract Map<String, HeartbeatResult> buildResult(Boolean result); | ||
|
|
||
| @Override | ||
| public Map<String, HeartbeatResult> call() { | ||
| if (retryEnable && retryMaximum > 0) { | ||
| Boolean result = Boolean.FALSE; | ||
| for (int i = 0; i < retryMaximum; i++) { | ||
| result = detect(); | ||
| if (result) { | ||
| break; | ||
| } | ||
| try { | ||
| Thread.sleep(retryInterval * 1000); | ||
| } catch (InterruptedException ex) { | ||
| log.warn("Retry heart beat detect sleep error", ex); | ||
| } | ||
| } | ||
| return buildResult(result); | ||
| } else { | ||
| return buildResult(detect()); | ||
| } | ||
| } | ||
| } |
72 changes: 72 additions & 0 deletions
72
...ackend/src/main/java/org/apache/shardingsphere/proxy/backend/cluster/HeartbeatDetect.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You 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. | ||
| */ | ||
|
|
||
| package org.apache.shardingsphere.proxy.backend.cluster; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.shardingsphere.cluster.configuration.config.HeartbeatConfiguration; | ||
| import org.apache.shardingsphere.cluster.heartbeat.response.HeartbeatResult; | ||
|
|
||
| import javax.sql.DataSource; | ||
| import java.sql.PreparedStatement; | ||
| import java.sql.ResultSet; | ||
| import java.sql.SQLException; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.Objects; | ||
|
|
||
| /** | ||
| * Heart beat detect. | ||
| */ | ||
| @Slf4j | ||
| public final class HeartbeatDetect extends AbstractHeartbeatDetect { | ||
|
|
||
| private String sql; | ||
|
|
||
| private String schemaName; | ||
|
|
||
| private String dataSourceName; | ||
|
|
||
| private DataSource dataSource; | ||
|
|
||
| public HeartbeatDetect(final String schemaName, final String dataSourceName, final DataSource dataSource, final HeartbeatConfiguration configuration) { | ||
| super(configuration.getRetryEnable(), configuration.getRetryMaximum(), configuration.getRetryInterval()); | ||
| this.sql = configuration.getSql(); | ||
| this.schemaName = schemaName; | ||
| this.dataSourceName = dataSourceName; | ||
| this.dataSource = dataSource; | ||
| } | ||
|
|
||
| @Override | ||
| protected Boolean detect() { | ||
| try { | ||
| PreparedStatement preparedStatement = dataSource.getConnection().prepareStatement(sql); | ||
| ResultSet result = preparedStatement.executeQuery(); | ||
| return Objects.nonNull(result) && result.next(); | ||
| } catch (SQLException ex) { | ||
| log.error("Heart beat detect error", ex); | ||
| } | ||
| return Boolean.FALSE; | ||
| } | ||
|
|
||
| @Override | ||
| protected Map<String, HeartbeatResult> buildResult(final Boolean result) { | ||
| Map<String, HeartbeatResult> heartBeatResultMap = new HashMap<>(1, 1); | ||
| heartBeatResultMap.put(schemaName, new HeartbeatResult(dataSourceName, result, System.currentTimeMillis())); | ||
| return heartBeatResultMap; | ||
| } | ||
| } |
121 changes: 121 additions & 0 deletions
121
...ckend/src/main/java/org/apache/shardingsphere/proxy/backend/cluster/HeartbeatHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You 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. | ||
| */ | ||
|
|
||
| package org.apache.shardingsphere.proxy.backend.cluster; | ||
|
|
||
| import com.google.common.base.Preconditions; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.shardingsphere.cluster.configuration.config.HeartbeatConfiguration; | ||
|
|
||
| import org.apache.shardingsphere.cluster.facade.ClusterFacade; | ||
| import org.apache.shardingsphere.cluster.heartbeat.response.HeartbeatResponse; | ||
| import org.apache.shardingsphere.cluster.heartbeat.response.HeartbeatResult; | ||
| import org.apache.shardingsphere.proxy.backend.schema.ShardingSphereSchema; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Arrays; | ||
| import java.util.Objects; | ||
| import java.util.Collection; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.FutureTask; | ||
| import java.util.concurrent.ExecutionException; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * Heart beat handler. | ||
| */ | ||
| @Slf4j | ||
| public final class HeartbeatHandler { | ||
|
|
||
| private HeartbeatConfiguration configuration; | ||
|
|
||
| /** | ||
| * Init heart beat handler. | ||
| * | ||
| * @param configuration heart beat configuration | ||
| */ | ||
| public void init(final HeartbeatConfiguration configuration) { | ||
| Preconditions.checkNotNull(configuration, "heart beat configuration can not be null."); | ||
| this.configuration = configuration; | ||
| } | ||
|
|
||
| /** | ||
| * Get heart beat handler instance. | ||
| * | ||
| * @return heart beat handler instance | ||
| */ | ||
| public static HeartbeatHandler getInstance() { | ||
| return HeartbeatHandlerHolder.INSTANCE; | ||
| } | ||
|
|
||
| /** | ||
| * Handle heart beat detect event. | ||
| * | ||
| * @param schemas ShardingSphere schemas | ||
| */ | ||
| public void handle(final Map<String, ShardingSphereSchema> schemas) { | ||
| ExecutorService executorService = Executors.newFixedThreadPool(countDataSource(schemas)); | ||
| List<FutureTask<Map<String, HeartbeatResult>>> futureTasks = new ArrayList<>(); | ||
| schemas.values().forEach(value -> value.getBackendDataSource().getDataSources().entrySet().forEach(entry -> { | ||
| FutureTask<Map<String, HeartbeatResult>> futureTask = new FutureTask<>(new HeartbeatDetect(value.getName(), entry.getKey(), | ||
| entry.getValue(), configuration)); | ||
| futureTasks.add(futureTask); | ||
| executorService.submit(futureTask); | ||
| })); | ||
| reportHeartbeat(futureTasks); | ||
| closeExecutor(executorService); | ||
| } | ||
|
|
||
| private Integer countDataSource(final Map<String, ShardingSphereSchema> schemas) { | ||
| return Long.valueOf(schemas.values().stream() | ||
| .collect(Collectors.summarizingInt(entry -> entry.getBackendDataSource() | ||
| .getDataSources().keySet().size())).getSum()).intValue(); | ||
| } | ||
|
|
||
| private void reportHeartbeat(final List<FutureTask<Map<String, HeartbeatResult>>> futureTasks) { | ||
| Map<String, Collection<HeartbeatResult>> heartbeatResultMap = new HashMap<>(); | ||
| futureTasks.stream().forEach(each -> { | ||
| try { | ||
| each.get().entrySet().forEach(entry -> { | ||
| if (Objects.isNull(heartbeatResultMap.get(entry.getKey()))) { | ||
| heartbeatResultMap.put(entry.getKey(), new ArrayList<>(Arrays.asList(entry.getValue()))); | ||
| } else { | ||
| heartbeatResultMap.get(entry.getKey()).add(entry.getValue()); | ||
| } | ||
| }); | ||
| } catch (InterruptedException | ExecutionException ex) { | ||
| log.error("Heart beat report error", ex); | ||
| } | ||
| }); | ||
| ClusterFacade.getInstance().reportHeartbeat(new HeartbeatResponse(heartbeatResultMap)); | ||
| } | ||
|
|
||
| private void closeExecutor(final ExecutorService executorService) { | ||
| if (null != executorService && !executorService.isShutdown()) { | ||
| executorService.shutdown(); | ||
| } | ||
| } | ||
|
|
||
| private static final class HeartbeatHandlerHolder { | ||
|
|
||
| public static final HeartbeatHandler INSTANCE = new HeartbeatHandler(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters