Merge branch 'develop' into h2

# Conflicts:
#	hippo4j-server/src/main/resources/application.properties
pull/665/head
weihu 3 years ago
commit 4d1ccc7382

@ -56,6 +56,8 @@ Hippo-4J 获得了一些宝贵的荣誉,这属于每一位对 Hippo-4J 做出
## 友情链接
- [[ LiteFlow ]](https://liteflow.yomahub.com/):轻量,快速,稳定可编排的组件式规则引擎。
- [[ Sa-Token ]](https://github.com/dromara/sa-token):一个轻量级 java 权限认证框架,让鉴权变得简单、优雅!
- [[ HertzBeat ]](https://github.com/dromara/hertzbeat):易用友好的云监控系统, 无需 Agent, 强大自定义监控能力。

@ -2,14 +2,11 @@
sidebar_position: 3
---
import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';
# 快速开始
## 服务启动
MySQL 创建名为 `hippo4j_manager`数据库,字符集选择 `utf8mb4`,并导入 [Hippo4J 初始化 SQL 语句](https://github.com/longtai-cn/hippo4j/blob/develop/hippo4j-server/conf/hippo4j_manager.sql)。
MySQL 数据库导入 [Hippo4J 初始化 SQL 语句](https://github.com/longtai-cn/hippo4j/blob/develop/hippo4j-server/conf/hippo4j_manager.sql)。
使用 Docker 运行服务端,可以灵活定制相关参数。如果 MySQL 非 Docker 部署,`MYSQL_HOST` 需要使用本地 IP。

@ -0,0 +1,101 @@
/*
* 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.common.model;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
/**
* web ip and port info
*/
@Data
@Slf4j
public class WebIpAndPortInfo {
protected static final String ALL = "*";
protected static final String SPOT = "\\.";
protected static final String COLON = ":";
private String ip;
private String port;
private String[] ipSegment;
public WebIpAndPortInfo(String ip, String port) {
this.ip = ip;
this.port = port;
this.ipSegment = ip.split(SPOT);
}
public static WebIpAndPortInfo build(String node) {
if (ALL.equals(node)) {
return new WebIpAndPortInfo(ALL, ALL);
}
String[] ipPort = node.split(COLON);
if (ipPort.length != 2) {
log.error("The IP address format is error : {}", node);
return null;
}
return new WebIpAndPortInfo(ipPort[0], ipPort[1]);
}
/**
* check
*
* @param appIpSegment application ip segment
* @param port application port
*/
public boolean check(String[] appIpSegment, String port) {
return checkPort(port) && checkIp(appIpSegment);
}
/**
* check ip
*
* @param appIpSegment application ip segment
*/
protected boolean checkIp(String[] appIpSegment) {
if (ALL.equals(this.ip)) {
return true;
}
boolean flag = true;
for (int i = 0; i < ipSegment.length && flag; i++) {
String propIp = ipSegment[i];
String appIp = appIpSegment[i];
flag = contrastSegment(appIp, propIp);
}
return flag;
}
/**
* check port
*
* @param port application port
*/
protected boolean checkPort(String port) {
return contrastSegment(port, this.port);
}
/**
* Check whether the strings are the same
*
* @param appIp appIp
* @param propIp propIp
*/
protected boolean contrastSegment(String appIp, String propIp) {
return ALL.equals(propIp) || appIp.equals(propIp);
}
}

@ -23,11 +23,8 @@ public class FileUtilTest {
@Test
public void assertReadUtf8String() {
String testText = "abcd简体繁体\uD83D\uDE04\uD83D\uDD25& *\n" +
"second line\n" +
"empty line next\n";
String testFilePath = "test/test_utf8.txt";
String contentByFileUtil = FileUtil.readUtf8String(testFilePath);
Assert.isTrue(testText.equals(contentByFileUtil));
Assert.notEmpty(contentByFileUtil);
}
}

@ -17,5 +17,26 @@
package cn.hippo4j.common.toolkit;
import org.junit.Test;
import static org.junit.Assert.assertTrue;
public class ThreadUtilTest {
@Test
public void testNewThread() {
// Setup
final Runnable runnable = null;
// Run the test
final Thread result = ThreadUtil.newThread(runnable, "name", false);
// Verify the results
Assert.notNull(result);
}
@Test
public void testSleep() {
assertTrue(ThreadUtil.sleep(0L));
}
}

@ -17,14 +17,14 @@
package cn.hippo4j.config.springboot.starter.refresher.event;
import cn.hippo4j.adapter.web.WebThreadPoolHandlerChoose;
import cn.hippo4j.adapter.web.WebThreadPoolService;
import cn.hippo4j.common.config.ApplicationContextHolder;
import cn.hippo4j.common.model.WebIpAndPortInfo;
import cn.hippo4j.common.toolkit.Assert;
import cn.hippo4j.common.toolkit.StringUtil;
import cn.hippo4j.core.toolkit.inet.InetUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.web.context.WebServerInitializedEvent;
import org.springframework.context.event.EventListener;
import java.util.Arrays;
import java.util.Objects;
@ -36,33 +36,33 @@ import java.util.Objects;
public abstract class AbstractRefreshListener<M> implements RefreshListener<Hippo4jConfigDynamicRefreshEvent, M> {
protected static final String ALL = "*";
protected static final String SPOT = "\\.";
protected static final String SEPARATOR = ",";
protected static final String COLON = ":";
/**
* application ip
*/
protected final String[] ipSegment;
/**
* application post
* application ip and application post
*/
protected String port;
protected static volatile WebIpAndPortInfo webIpAndPort;
protected void initIpAndPort() {
if (webIpAndPort == null) {
synchronized (AbstractRefreshListener.class) {
if (webIpAndPort == null) {
webIpAndPort = getWebIpAndPortInfo();
}
}
}
}
AbstractRefreshListener() {
private WebIpAndPortInfo getWebIpAndPortInfo() {
InetUtils inetUtils = ApplicationContextHolder.getBean(InetUtils.class);
InetUtils.HostInfo loopBackHostInfo = inetUtils.findFirstNonLoopBackHostInfo();
Assert.notNull(loopBackHostInfo, "Unable to get the application IP address");
ipSegment = loopBackHostInfo.getIpAddress().split(SPOT);
}
@EventListener(WebServerInitializedEvent.class)
public void webServerInitializedListener(WebServerInitializedEvent event) {
port = String.valueOf(event.getWebServer().getPort());
String ip = loopBackHostInfo.getIpAddress();
WebThreadPoolHandlerChoose webThreadPoolHandlerChoose = ApplicationContextHolder.getBean(WebThreadPoolHandlerChoose.class);
WebThreadPoolService webThreadPoolService = webThreadPoolHandlerChoose.choose();
// when get the port at startup, can get the message: "port xxx was already in use" or use two ports
String port = String.valueOf(webThreadPoolService.getWebServer().getPort());
return new WebIpAndPortInfo(ip, port);
}
/**
@ -81,6 +81,9 @@ public abstract class AbstractRefreshListener<M> implements RefreshListener<Hipp
*/
@Override
public boolean match(M properties) {
if (webIpAndPort == null) {
initIpAndPort();
}
String nodes = getNodes(properties);
if (StringUtil.isEmpty(nodes) || ALL.equals(nodes)) {
return true;
@ -88,9 +91,9 @@ public abstract class AbstractRefreshListener<M> implements RefreshListener<Hipp
String[] splitNodes = nodes.split(SEPARATOR);
return Arrays.stream(splitNodes)
.distinct()
.map(IpAndPort::build)
.map(WebIpAndPortInfo::build)
.filter(Objects::nonNull)
.anyMatch(i -> i.check(ipSegment, port));
.anyMatch(i -> i.check(webIpAndPort.getIpSegment(), webIpAndPort.getPort()));
}
/**
@ -103,79 +106,4 @@ public abstract class AbstractRefreshListener<M> implements RefreshListener<Hipp
return ALL;
}
/**
* ip + port
*/
@Data
protected static class IpAndPort {
private String ip;
private String port;
private String[] propIpSegment;
private IpAndPort(String ip, String port) {
this.ip = ip;
this.port = port;
this.propIpSegment = ip.split(SPOT);
}
public static IpAndPort build(String node) {
if (ALL.equals(node)) {
return new IpAndPort(ALL, ALL);
}
String[] ipPort = node.split(COLON);
if (ipPort.length != 2) {
log.error("The IP address format is error : {}", node);
return null;
}
return new IpAndPort(ipPort[0], ipPort[1]);
}
/**
* check
*
* @param appIpSegment application ip segment
* @param port application port
*/
public boolean check(String[] appIpSegment, String port) {
return checkPort(port) && checkIp(appIpSegment);
}
/**
* check ip
*
* @param appIpSegment application ip segment
*/
protected boolean checkIp(String[] appIpSegment) {
if (ALL.equals(this.ip)) {
return true;
}
boolean flag = true;
for (int i = 0; i < propIpSegment.length && flag; i++) {
String propIp = propIpSegment[i];
String appIp = appIpSegment[i];
flag = contrastSegment(appIp, propIp);
}
return flag;
}
/**
* check port
*
* @param port application port
*/
protected boolean checkPort(String port) {
return contrastSegment(port, this.port);
}
/**
* Check whether the strings are the same
*
* @param appIp appIp
* @param propIp propIp
*/
protected boolean contrastSegment(String appIp, String propIp) {
return ALL.equals(propIp) || appIp.equals(propIp);
}
}
}

@ -60,7 +60,7 @@ public class AdapterExecutorsRefreshListener extends AbstractRefreshListener<Ada
continue;
}
if (!Objects.equals(adapterExecutorProperties.getCorePoolSize(), each.getCorePoolSize())
|| !Objects.equals(adapterExecutorProperties.getMaximumPoolSize(), each.getMaximumPoolSize())) {
|| !Objects.equals(adapterExecutorProperties.getMaximumPoolSize(), each.getMaximumPoolSize())) {
threadPoolAdapterMap.forEach((key, val) -> {
if (Objects.equals(val.mark(), each.getMark())) {
val.updateThreadPool(BeanUtil.toBean(each, ThreadPoolAdapterParameter.class));

@ -22,19 +22,39 @@ import cn.hippo4j.common.toolkit.ContentUtil;
import cn.hippo4j.common.toolkit.GroupKey;
import cn.hippo4j.common.toolkit.JSONUtil;
import cn.hippo4j.common.web.base.Result;
import cn.hippo4j.core.executor.support.ThreadFactoryBuilder;
import cn.hippo4j.springboot.starter.remote.HttpAgent;
import cn.hippo4j.springboot.starter.remote.ServerHealthCheck;
import cn.hippo4j.core.executor.support.ThreadFactoryBuilder;
import cn.hutool.core.util.IdUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
import java.net.URLDecoder;
import java.util.*;
import java.util.concurrent.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import static cn.hippo4j.common.constant.Constants.*;
import static cn.hippo4j.common.constant.Constants.CONFIG_CONTROLLER_PATH;
import static cn.hippo4j.common.constant.Constants.CONFIG_LONG_POLL_TIMEOUT;
import static cn.hippo4j.common.constant.Constants.GROUP_KEY_DELIMITER_TRANSLATION;
import static cn.hippo4j.common.constant.Constants.LINE_SEPARATOR;
import static cn.hippo4j.common.constant.Constants.LISTENER_PATH;
import static cn.hippo4j.common.constant.Constants.LONG_PULLING_CLIENT_IDENTIFICATION;
import static cn.hippo4j.common.constant.Constants.LONG_PULLING_TIMEOUT;
import static cn.hippo4j.common.constant.Constants.LONG_PULLING_TIMEOUT_NO_HANGUP;
import static cn.hippo4j.common.constant.Constants.NULL;
import static cn.hippo4j.common.constant.Constants.PROBE_MODIFY_REQUEST;
import static cn.hippo4j.common.constant.Constants.WEIGHT_CONFIGS;
import static cn.hippo4j.common.constant.Constants.WORD_SEPARATOR;
/**
* Client worker.
@ -44,8 +64,6 @@ public class ClientWorker {
private long timeout;
private double currentLongingTaskCount = 0;
private final HttpAgent agent;
private final String identify;
@ -75,26 +93,14 @@ public class ClientWorker {
this.executorService = Executors.newSingleThreadScheduledExecutor(
ThreadFactoryBuilder.builder().prefix("client.long.polling.executor").daemon(true).build());
log.info("Client identify: {}", identify);
this.executor.scheduleWithFixedDelay(() -> {
this.executor.schedule(() -> {
try {
awaitApplicationComplete.await();
checkConfigInfo();
executorService.execute(new LongPollingRunnable());
} catch (Throwable ex) {
log.error("Sub check rotate check error.", ex);
}
}, 1L, 1024L, TimeUnit.MILLISECONDS);
}
public void checkConfigInfo() {
int listenerSize = cacheMap.size();
double perTaskConfigSize = 3000D;
int longingTaskCount = (int) Math.ceil(listenerSize / perTaskConfigSize);
if (longingTaskCount > currentLongingTaskCount) {
for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
executorService.execute(new LongPollingRunnable());
}
currentLongingTaskCount = longingTaskCount;
}
}, 1L, TimeUnit.MILLISECONDS);
}
class LongPollingRunnable implements Runnable {

Loading…
Cancel
Save