diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/AskResponse.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/AskResponse.java index 474dd3a2..9d88e8a5 100644 --- a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/AskResponse.java +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/AskResponse.java @@ -17,4 +17,5 @@ import java.io.Serializable; @AllArgsConstructor public class AskResponse implements Serializable { private boolean success; + private Object extra; } diff --git a/oh-my-scheduler-server/pom.xml b/oh-my-scheduler-server/pom.xml index 8bcbb04a..e25770ac 100644 --- a/oh-my-scheduler-server/pom.xml +++ b/oh-my-scheduler-server/pom.xml @@ -21,6 +21,7 @@ 3.4.2 8.0.19 3.10 + 4.3.0 @@ -53,6 +54,14 @@ ${akka.version} + + + org.apache.curator + curator-recipes + ${curator.version} + + + org.apache.commons diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/common/config/CuratorConfig.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/common/config/CuratorConfig.java new file mode 100644 index 00000000..c1e09a4a --- /dev/null +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/common/config/CuratorConfig.java @@ -0,0 +1,36 @@ +package com.github.kfcfans.oms.server.common.config; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * ZooKeeper 连接配置 + * + * @author tjq + * @since 2020/4/4 + */ +@Configuration +public class CuratorConfig { + + @Value("${zookeeper.address}") + private String zkAddress; + + @Bean("omsCurator") + public CuratorFramework initCurator() { + CuratorFramework client = CuratorFrameworkFactory.builder() + .namespace("oms") + // zookeeper 地址，多值用 , 分割即可 + .connectString(zkAddress) + .sessionTimeoutMs(1000) + .connectionTimeoutMs(1000) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .build(); + client.start(); + return client; + } + +} diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/actors/OhMyServer.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/OhMyServer.java similarity index 72% rename from oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/actors/OhMyServer.java rename to oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/OhMyServer.java index ed1bc2aa..5ab545c3 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/actors/OhMyServer.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/OhMyServer.java @@ -1,5 +1,7 @@ -package com.github.kfcfans.oms.server.core.actors; +package com.github.kfcfans.oms.server.core.akka; +import akka.actor.ActorPath; +import akka.actor.ActorSelection; import akka.actor.ActorSystem; import akka.actor.Props; import com.github.kfcfans.common.RemoteConstant; @@ -7,6 +9,7 @@ import com.github.kfcfans.common.utils.NetUtils; import com.google.common.collect.Maps; import com.typesafe.config.Config; import com.typesafe.config.ConfigFactory; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; import java.util.Map; @@ -21,6 +24,7 @@ import java.util.Map; public class OhMyServer { public static ActorSystem actorSystem; + @Getter private static String actorSystemAddress; public void init() { @@ -41,4 +45,14 @@ public class OhMyServer { actorSystem.actorOf(Props.create(ServerActor.class), RemoteConstant.SERVER_ACTOR_NAME); } + + /** + * 获取 ServerActor 的 ActorSelection + * @param address IP:port + * @return ActorSelection + */ + public static ActorSelection getServerActor(String address) { + String path = String.format("akka://%s@%s/user/%s", RemoteConstant.SERVER_ACTOR_SYSTEM_NAME, address, RemoteConstant.SERVER_ACTOR_NAME); + return actorSystem.actorSelection(path); + } } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/Ping.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/Ping.java new file mode 100644 index 00000000..ed069f40 --- /dev/null +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/Ping.java @@ -0,0 +1,16 @@ +package com.github.kfcfans.oms.server.core.akka; + +import lombok.Data; + +import java.io.Serializable; + +/** + * 检测目标机器是否存活 + * + * @author tjq + * @since 2020/4/5 + */ +@Data +public class Ping implements Serializable { + private long currentTime; +} diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/actors/ServerActor.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/ServerActor.java similarity index 54% rename from oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/actors/ServerActor.java rename to oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/ServerActor.java index 0ba32790..14528dce 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/actors/ServerActor.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/core/akka/ServerActor.java @@ -1,7 +1,8 @@ -package com.github.kfcfans.oms.server.core.actors; +package com.github.kfcfans.oms.server.core.akka; import akka.actor.AbstractActor; import com.github.kfcfans.common.request.WorkerHeartbeat; +import com.github.kfcfans.common.response.AskResponse; import lombok.extern.slf4j.Slf4j; /** @@ -17,13 +18,22 @@ public class ServerActor extends AbstractActor { public Receive createReceive() { return receiveBuilder() .match(WorkerHeartbeat.class, this::onReceiveWorkerHeartbeat) + .match(Ping.class, this::onReceivePing) .matchAny(obj -> log.warn("[ServerActor] receive unknown request: {}.", obj)) .build(); } + /** + * 处理存活检测的请求 + * @param ping 存活检测请求 + */ + private void onReceivePing(Ping ping) { + AskResponse askResponse = new AskResponse(); + askResponse.setSuccess(true); + askResponse.setExtra(System.currentTimeMillis() - ping.getCurrentTime()); + getSender().tell(askResponse, getSelf()); + } + private void onReceiveWorkerHeartbeat(WorkerHeartbeat heartbeat) { - - - } } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/AppInfoDO.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/AppInfoDO.java index 7f23362d..80c71b16 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/AppInfoDO.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/AppInfoDO.java @@ -23,6 +23,8 @@ public class AppInfoDO { private String appName; private String description; + private String currentServer; + private Date gmtCreate; private Date gmtModified; } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/OmsLockDO.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/OmsLockDO.java index 54ee4e4c..bfb71f2f 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/OmsLockDO.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/OmsLockDO.java @@ -23,14 +23,14 @@ public class OmsLockDO { private Long id; private String lockName; - private String owner; + private String ownerIP; private Date gmtCreate; private Date gmtModified; - public OmsLockDO(String lockName, String owner) { + public OmsLockDO(String lockName, String ownerIP) { this.lockName = lockName; - this.owner = owner; + this.ownerIP = ownerIP; this.gmtCreate = new Date(); this.gmtModified = this.gmtCreate; } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/AppInfoRepository.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/AppInfoRepository.java index 55e43696..ac4bc0ae 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/AppInfoRepository.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/AppInfoRepository.java @@ -10,4 +10,6 @@ import org.springframework.data.jpa.repository.JpaRepository; * @since 2020/4/1 */ public interface AppInfoRepository extends JpaRepository { + + AppInfoDO findByAppName(String appName); } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/OmsLockRepository.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/OmsLockRepository.java index 36b38262..d2c9a49b 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/OmsLockRepository.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/OmsLockRepository.java @@ -19,4 +19,6 @@ public interface OmsLockRepository extends JpaRepository { @Transactional @Query(value = "delete from oms_lock where lock_name = ?1", nativeQuery = true) int deleteByLockName(String lockName); + + OmsLockDO findByLockName(String lockName); } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/ha/ServerSelectService.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/ha/ServerSelectService.java new file mode 100644 index 00000000..b3c7c634 --- /dev/null +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/ha/ServerSelectService.java @@ -0,0 +1,111 @@ +package com.github.kfcfans.oms.server.service.ha; + +import akka.actor.ActorSelection; +import akka.pattern.Patterns; +import com.github.kfcfans.common.response.AskResponse; +import com.github.kfcfans.oms.server.core.akka.OhMyServer; +import com.github.kfcfans.oms.server.core.akka.Ping; +import com.github.kfcfans.oms.server.persistence.model.AppInfoDO; +import com.github.kfcfans.oms.server.persistence.repository.AppInfoRepository; +import com.github.kfcfans.oms.server.service.lock.LockService; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.time.Duration; +import java.util.Date; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.TimeUnit; + +/** + * Worker请求分配Server服务 + * + * @author tjq + * @since 2020/4/5 + */ +@Slf4j +@Service +public class ServerSelectService { + + @Resource + private LockService lockService; + @Resource + private AppInfoRepository appInfoRepository; + + private static final int RETRY_TIMES = 10; + private static final long PING_TIMEOUT_MS = 5000; + private static final long WAIT_LOCK_TIME = 1000; + private static final String SERVER_ELECT_LOCK = "server_elect_%s"; + + /** + * 获取某个应用对应的Server + * 缺点：如果server死而复生，可能造成worker集群脑裂，不过感觉影响不是很大 & 概率极低，就不管了 + * @param appName 应用名称 + * @return 当前可用的Server + */ + public String getServer(String appName) { + + for (int i = 0; i < RETRY_TIMES; i++) { + + // 无锁获取当前数据库中的Server + AppInfoDO app = appInfoRepository.findByAppName(appName); + if (app == null) { + throw new RuntimeException(appName + " is not registered!"); + } + String originServer = app.getCurrentServer(); + if (isActive(originServer)) { + return originServer; + } + + // 获取失败，重新进行Server选举，需要加锁 + String lockName = String.format(SERVER_ELECT_LOCK, appName); + boolean lockStatus = lockService.lock(lockName); + if (!lockStatus) { + try { + Thread.sleep(1000); + }catch (Exception ignore) { + } + continue; + } + try { + + // 可能上一台机器已经完成了Server选举，需要再次判断 + AppInfoDO appInfo = appInfoRepository.findByAppName(appName); + if (isActive(appInfo.getCurrentServer())) { + return appInfo.getCurrentServer(); + } + + // 篡位，本机作为Server + appInfo.setCurrentServer(OhMyServer.getActorSystemAddress()); + appInfo.setGmtModified(new Date()); + + appInfoRepository.saveAndFlush(appInfo); + return appInfo.getCurrentServer(); + }catch (Exception e) { + log.warn("[ServerSelectService] write new server to db failed for app {}.", appName); + }finally { + lockService.unlock(lockName); + } + } + throw new RuntimeException("server elect failed for app " + appName); + } + + private boolean isActive(String serverAddress) { + if (StringUtils.isEmpty(serverAddress)) { + return false; + } + Ping ping = new Ping(); + ping.setCurrentTime(System.currentTimeMillis()); + + ActorSelection serverActor = OhMyServer.getServerActor(serverAddress); + try { + CompletionStage askCS = Patterns.ask(serverActor, ping, Duration.ofMillis(PING_TIMEOUT_MS)); + AskResponse response = (AskResponse) askCS.toCompletableFuture().get(PING_TIMEOUT_MS, TimeUnit.MILLISECONDS); + return response.isSuccess(); + }catch (Exception e) { + log.warn("[ServerSelectService] server({}) was down, try to elect a new server.", serverAddress); + } + return false; + } +} diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/impl/DatabaseLockService.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/impl/DatabaseLockService.java deleted file mode 100644 index 5c54fa05..00000000 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/impl/DatabaseLockService.java +++ /dev/null @@ -1,52 +0,0 @@ -package com.github.kfcfans.oms.server.service.impl; - -import com.github.kfcfans.common.utils.CommonUtils; -import com.github.kfcfans.common.utils.NetUtils; -import com.github.kfcfans.oms.server.persistence.model.OmsLockDO; -import com.github.kfcfans.oms.server.persistence.repository.OmsLockRepository; -import com.github.kfcfans.oms.server.service.LockService; -import lombok.extern.slf4j.Slf4j; -import org.springframework.dao.DataIntegrityViolationException; -import org.springframework.stereotype.Service; - -import javax.annotation.Resource; - -/** - * 基于数据库实现分布式锁 - * - * @author tjq - * @since 2020/4/2 - */ -@Slf4j -@Service -public class DatabaseLockService implements LockService { - - @Resource - private OmsLockRepository omsLockRepository; - - @Override - public boolean lock(String name) { - - try { - - OmsLockDO lock = new OmsLockDO(name, NetUtils.getLocalHost()); - omsLockRepository.saveAndFlush(lock); - - return true; - }catch (DataIntegrityViolationException ignore) { - log.info("[DatabaseLockService] other thread get the lock {}.", name); - } catch (Exception e) { - log.error("[DatabaseLockService] lock {} failed.", name, e); - } - return false; - } - - @Override - public void unlock(String name) { - try { - CommonUtils.executeWithRetry0(() -> omsLockRepository.deleteByLockName(name)); - }catch (Exception e) { - log.error("[DatabaseLockService] unlock {} failed.", name, e); - } - } -} diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/lock/DatabaseLockService.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/lock/DatabaseLockService.java new file mode 100644 index 00000000..2049864d --- /dev/null +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/lock/DatabaseLockService.java @@ -0,0 +1,73 @@ +package com.github.kfcfans.oms.server.service.lock; + +import com.github.kfcfans.common.utils.CommonUtils; +import com.github.kfcfans.common.utils.NetUtils; +import com.github.kfcfans.oms.server.persistence.model.OmsLockDO; +import com.github.kfcfans.oms.server.persistence.repository.OmsLockRepository; +import com.google.common.collect.Maps; +import lombok.extern.slf4j.Slf4j; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * 基于数据库实现的分布式锁 + * + * @author tjq + * @since 2020/4/5 + */ +@Slf4j +@Service +public class DatabaseLockService implements LockService { + + @Resource + private OmsLockRepository omsLockRepository; + + private Map lockName2FailedTimes = Maps.newConcurrentMap(); + private static final int MAX_FAILED_NUM = 5; + // 最长持有锁30秒 + private static final long LOCK_TIMEOUT_MS = 30000; + + @Override + public boolean lock(String name) { + + AtomicInteger failedCount = lockName2FailedTimes.computeIfAbsent(name, ignore -> new AtomicInteger(0)); + OmsLockDO newLock = new OmsLockDO(name, NetUtils.getLocalHost()); + try { + omsLockRepository.saveAndFlush(newLock); + failedCount.set(0); + return true; + }catch (DataIntegrityViolationException ignore) { + }catch (Exception e) { + log.warn("[DatabaseLockService] write lock to database failed, lockName = {}.", name, e); + } + + // 连续失败一段时间，需要判断是否为锁释放失败的情况 + if (failedCount.incrementAndGet() > MAX_FAILED_NUM) { + + OmsLockDO omsLockDO = omsLockRepository.findByLockName(name); + long lockedMillions = System.currentTimeMillis() - omsLockDO.getGmtCreate().getTime(); + if (lockedMillions > LOCK_TIMEOUT_MS) { + + log.warn("[DatabaseLockService] The lock({}) already timeout, will be deleted now.", omsLockDO); + unlock(name); + } else { + failedCount.set(0); + } + } + return false; + } + + @Override + public void unlock(String name) { + + try { + CommonUtils.executeWithRetry0(() -> omsLockRepository.deleteByLockName(name)); + }catch (Exception e) { + log.error("[DatabaseLockService] unlock {} failed.", name, e); + } + } +} diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/LockService.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/lock/LockService.java similarity index 77% rename from oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/LockService.java rename to oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/lock/LockService.java index e72bfddd..146d6a68 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/LockService.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/lock/LockService.java @@ -1,7 +1,7 @@ -package com.github.kfcfans.oms.server.service; +package com.github.kfcfans.oms.server.service.lock; /** - * 锁服务 + * 锁服务，所有方法都不允许抛出任何异常！ * * @author tjq * @since 2020/4/2 diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/ControllerExceptionHandler.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/ControllerExceptionHandler.java index beb0bc17..f0114a1a 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/ControllerExceptionHandler.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/ControllerExceptionHandler.java @@ -19,6 +19,6 @@ public class ControllerExceptionHandler { @ExceptionHandler(Exception.class) public ResultDTO exceptionHandler(Exception e) { log.error("[ControllerException] http request failed.", e); - return ResultDTO.failed(e); + return ResultDTO.failed(e.getMessage()); } } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/controller/ServerController.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/controller/ServerController.java new file mode 100644 index 00000000..bf333e4d --- /dev/null +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/controller/ServerController.java @@ -0,0 +1,30 @@ +package com.github.kfcfans.oms.server.web.controller; + +import com.github.kfcfans.oms.server.service.ha.ServerSelectService; +import com.github.kfcfans.oms.server.web.ResultDTO; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import javax.annotation.Resource; + +/** + * 处理内部请求的 Controller + * + * @author tjq + * @since 2020/4/4 + */ +@RestController +@RequestMapping("/server") +public class ServerController { + + @Resource + private ServerSelectService serverSelectService; + + @GetMapping("/acquire") + public ResultDTO acquireServer(String appName) { + String server = serverSelectService.getServer(appName); + return ResultDTO.success(server); + } + +} diff --git a/oh-my-scheduler-server/src/main/resources/application.properties b/oh-my-scheduler-server/src/main/resources/application.properties index 09396ca1..f64e807a 100644 --- a/oh-my-scheduler-server/src/main/resources/application.properties +++ b/oh-my-scheduler-server/src/main/resources/application.properties @@ -11,4 +11,7 @@ spring.datasource.hikari.maximum-pool-size=20 spring.datasource.hikari.minimum-idle=5 # JPA 相关配置 spring.jpa.show-sql=true -spring.jpa.hibernate.ddl-auto=update \ No newline at end of file +spring.jpa.hibernate.ddl-auto=update + +# ZooKeeper(多值逗号分割) +zookeeper.address=115.159.215.229:2181 \ No newline at end of file diff --git a/oh-my-scheduler-server/src/test/java/com/github/kfcfans/oms/server/test/ServiceTest.java b/oh-my-scheduler-server/src/test/java/com/github/kfcfans/oms/server/test/ServiceTest.java index 7a39b66a..6e1201e0 100644 --- a/oh-my-scheduler-server/src/test/java/com/github/kfcfans/oms/server/test/ServiceTest.java +++ b/oh-my-scheduler-server/src/test/java/com/github/kfcfans/oms/server/test/ServiceTest.java @@ -1,6 +1,6 @@ package com.github.kfcfans.oms.server.test; -import com.github.kfcfans.oms.server.service.LockService; +import com.github.kfcfans.oms.server.service.lock.LockService; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.boot.test.context.SpringBootTest;