add 规则引擎集群

jetlinks-components/rule-engine-component/pom.xml

@@ -18,6 +18,23 @@
             <artifactId>cron-utils</artifactId>
             <version>9.0.2</version>
         </dependency>
+
+
+        <!-- 另一种Spring集成starter，本章未使用 -->
+        <dependency>
+            <groupId>org.redisson</groupId>
+            <artifactId>redisson-spring-boot-starter</artifactId>
+            <version>3.13.6</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-web</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+
+
         <dependency>
             <groupId>org.jetlinks</groupId>
             <artifactId>rule-engine-support</artifactId>

jetlinks-components/rule-engine-component/src/main/java/org/jetlinks/community/rule/engine/cluster/ClusterUniqueTask.java

@@ -0,0 +1,359 @@
+package org.jetlinks.community.rule.engine.cluster;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.core.cluster.ClusterManager;
+import org.jetlinks.rule.engine.api.RuleData;
+import org.jetlinks.rule.engine.api.scheduler.ScheduleJob;
+import org.jetlinks.rule.engine.api.task.ExecutableTaskExecutor;
+import org.jetlinks.rule.engine.api.task.Task;
+import org.jetlinks.rule.engine.api.task.TaskExecutor;
+import org.jetlinks.rule.engine.defaults.AbstractExecutionContext;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.*;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName ClusterTask.java
+ * @Description 集群唯一任务
+ *  1、监听自身的任务情况
+ *  2、改变状态后向集群广播
+ *
+ *
+ *  管理自身状态（除开启和停止以外的状态）
+ *  任务自行进行管理，由work进行监听
+ * @createTime 2021年11月12日 10:57:00
+ */
+@Data
+@AllArgsConstructor
+@Slf4j
+public class ClusterUniqueTask implements Task ,Serializable{
+
+    private String workerId;
+
+    /**
+     * 当前服务器id
+     */
+    private final String currentSeverId;
+
+    private final String schedulerId;
+
+    /**
+     * 最后更新时间
+     */
+    private long lastStateTime;
+
+    private transient final RedissonClient redissonClient;
+
+    /**
+     * 开始时间
+     */
+    private long startTime;
+
+    private transient final AbstractExecutionContext context;
+
+    //默认心跳时间10s
+    private int pingTime=10;
+
+    /**
+     * 心跳，心跳包传递任务状态，集群实时更新任务状态
+     */
+    private transient String pingTopic ="cluster-unique-task-ping-%s";
+
+
+    /**
+     * 操作主题
+     */
+    private transient String operationTopic ="cluster-unique-task-operation-%s";
+
+    private Task.State taskState;
+
+
+
+    private transient RLock lock;
+    /**
+     * 该任务已死亡
+     */
+    private boolean isDead=true;
+
+    private transient final ClusterManager clusterManager;
+
+    private transient final TaskExecutor executor;
+
+    @Override
+    public String getId(){
+        return this.getContext().getInstanceId();
+    }
+
+    @Override
+    public String getName() {
+        return executor.getName();
+    }
+
+    @Override
+    public ScheduleJob getJob() {
+        return context.getJob();
+    }
+
+    @Override
+    public Mono<Void> setJob(ScheduleJob job) {
+        if(isReplica()){
+            return null;
+        }
+        return Mono.fromRunnable(() -> {
+            ScheduleJob old = context.getJob();
+            context.setJob(job);
+            try {
+                executor.validate();
+            } catch (Throwable e) {
+                context.setJob(old);
+                throw e;
+            }
+        });
+    }
+
+    public ClusterUniqueTask(String schedulerId,
+                             AbstractExecutionContext context,
+                             TaskExecutor executor,
+                             String currentSeverId,
+                             ClusterManager clusterManager,
+                             RedissonClient redissonClient) {
+        this.schedulerId = schedulerId;
+        this.context = context;
+        this.executor = executor;
+        this.currentSeverId=currentSeverId;
+        this.workerId=currentSeverId;
+        this.clusterManager=clusterManager;
+        this.pingTopic=String.format(this.pingTopic,context.getInstanceId());
+        this.operationTopic=String.format(this.operationTopic,context.getInstanceId());
+        this.lock = redissonClient.getLock("cluster-unique-"+this.getId());
+        this.redissonClient=redissonClient;
+        //先创建本地任务，再争夺任务的唯一锁，避免消息漏发
+        initUniqueTask().subscribe();
+    }
+
+    private Mono<Void> initUniqueTask(){
+        return creatUniqueTask()
+            .filter(result->!result)
+            .flatMap(ignore->this.creatUniqueFail())
+            .then();
+
+    }
+    private Mono<Boolean> creatUniqueTask() {
+        if(!isDead){
+            return Mono.just(false);
+        }
+        boolean result =false;
+        try {
+            //获取锁
+            result = lock.tryLock(-1,pingTime, TimeUnit.SECONDS);
+        }catch (InterruptedException e){
+            return Mono.just(false);
+        }
+        //争夺成功
+        if(result){
+            //创建任务,锁争夺成功后发送心跳
+            isDead=false;
+            return Flux.interval(Duration.ofSeconds(pingTime/2))
+                .flatMap(ignore->{
+                    lock.lock(pingTime,TimeUnit.SECONDS);
+                    return clusterManager.getTopic(pingTopic).publish(Mono.just(this));
+                })
+                .then(Mono.just(true));
+        }
+        //争夺失败
+        return Mono.just(false);
+    }
+
+
+    @Override
+    public Mono<Void> reload() {
+        return operation(OperationMessage.of(TaskOperation.RELOAD));
+    }
+
+    @Override
+    public Mono<Void> start() {
+        return operation(OperationMessage.of(TaskOperation.START));
+    }
+
+    @Override
+    public Mono<Void> pause() {
+        return operation(OperationMessage.of(TaskOperation.PAUSE));
+    }
+
+    @Override
+    public Mono<Void> shutdown() {
+        return operation(OperationMessage.of(TaskOperation.SHUTDOWN));
+    }
+
+    @Override
+    public Mono<Void> execute(RuleData data) {
+        return operation(OperationMessage.of(TaskOperation.EXECUTE,Arrays.asList(data)));
+    }
+
+    @Override
+    public Mono<State> getState() {
+        if(isReplica()){
+            //副本
+            return Mono.just(this.getTaskState());
+        }
+        return Mono.just(executor.getState());
+    }
+
+    @Override
+    public Mono<Void> debug(boolean debug) {
+        return operation(debug ?OperationMessage.of(TaskOperation.ENABLE_DEBUG)  :OperationMessage.of(TaskOperation.DISABLE_DEBUG));
+    }
+
+    @Override
+    public Mono<Long> getLastStateTime() {
+        return Mono.just(lastStateTime);
+    }
+
+    @Override
+    public Mono<Long> getStartTime() {
+        return Mono.just(startTime);
+    }
+
+    private Mono<Void> operation(OperationMessage message) {
+        TaskOperation operation=message.operation;
+        if(isReplica()){
+            //当前为任务副本，传递给任务执行者
+            return  clusterManager.getTopic(operationTopic)
+                .publish(Mono.just(message))
+                .then();
+        }
+        long currentTimeMillis = System.currentTimeMillis();
+        lastStateTime=currentTimeMillis;
+        //该任务不是其他任务的副本，即运行实例位于该机器中,改变状态后进行广播操作
+        switch (operation){
+            case START:
+                return Mono.<Void>fromRunnable(executor::start)
+                    .doOnSuccess((v) -> startTime =currentTimeMillis)
+                    .subscribeOn(Schedulers.boundedElastic());
+            case PAUSE:
+                return Mono.fromRunnable(executor::pause);
+            case RELOAD:
+                return Mono.<Void>fromRunnable(executor::reload)
+                    .subscribeOn(Schedulers.boundedElastic());
+            case SHUTDOWN:
+                this.taskState=State.shutdown;
+                //解锁
+                this.lock.unlock();
+                break;
+            case ENABLE_DEBUG:
+                return Mono.fromRunnable(() -> context.setDebug(true));
+            case DISABLE_DEBUG:
+                return Mono.fromRunnable(() -> context.setDebug(false));
+            case EXECUTE:
+                if(executor instanceof ExecutableTaskExecutor){
+                    RuleData data = (RuleData) message.getParams().get(0);
+                    return ((ExecutableTaskExecutor) executor).execute(data);
+                }
+                return Mono.empty();
+
+            default:break;
+        }
+        return Mono.empty();
+    }
+
+//    /**
+//     * 状态发生改变
+//     * @return
+//     */
+//    public Flux<?> handleStateChange(){
+//        //非本机任务状态发生改变
+//        return changeState.map(Function.identity()).filter(ignore->this.isReplica());
+//    }
+
+
+    public boolean isReplica(){
+        return !this.getCurrentSeverId().equals(this.getWorkerId());
+    }
+
+
+    private Mono<?> creatUniqueFail(){
+        return EmitterProcessor.create(true)
+            .flatMap(result -> clusterManager.getTopic(pingTopic)
+                .subscribePattern()
+                .mergeWith(clusterManager.getTopic(operationTopic).subscribePattern())
+                .flatMap(obj -> {
+                    isDead=false;
+                    Object message = obj.getMessage();
+                    if (message instanceof ClusterUniqueTask) {
+                        ClusterUniqueTask task = (ClusterUniqueTask) message;
+                        //心跳信息
+                        return Mono.just(task);
+                    } else if (message instanceof TaskOperation) {
+                        //操作信息
+                        TaskOperation task = (TaskOperation) message;
+                        return Mono.just(task);
+                    }
+                    return Mono.empty();
+                }))
+            //心跳超时
+            .timeout(Duration.ofSeconds(pingTime),
+                Mono.empty()
+                    .doOnNext(ignore->isDead=true)
+                    .concatWith(initUniqueTask())
+                    .then(Mono.just(this)))
+            .flatMap(msg -> {
+                if (msg instanceof ClusterUniqueTask) {
+                    //同步心跳信息
+                    ClusterUniqueTask task = (ClusterUniqueTask) msg;
+                    this.workerId = task.currentSeverId;
+                    this.taskState = task.taskState;
+                    this.lastStateTime = task.lastStateTime;
+                    this.startTime = task.startTime;
+                    return Mono.empty();
+                } else if (msg instanceof OperationMessage) {
+                    //同步操作信息
+                    OperationMessage message = (OperationMessage) msg;
+                    if(!isDead){
+                        //任务存活
+                        return operation(message);
+                    }
+                }
+                return Mono.empty();
+            })
+            .then();
+    }
+    enum TaskOperation implements Serializable {
+        START,
+        PAUSE,
+        RELOAD,
+        SHUTDOWN,
+        EXECUTE,
+        ENABLE_DEBUG,
+        DISABLE_DEBUG
+    }
+
+    @Data
+    public static class OperationMessage implements Serializable{
+        private TaskOperation operation;
+        private List<Object> params;
+
+        private OperationMessage(TaskOperation operation, List<Object> params) {
+            this.operation = operation;
+            this.params = params;
+        }
+
+        public static OperationMessage of(TaskOperation operation, List<Object> params) {
+            return new OperationMessage(operation,params);
+        }
+
+        public static OperationMessage of(TaskOperation operation) {
+            return new OperationMessage(operation,null);
+        }
+    }
+}

jetlinks-components/rule-engine-component/src/main/java/org/jetlinks/community/rule/engine/cluster/ClusterUniqueWork.java

@@ -0,0 +1,91 @@
+package org.jetlinks.community.rule.engine.cluster;
+
+import lombok.Data;
+import org.jetlinks.core.cluster.ClusterManager;
+import org.jetlinks.core.event.EventBus;
+import org.jetlinks.rule.engine.api.scheduler.ScheduleJob;
+import org.jetlinks.rule.engine.api.task.ConditionEvaluator;
+import org.jetlinks.rule.engine.api.task.Task;
+import org.jetlinks.rule.engine.api.task.TaskExecutorProvider;
+import org.jetlinks.rule.engine.api.worker.Worker;
+import org.jetlinks.rule.engine.cluster.scope.ClusterGlobalScope;
+import org.jetlinks.rule.engine.cluster.worker.ClusterExecutionContext;
+import org.jetlinks.rule.engine.defaults.DefaultExecutionContext;
+import org.jetlinks.rule.engine.defaults.scope.InMemoryGlobalScope;
+import org.redisson.api.RedissonClient;
+import reactor.core.publisher.Mono;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName ClusterUniqueWork.java
+ * @Description 服务器节点
+ *
+ * 1、管理每个节点上的task
+ * 2、监听节点的创建与关闭
+ * @createTime 2021年11月12日 11:20:00
+ */
+@Data
+public class ClusterUniqueWork implements Worker, Serializable {
+    private String id;
+    private String name;
+
+    private final Map<String, TaskExecutorProvider> executors = new ConcurrentHashMap<>();
+
+
+    private final EventBus eventBus;
+
+    private final ConditionEvaluator conditionEvaluator;
+
+    private  ClusterManager clusterManager;
+
+    private static final InMemoryGlobalScope scope = new InMemoryGlobalScope();
+
+
+    private final RedissonClient redissonClient;
+
+    public ClusterUniqueWork(String id, String name, EventBus eventBus, ConditionEvaluator conditionEvaluator, ClusterManager clusterManager,RedissonClient redissonClient) {
+        this.id = id;
+        this.name = name;
+        this.eventBus = eventBus;
+        this.conditionEvaluator = conditionEvaluator;
+        this.clusterManager = clusterManager;
+        this.redissonClient=redissonClient;
+    }
+
+    @Override
+    public Mono<Task> createTask(String schedulerId, ScheduleJob job) {
+        return Mono.justOrEmpty(executors.get(job.getExecutor()))
+            .switchIfEmpty(Mono.error(() -> new UnsupportedOperationException("unsupported executor:" + job.getExecutor())))
+            .flatMap(provider -> {
+                DefaultExecutionContext context = createContext(job);
+                return provider.createTask(context)
+                    .map(executor -> new ClusterUniqueTask(schedulerId, context,executor,id,clusterManager,redissonClient));
+            });
+    }
+
+    @Override
+    public Mono<List<String>> getSupportExecutors() {
+        return Mono.just(new ArrayList<>(executors.keySet()));
+    }
+
+    @Override
+    public Mono<State> getState() {
+        return Mono.just(State.working);
+    }
+
+    protected DefaultExecutionContext createContext(ScheduleJob job) {
+        return new DefaultExecutionContext(getId(), job, eventBus, conditionEvaluator, scope);
+    }
+
+    public void addExecutor(TaskExecutorProvider provider) {
+        executors.put(provider.getExecutor(), provider);
+    }
+
+}

jetlinks-components/rule-engine-component/src/main/java/org/jetlinks/community/rule/engine/configuration/RuleEngineConfiguration.java

@@ -1,6 +1,7 @@
 package org.jetlinks.community.rule.engine.configuration;
 
 import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.core.cluster.ClusterManager;
 import org.jetlinks.core.event.EventBus;
 import org.jetlinks.rule.engine.api.RuleEngine;
 import org.jetlinks.rule.engine.api.scheduler.Scheduler;
@@ -8,6 +9,8 @@ import org.jetlinks.rule.engine.api.task.ConditionEvaluator;
 import org.jetlinks.rule.engine.api.task.TaskExecutorProvider;
 import org.jetlinks.rule.engine.api.worker.Worker;
 import org.jetlinks.rule.engine.cluster.ClusterRuleEngine;
+import org.jetlinks.rule.engine.cluster.worker.ClusterWorker;
+import org.jetlinks.rule.engine.cluster.worker.RemoteWorker;
 import org.jetlinks.rule.engine.condition.ConditionEvaluatorStrategy;
 import org.jetlinks.rule.engine.condition.DefaultConditionEvaluator;
 import org.jetlinks.rule.engine.condition.supports.DefaultScriptEvaluator;
@@ -21,6 +24,7 @@ import org.jetlinks.rule.engine.model.RuleModelParserStrategy;
 import org.jetlinks.rule.engine.model.antv.AntVG6RuleModelParserStrategy;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -52,6 +56,7 @@ public class RuleEngineConfiguration {
     }
 
     @Bean
+    @ConditionalOnBean(LocalWorker.class)
     public BeanPostProcessor autoRegisterStrategy(DefaultRuleModelParser defaultRuleModelParser,
                                                   DefaultConditionEvaluator defaultConditionEvaluator,
                                                   LocalWorker worker) {
@@ -90,15 +95,14 @@ public class RuleEngineConfiguration {
     }
 
     @Bean
+    @ConditionalOnMissingBean(Worker.class)
     public LocalWorker localWorker(EventBus eventBus, ConditionEvaluator evaluator) {
         return new LocalWorker("local", "local", eventBus, evaluator);
     }
 
-
     @Bean
     @ConditionalOnMissingBean(RuleEngine.class)
     public RuleEngine defaultRuleEngine(Scheduler scheduler) {
-//        ClusterRuleEngine clusterRuleEngine = new ClusterRuleEngine();
         return new DefaultRuleEngine(scheduler);
     }
 

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/message/ClusterMessage.java → jetlinks-core/src/main/java/org/jetlinks/core/cluster/message/ClusterMessage.java

@@ -1,13 +1,11 @@
-package org.jetlinks.community.standalone.configuration.cluster.message;
+package org.jetlinks.core.cluster.message;
 
-import cn.hutool.core.util.ObjectUtil;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import org.jetlinks.core.cluster.ClusterTopic;
 import org.jetlinks.core.message.Message;
 
 import java.io.Serializable;
-import java.util.UUID;
 
 /**
  * @author lifang
@@ -29,7 +27,8 @@ public class ClusterMessage implements ClusterTopic.TopicMessage, Serializable {
 
     private long timestamp;
 
-    private ServiceMessage service;
+    private Object accept;
+
     public ClusterMessage(Message payload, String fromServer, String address) {
         this.payload = payload;
         this.fromServer = fromServer;
@@ -37,13 +36,12 @@ public class ClusterMessage implements ClusterTopic.TopicMessage, Serializable {
         this.messageId = payload.getMessageId();
     }
 
-    public ClusterMessage(ServiceMessage service, String fromServer, String address) {
-        this.service = service;
+    public ClusterMessage(Object accept, String fromServer, String address) {
+        this.accept = accept;
         this.fromServer = fromServer;
         this.address = address;
         this.messageId = payload.getMessageId();
     }
-
     @Override
     public String getTopic() {
         return address;
@@ -53,14 +51,4 @@ public class ClusterMessage implements ClusterTopic.TopicMessage, Serializable {
     public Object getMessage() {
         return payload;
     }
-
-
-    @AllArgsConstructor(staticName = "of")
-    @Data
-    public static class ServiceMessage{
-        //任务消息
-        String id;
-        Object payload;
-        String type;
-    }
 }

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/message/ClusterMessageReply.java → jetlinks-core/src/main/java/org/jetlinks/core/cluster/message/ClusterMessageReply.java

@@ -1,4 +1,4 @@
-package org.jetlinks.community.standalone.configuration.cluster.message;
+package org.jetlinks.core.cluster.message;
 
 import lombok.Data;
 import org.jetlinks.core.message.Message;

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/entity/DeviceShadowEntity.java

@@ -13,6 +13,7 @@ import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.DeviceMetadata;
 import org.jetlinks.core.metadata.PropertyMetadata;
 import javax.persistence.Column;
+import javax.persistence.Version;
 import java.sql.JDBCType;
 import java.util.ArrayList;
 import java.util.Map;

jetlinks-manager/rule-engine-manager/src/main/java/org/jetlinks/community/rule/engine/device/RuleSceneTaskExecutorProvider.java

@@ -1,9 +1,7 @@
 package org.jetlinks.community.rule.engine.device;
 
-import com.alibaba.fastjson.JSONObject;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections.CollectionUtils;
 import org.hswebframework.web.bean.FastBeanCopier;
 import org.jetlinks.community.ValueObject;
 import org.jetlinks.community.rule.engine.entity.RuleSceneEntity;
@@ -11,7 +9,6 @@ import org.jetlinks.core.event.EventBus;
 import org.jetlinks.core.event.Subscription;
 import org.jetlinks.core.message.DeviceMessage;
 import org.jetlinks.core.metadata.Jsonable;
-import org.jetlinks.reactor.ql.DefaultReactorQLRecord;
 import org.jetlinks.reactor.ql.ReactorQL;
 import org.jetlinks.reactor.ql.ReactorQLContext;
 import org.jetlinks.reactor.ql.ReactorQLRecord;

jetlinks-manager/rule-engine-manager/src/main/java/org/jetlinks/community/rule/engine/entity/RuleModelEntity.java

@@ -11,6 +11,7 @@ import org.jetlinks.community.rule.engine.enums.RuleInstanceState;
 import javax.persistence.Column;
 import javax.persistence.GeneratedValue;
 import javax.persistence.Table;
+import javax.persistence.Version;
 import java.sql.JDBCType;
 
 @Getter

jetlinks-manager/rule-engine-manager/src/main/java/org/jetlinks/community/rule/engine/entity/RuleSceneEntity.java

@@ -26,6 +26,7 @@ import org.springframework.util.StringUtils;
 import javax.persistence.Column;
 import javax.persistence.GeneratedValue;
 import javax.persistence.Table;
+import javax.persistence.Version;
 import javax.validation.constraints.Pattern;
 import java.io.Serializable;
 import java.sql.JDBCType;
@@ -90,6 +91,10 @@ public class RuleSceneEntity  extends GenericEntity<String> implements RecordCre
     @Schema(description = "创建者ID")
     private String creatorId;
 
+    @Column(name = "version")
+//    @Version
+    private Integer version;
+
     @Data
     public static class Trigger implements Serializable{
         private DeviceTrigger device;
@@ -142,6 +147,7 @@ public class RuleSceneEntity  extends GenericEntity<String> implements RecordCre
         instanceEntity.setName("场景联动:" + name);
         instanceEntity.setModelMeta(JSON.toJSONString(this));
         instanceEntity.setCreateTimeNow();
+//        instanceEntity.setModelVersion(getVersion());
         return instanceEntity;
     }
 

jetlinks-standalone/pom.xml

@@ -70,11 +70,11 @@
             <artifactId>guava</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>org.hswebframework.web</groupId>
-            <artifactId>hsweb-system-dictionary</artifactId>
-            <version>${hsweb.framework.version}</version>
-        </dependency>
+        <!--<dependency>-->
+            <!--<groupId>org.hswebframework.web</groupId>-->
+            <!--<artifactId>hsweb-system-dictionary</artifactId>-->
+            <!--<version>${hsweb.framework.version}</version>-->
+        <!--</dependency>-->
 
         <dependency>
             <groupId>io.netty</groupId>

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/ClusterConfiguration.java

@@ -1,17 +1,25 @@
 package org.jetlinks.community.standalone.configuration;
 
+import org.jetlinks.community.rule.engine.cluster.ClusterUniqueWork;
 import org.jetlinks.community.standalone.configuration.cluster.ClusterDeviceMessageBrokeMessageBroker;
 import org.jetlinks.community.standalone.configuration.cluster.ClusterDeviceMessageConnector;
-import org.jetlinks.community.standalone.configuration.cluster.ruleEngine.ClusterRuleEngine;
 import org.jetlinks.core.cluster.ClusterManager;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.event.EventBus;
 import org.jetlinks.core.server.MessageHandler;
 import org.jetlinks.core.server.session.DeviceSessionManager;
-import org.jetlinks.rule.engine.api.RuleEngine;
-import org.jetlinks.rule.engine.api.scheduler.Scheduler;
-import org.jetlinks.rule.engine.defaults.DefaultRuleEngine;
+import org.jetlinks.rule.engine.api.task.ConditionEvaluator;
+import org.jetlinks.rule.engine.api.task.TaskExecutorProvider;
+import org.jetlinks.rule.engine.condition.ConditionEvaluatorStrategy;
+import org.jetlinks.rule.engine.condition.DefaultConditionEvaluator;
+import org.jetlinks.rule.engine.defaults.LocalWorker;
+import org.jetlinks.rule.engine.model.DefaultRuleModelParser;
+import org.jetlinks.rule.engine.model.RuleModelParserStrategy;
 import org.jetlinks.supports.config.ClusterConfigStorageManager;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -39,7 +47,6 @@ public class ClusterConfiguration {
 
 
     @Bean(initMethod = "init")
-    @ConditionalOnProperty(name = "jetlinks.cluster",havingValue = "true")
     public ClusterDeviceMessageConnector clusterDeviceMessageConnector(EventBus eventBus,
                                                                        MessageHandler messageHandler,
                                                                        DeviceSessionManager sessionManager,
@@ -51,8 +58,33 @@ public class ClusterConfiguration {
 
 
     @Bean
-    public RuleEngine defaultRuleEngine(Scheduler scheduler,ClusterManager clusterManager,JetLinksProperties jetLinksProperties) {
-        return new ClusterRuleEngine(scheduler,clusterManager,jetLinksProperties.getServerId());
+    public BeanPostProcessor autoRegisterStrategy(DefaultRuleModelParser defaultRuleModelParser,
+                                                  DefaultConditionEvaluator defaultConditionEvaluator,
+                                                  ClusterUniqueWork worker) {
+        return new BeanPostProcessor() {
+            @Override
+            public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+                return bean;
+            }
+            @Override
+            public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+                if (bean instanceof RuleModelParserStrategy) {
+                    defaultRuleModelParser.register(((RuleModelParserStrategy) bean));
+                }
+                if (bean instanceof ConditionEvaluatorStrategy) {
+                    defaultConditionEvaluator.register(((ConditionEvaluatorStrategy) bean));
+                }
+                if (bean instanceof TaskExecutorProvider) {
+                    worker.addExecutor(((TaskExecutorProvider) bean));
+                }
+                return bean;
+            }
+        };
     }
 
+    @Bean
+    public ClusterUniqueWork clusterUniqueWork(JetLinksProperties properties, EventBus eventBus, ConditionEvaluator evaluator, ClusterManager clusterManager,
+                                               RedissonClient redissonClient) {
+        return new ClusterUniqueWork(properties.getServerId(), properties.getClusterName()+properties.getServerId(), eventBus, evaluator,clusterManager,redissonClient);
+    }
 }

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/JetLinksConfiguration.java

@@ -16,6 +16,7 @@ import org.jetlinks.community.device.service.AutoDiscoverDeviceRegistry;
 import org.jetlinks.community.device.service.data.DeviceDataService;
 import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
 import org.jetlinks.community.micrometer.MeterRegistryManager;
+import org.jetlinks.community.standalone.configuration.cluster.RedisClusterManager;
 import org.jetlinks.core.ProtocolSupports;
 import org.jetlinks.core.cluster.ClusterManager;
 import org.jetlinks.core.config.ConfigStorageManager;
@@ -30,7 +31,6 @@ import org.jetlinks.core.server.monitor.GatewayServerMonitor;
 import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.core.spi.ServiceContext;
 import org.jetlinks.supports.cluster.ClusterDeviceRegistry;
-import org.jetlinks.supports.cluster.redis.RedisClusterManager;
 import org.jetlinks.supports.config.EventBusStorageManager;
 import org.jetlinks.supports.event.BrokerEventBus;
 import org.jetlinks.supports.protocol.ServiceLoaderProtocolSupports;

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/ClusterDeviceMessageBrokeMessageBroker.java

@@ -1,7 +1,7 @@
 package org.jetlinks.community.standalone.configuration.cluster;
 
 import lombok.extern.slf4j.Slf4j;
-import org.jetlinks.community.standalone.configuration.cluster.message.ClusterMessage;
+import org.jetlinks.core.cluster.message.ClusterMessage;
 import org.jetlinks.core.cluster.ClusterManager;
 import org.jetlinks.core.cluster.ServerNode;
 import org.jetlinks.core.device.StandaloneDeviceMessageBroker;
@@ -19,7 +19,7 @@ import java.util.stream.Collectors;
  * @author lifang
  * @version 1.0.0
  * @ClassName ClusterDeviceMessageBrokeMessageBroker.java
- * @Description 上行数据
+ * @Description 下行数据
  * @createTime 2021年11月05日 08:47:00
  */
 @Slf4j

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/ClusterDeviceMessageConnector.java

@@ -5,7 +5,7 @@ import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.jetlinks.community.device.message.DeviceMessageConnector;
-import org.jetlinks.community.standalone.configuration.cluster.message.ClusterMessage;
+import org.jetlinks.core.cluster.message.ClusterMessage;
 import org.jetlinks.core.cluster.ClusterManager;
 import org.jetlinks.core.cluster.ServerNode;
 import org.jetlinks.core.device.DeviceOperator;

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/RedisClusterManager.java

@@ -0,0 +1,121 @@
+package org.jetlinks.community.standalone.configuration.cluster;
+
+import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.core.cache.Caches;
+import org.jetlinks.core.cluster.*;
+import org.jetlinks.supports.cluster.redis.RedisClusterCache;
+import org.jetlinks.supports.cluster.redis.RedisClusterCounter;
+import org.jetlinks.supports.cluster.redis.RedisClusterNotifier;
+import org.jetlinks.supports.cluster.redis.RedisClusterSet;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import reactor.core.publisher.Flux;
+
+import java.time.Duration;
+import java.util.Map;
+
+@SuppressWarnings("all")
+@Slf4j
+public class RedisClusterManager implements ClusterManager {
+
+    private String clusterName;
+
+    private String serverId;
+
+    private Map<String, ClusterTopic> topics = Caches.newCache();
+    private Map<String, ClusterCache> caches = Caches.newCache();
+    private Map<String, ClusterSet> sets = Caches.newCache();
+
+    private ReactiveRedisTemplate<?, ?> commonOperations;
+
+    private RedisHaManager haManager;
+
+    private RedisClusterNotifier notifier;
+
+    private ReactiveRedisOperations<String, String> stringOperations;
+
+    private ReactiveRedisTemplate<String, ?> queueRedisTemplate;
+
+    public RedisClusterManager(String name, ServerNode serverNode, ReactiveRedisTemplate<?, ?> operations) {
+        this.clusterName = name;
+        this.commonOperations = operations;
+        this.notifier = new RedisClusterNotifier(name, serverNode.getId(), this);
+        this.serverId = serverNode.getId();
+        this.haManager = new RedisHaManager(name, serverNode, this, (ReactiveRedisTemplate) operations);
+        this.stringOperations = new ReactiveRedisTemplate<>(operations.getConnectionFactory(), RedisSerializationContext.string());
+
+        this.queueRedisTemplate = new ReactiveRedisTemplate<>(operations.getConnectionFactory(),
+                RedisSerializationContext.<String, Object>newSerializationContext()
+                        .key(RedisSerializer.string())
+                        .value((RedisSerializationContext.SerializationPair<Object>) operations.getSerializationContext().getValueSerializationPair())
+                        .hashKey(RedisSerializer.string())
+                        .hashValue(operations.getSerializationContext().getHashValueSerializationPair())
+                        .build());
+    }
+
+    public RedisClusterManager(String name, String serverId, ReactiveRedisTemplate<?, ?> operations) {
+        this(name, ServerNode.builder().id(serverId).build(), operations);
+    }
+
+    @Override
+    public String getCurrentServerId() {
+        return serverId;
+    }
+
+    public void startup() {
+        this.notifier.startup();
+        this.haManager.startup();
+    }
+
+    public void shutdown() {
+        this.haManager.shutdown();
+    }
+
+    @Override
+    public HaManager getHaManager() {
+        return haManager;
+    }
+
+    @SuppressWarnings("all")
+    protected <K, V> ReactiveRedisTemplate<K, V> getRedis() {
+        return (ReactiveRedisTemplate<K, V>) commonOperations;
+    }
+
+    @Override
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    public ClusterNotifier getNotifier() {
+        return notifier;
+    }
+
+    @Override
+    @Deprecated
+    public <T> ClusterQueue<T> getQueue(String queueId) {
+//        return queues.computeIfAbsent(queueId, id -> new RedisClusterQueue<>(id, this.queueRedisTemplate));
+        return null;
+    }
+
+    @Override
+    public <T> ClusterTopic<T> getTopic(String topic) {
+        return topics.computeIfAbsent(topic, id -> new RedisClusterTopic(id, this.getRedis()));
+    }
+
+    @Override
+    public <K, V> ClusterCache<K, V> getCache(String cache) {
+        return caches.computeIfAbsent(cache, id -> new RedisClusterCache<K, V>(id, this.getRedis()));
+    }
+
+    @Override
+    public <V> ClusterSet<V> getSet(String name) {
+        return sets.computeIfAbsent(name, id -> new RedisClusterSet<V>(id, this.getRedis()));
+    }
+
+    @Override
+    public ClusterCounter getCounter(String name) {
+        return new RedisClusterCounter(stringOperations, clusterName + ":counter:" + name);
+    }
+}

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/RedisClusterTopic.java

@@ -0,0 +1,70 @@
+package org.jetlinks.community.standalone.configuration.cluster;
+
+import org.jetlinks.core.cluster.ClusterTopic;
+import org.reactivestreams.Publisher;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import reactor.core.Disposable;
+import reactor.core.publisher.*;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class RedisClusterTopic<T> implements ClusterTopic<T> {
+
+    private final String topicName;
+
+    private final ReactiveRedisOperations<Object, T> operations;
+
+    private final FluxProcessor<TopicMessage<T>, TopicMessage<T>> processor;
+
+    private final FluxSink<TopicMessage<T>> sink;
+
+    private final AtomicBoolean subscribed = new AtomicBoolean();
+
+    public RedisClusterTopic(String topic, ReactiveRedisOperations<Object, T> operations) {
+        this.topicName = topic;
+        this.operations = operations;
+        processor = EmitterProcessor.create(false);
+        sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
+    }
+
+    private Disposable disposable;
+
+    private void doSubscribe() {
+        if (subscribed.compareAndSet(false, true)) {
+            disposable = operations
+                    .listenToPattern(topicName)
+                    .subscribe(data -> {
+                        if (!processor.hasDownstreams()) {
+                            disposable.dispose();
+                            subscribed.compareAndSet(true, false);
+                        } else {
+                            sink.next(new TopicMessage<T>() {
+                                @Override
+                                public String getTopic() {
+                                    return data.getChannel();
+                                }
+
+                                @Override
+                                public T getMessage() {
+                                    return data.getMessage();
+                                }
+                            });
+                        }
+                    });
+        }
+    }
+
+    @Override
+    public Flux<TopicMessage<T>> subscribePattern() {
+        return processor
+                .doOnSubscribe((r) -> doSubscribe());
+    }
+
+    @Override
+    public Mono<Integer> publish(Publisher<? extends T> publisher) {
+        return Flux.from(publisher)
+                .flatMap(data -> operations.convertAndSend(topicName, data))
+                .last(1L)
+                .map(Number::intValue);
+    }
+}

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/RedisHaManager.java

@@ -0,0 +1,210 @@
+package org.jetlinks.community.standalone.configuration.cluster;
+
+import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.core.cluster.ClusterManager;
+import org.jetlinks.core.cluster.ClusterTopic;
+import org.jetlinks.core.cluster.HaManager;
+import org.jetlinks.core.cluster.ServerNode;
+import org.springframework.data.redis.core.ReactiveHashOperations;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import reactor.core.Disposable;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxProcessor;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class RedisHaManager implements HaManager {
+    private final ServerNode current;
+
+    private final String haName;
+
+    private final ClusterTopic<ServerNode> offlineTopic;
+
+    private ClusterManager clusterManager;
+
+    private final Map<String, ServerNode> allNode = new ConcurrentHashMap<>();
+
+    private final ReactiveRedisOperations<String, ServerNode> operations;
+
+    private final ClusterTopic<ServerNode> keepalive;
+
+    private final ReactiveHashOperations<String, String, ServerNode> inRedisNode;
+    private final String allNodeHashKey;
+
+
+    private final FluxProcessor<ServerNode, ServerNode> onlineProcessor = EmitterProcessor.create(false);
+    private final FluxProcessor<ServerNode, ServerNode> offlineProcessor = EmitterProcessor.create(false);
+    private volatile boolean started = false;
+
+    public RedisHaManager(String name,
+                          ServerNode current,
+                          ClusterManager clusterManager,
+                          ReactiveRedisOperations<String, ServerNode> operations) {
+        this.haName = name;
+        this.current = current.copy();
+        this.current.setUptime(System.currentTimeMillis());
+        this.current.setLeader(false);
+        this.clusterManager = clusterManager;
+        this.operations = operations;
+        this.inRedisNode = this.operations.opsForHash();
+        this.offlineTopic = clusterManager.getTopic("__ha_offline_topic:".concat(haName));
+        this.keepalive = clusterManager.getTopic("__ha_keepalive:".concat(haName));
+        this.allNodeHashKey = "__ha_all_node:".concat(haName);
+
+    }
+
+    public void checkAlive() {
+        current.setLastKeepAlive(System.currentTimeMillis());
+
+        inRedisNode.put(allNodeHashKey, current.getId(), current)
+                   .subscribe();
+
+        keepalive.publish(Mono.just(current)).subscribe();
+
+        Map<String, ServerNode> maybeOffline = getAllNode()
+                .stream()
+                .filter(node -> System.currentTimeMillis() - node.getLastKeepAlive() > TimeUnit.SECONDS.toMillis(30))
+                .filter(node -> !node.isSame(current))
+                .collect(Collectors.toMap(ServerNode::getId, Function.identity()));
+
+        //检查节点是否离线
+        inRedisNode.keys(allNodeHashKey)
+                   .filter(maybeOffline::containsKey)
+                   .map(maybeOffline::get)
+                   .collectList()
+                   .filter(list -> !list.isEmpty())
+                   .flatMapMany(list -> inRedisNode
+                           .remove(allNodeHashKey, list.stream().map(ServerNode::getId).toArray())
+                           .thenMany(Flux.fromIterable(list))
+                   )
+                   .as(offlineTopic::publish)
+                   .subscribe();
+    }
+
+    private void electionLeader() {
+        allNode.values()
+               .stream()
+               .peek(serverNode -> serverNode.setLeader(false))
+               .min(Comparator.comparing(ServerNode::getUptime))
+               .ifPresent(serverNode -> serverNode.setLeader(true));
+    }
+
+    public void shutdown() {
+        inRedisNode
+                .remove(allNodeHashKey, current.getId())
+                .then(offlineTopic
+                              .publish(Mono.just(current)))
+                .block();
+    }
+
+    public synchronized void startup() {
+        if (started) {
+            return;
+        }
+        started = true;
+        allNode.put(current.getId(), current);
+
+        //注册自己
+        inRedisNode.put(allNodeHashKey, current.getId(), current)
+                   .flatMapMany(r -> inRedisNode.values(allNodeHashKey))
+                   .collectList()
+                   .doOnNext(node -> {
+                       for (ServerNode serverNode : node) {
+                           serverNode.setLastKeepAlive(System.currentTimeMillis());
+                           allNode.put(serverNode.getId(), serverNode);
+                       }
+                       electionLeader();
+                       Flux.interval(Duration.ZERO, Duration.ofSeconds(5))
+                           .doOnNext(i -> this.checkAlive())
+                           .subscribe();
+                   })
+                   .block();
+
+        offlineTopic.subscribe()
+                    .subscribe(serverNode -> {
+                        //自己
+                        if (currentServer().isSame(serverNode)) {
+                            return;
+                        }
+                        if (allNode.remove(serverNode.getId()) != null) {
+                            log.debug("[{}]:server node [{}] offline", haName, serverNode.getId());
+                            //node offline
+                            inRedisNode
+                                    .remove(allNodeHashKey, serverNode.getId())
+                                    .subscribe();
+                            electionLeader();
+                            if (offlineProcessor.hasDownstreams()) {
+                                offlineProcessor.onNext(serverNode);
+                            }
+                        }
+                    });
+        //其他节点定时发布
+        keepalive.subscribe()
+                 .subscribe(serverNode -> {
+                     //自己
+                     if (currentServer().isSame(serverNode)) {
+                         return;
+                     }
+                     serverNode.setLastKeepAlive(System.currentTimeMillis());
+                     allNode.compute(serverNode.getId(), (id, node) -> {
+                         if (node != null) {
+                             node.setLastKeepAlive(System.currentTimeMillis());
+                             return node;
+                         }
+                         return null;
+                     });
+                     if (!allNode.containsKey(serverNode.getId())) {
+                         allNode.put(serverNode.getId(), serverNode);
+                         electionLeader();
+                         log.debug("[{}]:server node [{}] online", haName, serverNode.getId());
+                         //node join
+                         if (onlineProcessor.hasDownstreams()) {
+                             onlineProcessor.onNext(serverNode);
+                         }
+                     }
+
+                 });
+
+
+    }
+
+    @Override
+    public ServerNode currentServer() {
+        return current;
+    }
+
+    @Override
+    public Flux<ServerNode> subscribeServerOnline() {
+        return onlineProcessor
+                .filter(node -> !node.getId().equals(current.getId()));
+    }
+
+    @Override
+    public Flux<ServerNode> subscribeServerOffline() {
+        return offlineProcessor
+                .filter(node -> !node.getId().equals(current.getId()));
+    }
+
+    @Override
+    public List<ServerNode> getAllNode() {
+        return new ArrayList<>(allNode.values());
+    }
+
+    @Override
+    public Disposable doOnReBalance(Consumer<List<ServerNode>> runnable) {
+        return () -> {
+        };
+    }
+}

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/ruleEngine/ClusterRuleEngine.java

@@ -1,111 +0,0 @@
-package org.jetlinks.community.standalone.configuration.cluster.ruleEngine;
-
-import lombok.AllArgsConstructor;
-import org.jetlinks.community.rule.engine.enums.RuleInstanceState;
-import org.jetlinks.community.standalone.configuration.cluster.message.ClusterMessage;
-import org.jetlinks.core.cluster.ClusterManager;
-import org.jetlinks.rule.engine.api.model.RuleModel;
-import org.jetlinks.rule.engine.api.scheduler.Scheduler;
-import org.jetlinks.rule.engine.api.task.Task;
-import org.jetlinks.rule.engine.defaults.DefaultRuleEngine;
-import reactor.core.Disposable;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
-
-import javax.annotation.PostConstruct;
-import java.time.Duration;
-
-/**
- * @author lifang
- * @version 1.0.0
- * @ClassName ClusterRuleEngine.java
- * @Description TODO
- * @createTime 2021年11月10日 16:10:00
- */
-
-
-public class ClusterRuleEngine extends DefaultRuleEngine {
-    private ClusterManager clusterManager;
-    private volatile  boolean started=false;
-    private String serverId;
-    private Disposable result =null;
-    public ClusterRuleEngine(Scheduler scheduler) {
-        super(scheduler);
-    }
-
-    public ClusterRuleEngine(Scheduler scheduler, ClusterManager clusterManager, String serverId) {
-        super(scheduler);
-        this.clusterManager = clusterManager;
-        this.serverId = serverId;
-    }
-
-    @PostConstruct
-    public synchronized void init(){
-        if(started){
-            return;
-        }
-        started=true;
-        Flux.interval(Duration.ofSeconds(10))
-            .doOnNext(ignore->{
-                if(result==null||result.isDisposed()){
-                    result = clusterManager
-                        .getTopic("cluster-rule-engine")
-                        .subscribePattern()
-                        .publishOn(Schedulers.boundedElastic())
-                        .flatMap(obj -> {
-                            if (obj instanceof ClusterMessage) {
-                                ClusterMessage clusterMessage = (ClusterMessage) obj;
-                                String fromServer = clusterMessage.getFromServer();
-                                if (this.serverId.equals(fromServer)) {
-                                    return Flux.empty();
-                                }
-                                ClusterMessage.ServiceMessage serviceMessage = clusterMessage.getService();
-                                String type = serviceMessage.getType();
-                                RuleInstanceState ruleInstanceState = RuleInstanceState.valueOf(type);
-                                return execute(ruleInstanceState, serviceMessage.getId(), serviceMessage.getPayload());
-                            }
-                            return Flux.empty();
-                        }).subscribe();
-                }
-            }).subscribe();
-
-        //广播消息
-
-    }
-
-    private Mono<Void> execute(RuleInstanceState state, String id, Object payload){
-        RuleModel model= (RuleModel) payload;
-        switch (state){
-            case started:
-                return super.startRule(id,model).then();
-            case stopped:
-                return super.shutdown(id);
-            case disable:
-            default:return Mono.empty();
-        }
-    }
-
-    @Override
-    public Flux<Task> startRule(String instanceId, RuleModel model) {
-        ClusterMessage.ServiceMessage serviceMessage = ClusterMessage.ServiceMessage.of(instanceId, model, RuleInstanceState.started.getValue());
-        return super.startRule(instanceId, model)
-            .publishOn(Schedulers.parallel())
-            .doOnNext(ignore->clusterManager.getTopic("cluster-rule-engine")
-                .publish(Mono.just(
-                    new ClusterMessage(
-                        serviceMessage,this.serverId,null))).subscribe());
-    }
-
-    @Override
-    public Mono<Void> shutdown(String instanceId) {
-        ClusterMessage.ServiceMessage serviceMessage = ClusterMessage.ServiceMessage.of(instanceId, null, RuleInstanceState.stopped.getValue());
-        return super.shutdown(instanceId)
-            .publishOn(Schedulers.parallel())
-            .doOnNext(ignore->clusterManager.getTopic("cluster-rule-engine")
-                .publish(Mono.just(
-                    new ClusterMessage(
-                        serviceMessage,this.serverId,null))).subscribe());
-    }
-
-}