mirror of https://github.com/longtai-cn/hippo4j
Add spring cloud stream rocketmq thread pool docking (#215)
chen.ma
2 years ago
parent
7d014306c4
commit
29baf830e4
@ -0,0 +1,59 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.adapter.base;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Thread-pool adapter cache config.
|
||||
*/
|
||||
@Data
|
||||
public class ThreadPoolAdapterCacheConfig {
|
||||
|
||||
/**
|
||||
* Mark
|
||||
*/
|
||||
private String mark;
|
||||
|
||||
/**
|
||||
* Tenant item key
|
||||
*/
|
||||
private String tenantItemKey;
|
||||
|
||||
/**
|
||||
* Client identify
|
||||
*/
|
||||
private String clientIdentify;
|
||||
|
||||
/**
|
||||
* Active
|
||||
*/
|
||||
private String active;
|
||||
|
||||
/**
|
||||
* Client address
|
||||
*/
|
||||
private String clientAddress;
|
||||
|
||||
/**
|
||||
* Thread-pool adapter states
|
||||
*/
|
||||
private List<ThreadPoolAdapterState> threadPoolAdapterStates;
|
||||
}
|
||||
@ -0,0 +1,47 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.config.controller;
|
||||
|
||||
import cn.hippo4j.adapter.base.ThreadPoolAdapterCacheConfig;
|
||||
import cn.hippo4j.common.web.base.Result;
|
||||
import cn.hippo4j.common.web.base.Results;
|
||||
import cn.hippo4j.config.service.ThreadPoolAdapterService;
|
||||
import lombok.AllArgsConstructor;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static cn.hippo4j.common.constant.Constants.REGISTER_ADAPTER_PATH;
|
||||
|
||||
/**
|
||||
* Thread-pool adapter controller.
|
||||
*/
|
||||
@RestController
|
||||
@AllArgsConstructor
|
||||
public class ThreadPoolAdapterController {
|
||||
|
||||
private final ThreadPoolAdapterService threadPoolAdapterService;
|
||||
|
||||
@PostMapping(REGISTER_ADAPTER_PATH)
|
||||
public Result registerAdapterThreadPool(@RequestBody List<ThreadPoolAdapterCacheConfig> requestParameter) {
|
||||
threadPoolAdapterService.register(requestParameter);
|
||||
return Results.success();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,69 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.config.model.biz.adapter;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Thread-pool adapter req DTO.
|
||||
*/
|
||||
@Data
|
||||
public class ThreadPoolAdapterReqDTO {
|
||||
|
||||
/**
|
||||
* Mark
|
||||
*/
|
||||
private String mark;
|
||||
|
||||
/**
|
||||
* Tenant
|
||||
*/
|
||||
private String tenant;
|
||||
|
||||
/**
|
||||
* Item
|
||||
*/
|
||||
private String item;
|
||||
|
||||
/**
|
||||
* Thread pool key
|
||||
*/
|
||||
private String threadPoolKey;
|
||||
|
||||
/**
|
||||
* Identify
|
||||
*/
|
||||
private String identify;
|
||||
|
||||
/**
|
||||
* Core size
|
||||
*/
|
||||
private Integer coreSize;
|
||||
|
||||
/**
|
||||
* Maximum size
|
||||
*/
|
||||
private Integer maximumSize;
|
||||
|
||||
/**
|
||||
* Client address list
|
||||
*/
|
||||
private List<String> clientAddressList;
|
||||
}
|
||||
@ -0,0 +1,57 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.config.model.biz.adapter;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* Thread-pool adapter resp DTO.
|
||||
*/
|
||||
@Data
|
||||
public class ThreadPoolAdapterRespDTO {
|
||||
|
||||
/**
|
||||
* Identify
|
||||
*/
|
||||
private String identify;
|
||||
|
||||
/**
|
||||
* Active
|
||||
*/
|
||||
private String active;
|
||||
|
||||
/**
|
||||
* Client address
|
||||
*/
|
||||
private String clientAddress;
|
||||
|
||||
/**
|
||||
* Thread pool key
|
||||
*/
|
||||
private String threadPoolKey;
|
||||
|
||||
/**
|
||||
* Core size
|
||||
*/
|
||||
private Integer coreSize;
|
||||
|
||||
/**
|
||||
* Maximum size
|
||||
*/
|
||||
private Integer maximumSize;
|
||||
}
|
||||
@ -0,0 +1,102 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.config.service;
|
||||
|
||||
import cn.hippo4j.adapter.base.ThreadPoolAdapterCacheConfig;
|
||||
import cn.hippo4j.adapter.base.ThreadPoolAdapterState;
|
||||
import cn.hippo4j.common.toolkit.CollectionUtil;
|
||||
import cn.hippo4j.common.toolkit.JSONUtil;
|
||||
import cn.hippo4j.common.toolkit.StringUtil;
|
||||
import cn.hippo4j.common.web.base.Result;
|
||||
import cn.hippo4j.config.model.biz.adapter.ThreadPoolAdapterReqDTO;
|
||||
import cn.hippo4j.config.model.biz.adapter.ThreadPoolAdapterRespDTO;
|
||||
import cn.hutool.core.text.StrBuilder;
|
||||
import cn.hutool.http.HttpUtil;
|
||||
import com.google.common.collect.Lists;
|
||||
import com.google.common.collect.Maps;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
|
||||
import static cn.hippo4j.common.constant.Constants.HTTP_EXECUTE_TIMEOUT;
|
||||
import static cn.hippo4j.common.constant.Constants.IDENTIFY_SLICER_SYMBOL;
|
||||
|
||||
/**
|
||||
* Thread-pool adapter service.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class ThreadPoolAdapterService {
|
||||
|
||||
/**
|
||||
* Map<mark, Map<tenantItem, Map<threadPoolKey, List<String>>>>
|
||||
*/
|
||||
private final Map<String, Map<String, Map<String, List<String>>>> THREAD_POOL_ADAPTER_MAP = Maps.newConcurrentMap();
|
||||
|
||||
public synchronized void register(List<ThreadPoolAdapterCacheConfig> requestParameter) {
|
||||
for (ThreadPoolAdapterCacheConfig each : requestParameter) {
|
||||
String mark = each.getMark();
|
||||
Map<String, Map<String, List<String>>> actual = THREAD_POOL_ADAPTER_MAP.get(mark);
|
||||
if (CollectionUtil.isEmpty(actual)) {
|
||||
actual = Maps.newHashMap();
|
||||
THREAD_POOL_ADAPTER_MAP.put(mark, actual);
|
||||
}
|
||||
Map<String, List<String>> tenantItemMap = actual.get(each.getTenantItemKey());
|
||||
if (CollectionUtil.isEmpty(tenantItemMap)) {
|
||||
tenantItemMap = Maps.newHashMap();
|
||||
actual.put(each.getTenantItemKey(), tenantItemMap);
|
||||
}
|
||||
List<ThreadPoolAdapterState> threadPoolAdapterStates = each.getThreadPoolAdapterStates();
|
||||
for (ThreadPoolAdapterState adapterState : threadPoolAdapterStates) {
|
||||
List<String> threadPoolKeyList = tenantItemMap.get(adapterState.getThreadPoolKey());
|
||||
if (CollectionUtil.isEmpty(threadPoolKeyList)) {
|
||||
threadPoolKeyList = Lists.newArrayList();
|
||||
tenantItemMap.put(adapterState.getThreadPoolKey(), threadPoolKeyList);
|
||||
}
|
||||
threadPoolKeyList.add(each.getClientAddress());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public List<ThreadPoolAdapterRespDTO> query(ThreadPoolAdapterReqDTO requestParameter) {
|
||||
List<String> actual = Optional.ofNullable(THREAD_POOL_ADAPTER_MAP.get(requestParameter.getMark()))
|
||||
.map(each -> each.get(requestParameter.getTenant() + IDENTIFY_SLICER_SYMBOL + requestParameter.getItem()))
|
||||
.map(each -> each.get(requestParameter.getThreadPoolKey()))
|
||||
.orElse(Lists.newArrayList());
|
||||
List<ThreadPoolAdapterRespDTO> result = Lists.newCopyOnWriteArrayList();
|
||||
actual.parallelStream().forEach(each -> {
|
||||
String urlString = StrBuilder.create("http://", each, "/adapter/thread-pool/info").toString();
|
||||
Map<String, Object> param = Maps.newHashMap();
|
||||
param.put("mark", requestParameter.getMark());
|
||||
param.put("threadPoolKey", requestParameter.getThreadPoolKey());
|
||||
try {
|
||||
String resultStr = HttpUtil.get(urlString, param, HTTP_EXECUTE_TIMEOUT);
|
||||
if (StringUtil.isNotBlank(resultStr)) {
|
||||
Result<ThreadPoolAdapterRespDTO> restResult = JSONUtil.parseObject(resultStr, Result.class);
|
||||
result.add(restResult.getData());
|
||||
}
|
||||
} catch (Throwable ex) {
|
||||
log.error("Failed to get third-party thread pool data.", ex);
|
||||
}
|
||||
});
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,63 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.console.controller;
|
||||
|
||||
import cn.hippo4j.common.toolkit.JSONUtil;
|
||||
import cn.hippo4j.common.web.base.Result;
|
||||
import cn.hippo4j.common.web.base.Results;
|
||||
import cn.hippo4j.config.model.biz.adapter.ThreadPoolAdapterReqDTO;
|
||||
import cn.hippo4j.config.model.biz.adapter.ThreadPoolAdapterRespDTO;
|
||||
import cn.hippo4j.config.service.ThreadPoolAdapterService;
|
||||
import cn.hutool.core.text.StrBuilder;
|
||||
import cn.hutool.http.HttpUtil;
|
||||
import lombok.AllArgsConstructor;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static cn.hippo4j.common.constant.Constants.HTTP_EXECUTE_TIMEOUT;
|
||||
import static cn.hippo4j.common.constant.Constants.REGISTER_ADAPTER_BASE_PATH;
|
||||
|
||||
/**
|
||||
* Thread-pool adapter controller.
|
||||
*/
|
||||
@AllArgsConstructor
|
||||
@RestController("threadPoolAdapterConsoleController")
|
||||
public class ThreadPoolAdapterController {
|
||||
|
||||
private final ThreadPoolAdapterService threadPoolAdapterService;
|
||||
|
||||
@GetMapping(REGISTER_ADAPTER_BASE_PATH + "/query")
|
||||
public Result<List<ThreadPoolAdapterRespDTO>> queryAdapterThreadPool(ThreadPoolAdapterReqDTO requestParameter) {
|
||||
List<ThreadPoolAdapterRespDTO> result = threadPoolAdapterService.query(requestParameter);
|
||||
return Results.success(result);
|
||||
}
|
||||
|
||||
@PostMapping(REGISTER_ADAPTER_BASE_PATH + "/update")
|
||||
public Result<Void> updateAdapterThreadPool(@RequestBody ThreadPoolAdapterReqDTO requestParameter) {
|
||||
for (String each : requestParameter.getClientAddressList()) {
|
||||
String urlString = StrBuilder.create("http://", each, "/adapter/thread-pool/update").toString();
|
||||
HttpUtil.post(urlString, JSONUtil.toJSONString(requestParameter), HTTP_EXECUTE_TIMEOUT);
|
||||
}
|
||||
return Results.success();
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,82 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.springboot.starter.core;
|
||||
|
||||
import cn.hippo4j.adapter.base.ThreadPoolAdapter;
|
||||
import cn.hippo4j.adapter.base.ThreadPoolAdapterCacheConfig;
|
||||
import cn.hippo4j.common.config.ApplicationContextHolder;
|
||||
import cn.hippo4j.common.toolkit.CollectionUtil;
|
||||
import cn.hippo4j.common.web.base.Result;
|
||||
import cn.hippo4j.core.toolkit.inet.InetUtils;
|
||||
import cn.hippo4j.springboot.starter.config.BootstrapProperties;
|
||||
import cn.hippo4j.springboot.starter.remote.HttpAgent;
|
||||
import cn.hippo4j.springboot.starter.toolkit.CloudCommonIdUtil;
|
||||
import com.google.common.collect.Lists;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.ApplicationArguments;
|
||||
import org.springframework.boot.ApplicationRunner;
|
||||
import org.springframework.core.env.ConfigurableEnvironment;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static cn.hippo4j.common.constant.Constants.IDENTIFY_SLICER_SYMBOL;
|
||||
import static cn.hippo4j.common.constant.Constants.REGISTER_ADAPTER_PATH;
|
||||
|
||||
/**
|
||||
* Thread-pool adapter register.
|
||||
*/
|
||||
@Slf4j
|
||||
@AllArgsConstructor
|
||||
public class ThreadPoolAdapterRegister implements ApplicationRunner {
|
||||
|
||||
private final HttpAgent httpAgent;
|
||||
|
||||
private final BootstrapProperties properties;
|
||||
|
||||
private final ConfigurableEnvironment environment;
|
||||
|
||||
private final InetUtils hippo4JInetUtils;
|
||||
|
||||
@Override
|
||||
public void run(ApplicationArguments args) throws Exception {
|
||||
Map<String, ThreadPoolAdapter> threadPoolAdapterMap = ApplicationContextHolder.getBeansOfType(ThreadPoolAdapter.class);
|
||||
List<ThreadPoolAdapterCacheConfig> cacheConfigList = Lists.newArrayList();
|
||||
threadPoolAdapterMap.forEach((key, val) -> {
|
||||
ThreadPoolAdapterCacheConfig cacheConfig = new ThreadPoolAdapterCacheConfig();
|
||||
cacheConfig.setMark(val.mark());
|
||||
String tenantItemKey = properties.getNamespace() + IDENTIFY_SLICER_SYMBOL + properties.getItemId();
|
||||
cacheConfig.setTenantItemKey(tenantItemKey);
|
||||
String clientAddress = CloudCommonIdUtil.getDefaultInstanceId(environment, hippo4JInetUtils);
|
||||
cacheConfig.setClientAddress(clientAddress);
|
||||
cacheConfig.setThreadPoolAdapterStates(val.getThreadPoolStates());
|
||||
cacheConfigList.add(cacheConfig);
|
||||
});
|
||||
if (CollectionUtil.isNotEmpty(cacheConfigList)) {
|
||||
try {
|
||||
Result result = httpAgent.httpPost(REGISTER_ADAPTER_PATH, cacheConfigList);
|
||||
if (!result.isSuccess()) {
|
||||
log.warn("Failed to register third-party thread pool data.");
|
||||
}
|
||||
} catch (Throwable ex) {
|
||||
log.error("Failed to register third-party thread pool data.", ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue