首次提交

.gitignore

@@ -9,6 +9,12 @@
*.war
*.ear
### IntelliJ IDEA ###
.idea
*.iws
*.iml
*.ipr
# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
hs_err_pid*

pom.xml

@@ -0,0 +1,127 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.yihu.hos</groupId>
        <artifactId>commons</artifactId>
        <version>1.3.0</version>
        <relativePath>..</relativePath>
    </parent>
    <artifactId>hos-arbiter</artifactId>
    <version>1.3.0</version>
    <packaging>jar</packaging>
    <name>hos-arbiter</name>
    <description>ESB服务状态协调模块</description>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
        <!--<spring-data-releasetrain.version>Fowler-SR2</spring-data-releasetrain.version>-->
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</artifactId>
            <exclusions>
                <exclusion>
                    <artifactId>netty-codec-http</artifactId>
                    <groupId>io.netty</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>netty-transport</artifactId>
                    <groupId>io.netty</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>netty-transport-native-epoll</artifactId>
                    <groupId>io.netty</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>netty-handler</artifactId>
                    <groupId>io.netty</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>netty-buffer</artifactId>
                    <groupId>io.netty</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>netty-codec</artifactId>
                    <groupId>io.netty</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-config</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <!-- test start -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- test end -->
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-core</artifactId>
        </dependency>
        <dependency>
            <groupId>com.yihu.hos</groupId>
            <artifactId>hos-web-framework</artifactId>
            <version>${hos-version}</version>
        </dependency>
        <dependency>
            <groupId>io.zbus</groupId>
            <artifactId>zbus</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-jms</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>${spring-boot-version}</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <mainClass>com.yihu.hos.arbiter.HosArbiterApplication</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

src/main/java/com/yihu/hos/arbiter/HosArbiterApplication.java

@@ -0,0 +1,30 @@
package com.yihu.hos.arbiter;
import com.yihu.hos.arbiter.services.ProxyService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
@EnableEurekaClient
@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class, HibernateJpaAutoConfiguration.class})
public class HosArbiterApplication implements CommandLineRunner {
    private ProxyService proxyService;
    public static void main(String[] args) {
        SpringApplication.run(HosArbiterApplication.class, args);
    }
    @Autowired
    public void setProxyService(ProxyService proxyService) {
        this.proxyService = proxyService;
    }
    @Override
    public void run(String... strings) throws Exception {
        proxyService.start();
    }
}

src/main/java/com/yihu/hos/arbiter/configuration/ActivemqConfiguration.java

@@ -0,0 +1,29 @@
package com.yihu.hos.arbiter.configuration;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
/**
 * @created Airhead 2016/8/4.
 */
@Configuration
public class ActivemqConfiguration {
    @Value("${spring.activemq.broker-url}")
    private String brokerURL;
    @Value("${spring.activemq.user}")
    private String user;
    @Value("${spring.activemq.password}")
    private String password;
    public String getBrokerURL() {
        return brokerURL;
    }
    public String getUser() {
        return user;
    }
    public String getPassword() {
        return password;
    }
}

src/main/java/com/yihu/hos/arbiter/configuration/ArbiterServerConfiguration.java

@@ -0,0 +1,54 @@
package com.yihu.hos.arbiter.configuration;
import io.zbus.mq.Broker;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @author Airhead
 * @since 2016/12/9.
 */
@Configuration
public class ArbiterServerConfiguration {
    @Value("${arbiter.central.url}")
    private String centralUrl;
    @Value("${arbiter.tenant.name}")
    private String tenant;
    @Value("${arbiter.mycat.name}")
    private String mycatName;
    private Broker broker;
    public String getMycatName() {
        return mycatName;
    }
    public String getCentralUrl() {
        return centralUrl;
    }
    public String getTenant() {
        return tenant;
    }
    @Bean
    public Broker getBroker() {
        broker = new Broker(this.centralUrl);
        return broker;
    }
    @Override
    protected void finalize() throws Throwable {
        if (broker != null) {
            broker.close();
        }
        super.finalize();
    }
}

src/main/java/com/yihu/hos/arbiter/controllers/ArbiterServerController.java

@@ -0,0 +1,36 @@
package com.yihu.hos.arbiter.controllers;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.web.framework.model.bo.ArbiterServer;
import com.yihu.hos.arbiter.services.ArbiterServerService;
import com.yihu.hos.core.datatype.DateUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.io.IOException;
/**
 * @author Airhead
 * @since 2016/12/9.
 */
@RestController
@RequestMapping("/arbiter")
public class ArbiterServerController {
    @Autowired
    private ArbiterServerService arbiterServerService;
    @Autowired
    private ObjectMapper objectMapper;
    @RequestMapping()
    public void register(@RequestBody String body) {
        try {
            ArbiterServer arbiterServer = objectMapper.readValue(body, ArbiterServer.class);
            arbiterServer.setCreateTime(DateUtil.getSysDateTime());
            arbiterServerService.save(arbiterServer);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

src/main/java/com/yihu/hos/arbiter/controllers/BrokerServerController.java

@@ -0,0 +1,75 @@
package com.yihu.hos.arbiter.controllers;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.arbiter.services.BrokerServerService;
import com.yihu.hos.core.datatype.DateUtil;
import com.yihu.hos.web.framework.model.bo.BrokerServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import java.io.IOException;
/**
 * @created Airhead 2016/7/27.
 */
@RestController
@RequestMapping("/brokerServer")
public class BrokerServerController {
    @Autowired
    private BrokerServerService brokerServerService;
    @RequestMapping(method = RequestMethod.GET)
    public String get() {
        try {
            BrokerServer brokerServer = brokerServerService.get();
            if (brokerServer == null) {
                return "{}";
            }
            ObjectMapper objectMapper = new ObjectMapper();
            return objectMapper.writeValueAsString(brokerServer);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "{}";
    }
    @RequestMapping(value = "login", method = RequestMethod.POST)
    public void login(String brokerServer) {
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            BrokerServer server = objectMapper.readValue(brokerServer, BrokerServer.class);
            server.setCreateTime(DateUtil.getSysDateTime());
            brokerServerService.login(server);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    @RequestMapping(value = "heartbeat", method = RequestMethod.POST)
    public void heartbeat(String brokerServer) {
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            BrokerServer server = objectMapper.readValue(brokerServer, BrokerServer.class);
            brokerServerService.heartbeat(server);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    @RequestMapping(value = "logout", method = RequestMethod.POST)
    public void logout(String brokerServer) {
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            BrokerServer server = objectMapper.readValue(brokerServer, BrokerServer.class);
            brokerServerService.logout(server);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

src/main/java/com/yihu/hos/arbiter/controllers/EndpointController.java

@@ -0,0 +1,44 @@
package com.yihu.hos.arbiter.controllers;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.web.framework.model.bo.Endpoint;
import com.yihu.hos.arbiter.services.EndpointService;
import com.yihu.hos.core.datatype.DateUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import java.io.IOException;
/**
 * @created Airhead 2016/7/27.
 */
@RestController
@RequestMapping("/endpoint")
public class EndpointController {
    @Autowired
    private EndpointService endpointService;
    @RequestMapping(method = RequestMethod.POST)
    public void save(String service) {
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            Endpoint endpoint = objectMapper.readValue(service, Endpoint.class);
            endpoint.setCreateTime(DateUtil.getSysDateTime());
            endpointService.save(endpoint);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    @RequestMapping(method = RequestMethod.GET)
    public String get(String serviceName) {
        return endpointService.get(serviceName).toString();
    }
    @RequestMapping(method = RequestMethod.PUT)
    public String put(String serviceName, String ClientInfo) {
        return endpointService.get(serviceName).toString();
    }
}

src/main/java/com/yihu/hos/arbiter/controllers/ServiceFlowController.java

@@ -0,0 +1,19 @@
package com.yihu.hos.arbiter.controllers;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
/**
 * @author Airhead
 * @since 2017/2/6.
 */
@RestController
@RequestMapping("/serviceFlow")
public class ServiceFlowController {
    @RequestMapping(method = RequestMethod.GET)
    public String get() {
        return "{}";
    }
}

src/main/java/com/yihu/hos/arbiter/listener/ApplicationStartListener.java

@@ -0,0 +1,25 @@
package com.yihu.hos.arbiter.listener;
import com.yihu.hos.arbiter.services.ArbiterServerService;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
public class ApplicationStartListener implements ApplicationListener<ContextRefreshedEvent> {
    private static Logger logger = LoggerFactory.getLogger(ApplicationStartListener.class);
    @Autowired
    private ArbiterServerService arbiterServerService;
    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        arbiterServerStart();
    }
    public void arbiterServerStart() {
//        arbiterServerService.start();   //后期可以统一到Router上来处理
    }
}

src/main/java/com/yihu/hos/arbiter/routers/ArbiterServerRouter.java

@@ -0,0 +1,17 @@
package com.yihu.hos.arbiter.routers;
import org.apache.camel.builder.RouteBuilder;
/**
 * @author Airhead
 * @since 2016/12/19.
 */
//因为配置问题，暂时服务启动，先屏蔽，才使用的是Spring的事件来注册的。
//@Component
public class ArbiterServerRouter  extends RouteBuilder {
    @Override
    public void configure() throws Exception {
//        from("direct:arbiterServer")
//                .to("bean:arbiterServerService?method=serverServiceFlow");
    }
}

src/main/java/com/yihu/hos/arbiter/routers/EndpointEventRouter.java

@@ -0,0 +1,36 @@
package com.yihu.hos.arbiter.routers;
import com.yihu.hos.arbiter.configuration.ActivemqConfiguration;
import com.yihu.hos.web.framework.constant.EndPointConstant;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.model.ModelCamelContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.jms.ConnectionFactory;
/**
 * @created Airhead 2016/8/1.
 */
@Component
public class EndpointEventRouter extends RouteBuilder {
    @Autowired
    private ActivemqConfiguration activemqConfiguration;
    @Override
    public void configure() throws Exception {
        ModelCamelContext context = this.getContext();
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                activemqConfiguration.getUser(), activemqConfiguration.getPassword(), activemqConfiguration.getBrokerURL());
        // Note we can explicit name the component
        context.addComponent(EndPointConstant.CAMEL_COMPONENT, JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
        from(EndPointConstant.CAMEL_ENDPOINT)
                .choice()
                .when(header("event").isEqualTo(EndPointConstant.ADD_ENDPOINT)).to("bean:endpointService?method=addEndpoint")
                .when(header("event").isEqualTo(EndPointConstant.DELETE_ENDPOINT)).to("bean:endpointService?method=deleteEndpoint")
                .when(header("event").isEqualTo(EndPointConstant.OFF_SERVER)).to("bean:endpointService?method=offServer")
                .endChoice();
    }
}

src/main/java/com/yihu/hos/arbiter/routers/MycatRouter.java

@@ -0,0 +1,39 @@
package com.yihu.hos.arbiter.routers;
import com.yihu.hos.arbiter.configuration.ActivemqConfiguration;
import com.yihu.hos.web.framework.constant.MycatConstant;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.model.ModelCamelContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.jms.ConnectionFactory;
/**
 * mycat操作  消息router
 * @author HZY
 * @vsrsion 1.0
 * Created at 2017/1/5.
 */
@Component
public class MycatRouter extends RouteBuilder {
    @Autowired
    private ActivemqConfiguration activemqConfiguration;
    @Override
    public void configure() throws Exception {
        ModelCamelContext context = this.getContext();
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                activemqConfiguration.getUser(), activemqConfiguration.getPassword(), activemqConfiguration.getBrokerURL());
        // Note we can explicit name the component
        context.addComponent(MycatConstant.CAMEL_COMPONENT, JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
        from(MycatConstant.CAMEL_ENDPOINT)
                .choice()
                .when(header("tenant").isNotNull()).to("bean:mycatService?method=handleServiceFlow")
                .when(header("event").isEqualTo(MycatConstant.EXECUTE_MYCAT)).to("bean:mycatService?method=updateMycat")
                .endChoice();
    }
}

src/main/java/com/yihu/hos/arbiter/routers/SerivceHealthRouter.java

@@ -0,0 +1,15 @@
package com.yihu.hos.arbiter.routers;
import org.apache.camel.builder.RouteBuilder;
/**
 * @created Airhead 2016/8/1.
 */
//@Component
public class SerivceHealthRouter extends RouteBuilder {
    @Override
    public void configure() throws Exception {
//        from("timer:healthCheck?period={{arbiter.timer.period}}")
//                .to("bean:endpointService?method=check");
    }
}

src/main/java/com/yihu/hos/arbiter/routers/ServiceFlowEventRouter.java

@@ -0,0 +1,42 @@
package com.yihu.hos.arbiter.routers;
import com.yihu.hos.arbiter.configuration.ActivemqConfiguration;
import com.yihu.hos.web.framework.constant.ServiceFlowConstant;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.model.ModelCamelContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.jms.ConnectionFactory;
/**
 * @created Airhead 2016/8/1.
 */
@Component
public class ServiceFlowEventRouter extends RouteBuilder {
    @Autowired
    private ActivemqConfiguration activemqConfiguration;
    @Override
    public void configure() throws Exception {
        ModelCamelContext context = this.getContext();
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                activemqConfiguration.getUser(), activemqConfiguration.getPassword(), activemqConfiguration.getBrokerURL());
        // Note we can explicit name the component
        context.addComponent(ServiceFlowConstant.CAMEL_COMPONENT, JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
        from(ServiceFlowConstant.CAMEL_ENDPOINT)
                .choice()
//                .when(header("tenant").isNotNull()).to("bean:serviceFlowService?method=handleServiceFlow")
                .when(header("event").isEqualTo(ServiceFlowConstant.SERVICE_FLOW_STARTED)).to("bean:serviceFlowService?method=serviceFlowStarted")
                .when(header("event").isEqualTo(ServiceFlowConstant.SERVICE_FLOW_STOPPED)).to("bean:serviceFlowService?method=serviceFlowStopped")
                .when(header("event").isEqualTo(ServiceFlowConstant.SERVICE_FLOW_ADDED)).to("bean:serviceFlowService?method=serviceFlowAdd")
                .when(header("event").isEqualTo(ServiceFlowConstant.SERVICE_FLOW_MODIFIED_ADD)).to("bean:serviceFlowService?method=serviceFlowModifyAdd")
                .when(header("event").isEqualTo(ServiceFlowConstant.SERVICE_FLOW_MODIFIED_REDUCE)).to("bean:serviceFlowService?method=serviceFlowModifyReduce")
                .when(header("event").isEqualTo(ServiceFlowConstant.SERVICE_FLOW_DELETED)).to("bean:serviceFlowService?method=serviceFlowDelete")
                .when(header("event").isEqualTo(ServiceFlowConstant.BROKER_SERVER_ON)).to("bean:serviceFlowService?method=brokerServerOn")
                .when(header("event").isEqualTo(ServiceFlowConstant.BROKER_SERVER_OFF)).to("bean:serviceFlowService?method=brokerServerOff")
                .endChoice();
    }
}

src/main/java/com/yihu/hos/arbiter/routers/ShellRouter.java

@@ -0,0 +1,39 @@
package com.yihu.hos.arbiter.routers;
import com.yihu.hos.arbiter.configuration.ActivemqConfiguration;
import com.yihu.hos.web.framework.constant.SSHConstant;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.model.ModelCamelContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.jms.ConnectionFactory;
/**
 * 中心shell命令消息
 * @author HZY
 * @vsrsion 1.0
 * Created at 2017/1/5.
 */
@Component
public class ShellRouter extends RouteBuilder {
    @Autowired
    private ActivemqConfiguration activemqConfiguration;
    @Override
    public void configure() throws Exception {
        ModelCamelContext context = this.getContext();
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                activemqConfiguration.getUser(), activemqConfiguration.getPassword(), activemqConfiguration.getBrokerURL());
        // Note we can explicit name the component
        context.addComponent(SSHConstant.CAMEL_COMPONENT, JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
        from(SSHConstant.CAMEL_ENDPOINT)
                .choice()
                .when(header("tenant").isNotNull()).to("bean:linuxShellService?method=handleServiceFlow")
                .when(header("event").isEqualTo(SSHConstant.ARBITER_SHELL_SEND)).to("bean:linuxShellService?method=sendShell")
                .endChoice();
    }
}

src/main/java/com/yihu/hos/arbiter/services/ArbiterServerService.java

@@ -0,0 +1,54 @@
package com.yihu.hos.arbiter.services;
import com.yihu.hos.arbiter.configuration.ArbiterServerConfiguration;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import com.yihu.hos.web.framework.model.bo.ArbiterServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;
/**
 * @author Airhead
 * @since 2016/12/9.
 */
@Service
public class ArbiterServerService {
    private static final Logger logger = LoggerFactory.getLogger(ArbiterServerService.class);
    private MongoOperations mongoOperations;
    private ArbiterServerConfiguration configuration;
    public ArbiterServerService() {
    }
    @Autowired
    public void setMongoOperations(MongoOperations mongoOperations){
        this.mongoOperations = mongoOperations;
    }
    @Autowired
    public void setConfiguration(ArbiterServerConfiguration configuration){
        this.configuration = configuration;
    }
    public void save(ArbiterServer arbiterServer) {
        Query query = new Query();
        query.addCriteria(Criteria.where("tenant").is(arbiterServer.getTenant()));
        Update update = new Update();
        update.set("tenant", arbiterServer.getTenant());
        update.set("url", arbiterServer.getUrl());
        mongoOperations.upsert(query, update, ArbiterServer.class);
    }
    public ArbiterServer get(String tenant) {
        Query query = new Query();
        query.addCriteria(Criteria.where("tenant").is(tenant));
        return mongoOperations.findOne(query, ArbiterServer.class);
    }
}

src/main/java/com/yihu/hos/arbiter/services/BrokerServerService.java

@@ -0,0 +1,172 @@
package com.yihu.hos.arbiter.services;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import com.yihu.hos.web.framework.constant.ServiceFlowConstant;
import com.yihu.hos.web.framework.model.bo.BrokerServer;
import com.yihu.hos.web.framework.model.bo.ServiceFlow;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;
import java.util.*;
/**
 * @created Airhead 2016/7/27.
 */
@Service("brokerServerService")
public class BrokerServerService {
    private static final Logger logger = LoggerFactory.getLogger(BrokerServerService.class);
    @Autowired
    private MongoOperations mongoOperations;
    @Autowired
    private CamelContext camelContext;
    @Autowired
    private ProducerTemplate producerTemplate;
    @Autowired
    private ObjectMapper objectMapper;
    /**
     * 暂不提供动态均衡算法，只是随机返回一个。
     *
     * @return
     */
    public BrokerServer get() {
        Query query = new Query();
        query.addCriteria(Criteria.where("enable").is(true));
        query.with(new Sort(new Sort.Order(Sort.Direction.DESC, "updateTime")));
        return mongoOperations.findOne(query, BrokerServer.class);
    }
    public List<BrokerServer> get(boolean one) {
        if (one) {
            BrokerServer brokerServer = get();
            List<BrokerServer> brokerServers = new ArrayList<>();
            brokerServers.add(brokerServer);
            return brokerServers;
        }
        Query query = new Query();
        query.addCriteria(Criteria.where("enable").is(true));
        query.with(new Sort(new Sort.Order(Sort.Direction.DESC, "updateTime")));
        return mongoOperations.find(query, BrokerServer.class);
    }
    public List<BrokerServer> getBrokerList(String routeCode) {
        Query query = new Query();
        query.addCriteria(Criteria.where("serviceFlows.routeCode").is(routeCode));
        return mongoOperations.find(query, BrokerServer.class);
    }
    public void addServiceFlow(BrokerServer brokerServer, ServiceFlow serviceFlow) {
        BrokerServer.ServiceFlow flow = new BrokerServer.ServiceFlow();
        flow.setFlowId(serviceFlow.getId());
        flow.setRouteCode(serviceFlow.getRouteCode());
        flow.setType(serviceFlow.getFlowType());
        brokerServer.addServiceFlow(flow);
        Query query = new Query();
        query.addCriteria(Criteria.where("hostName").is(brokerServer.getHostName()));
        query.addCriteria(Criteria.where("hostAddress").is(brokerServer.getHostAddress()));
        query.addCriteria(Criteria.where("port").is(brokerServer.getPort()));
        Update update = new Update();
        update.set("hostName", brokerServer.getHostName());
        update.set("hostAddress", brokerServer.getHostAddress());
        update.set("port", brokerServer.getPort());
        update.set("updateTime", brokerServer.getUpdateTime());
        update.set("enable", brokerServer.isEnable());
        update.addToSet("serviceFlows", serviceFlow);
        mongoOperations.upsert(query, update, BrokerServer.class);
    }
    public void removeServiceFlow(BrokerServer brokerServer, ServiceFlow serviceFlow) {
        Query query = new Query();
        query.addCriteria(Criteria.where("hostName").is(brokerServer.getHostName()));
        query.addCriteria(Criteria.where("hostAddress").is(brokerServer.getHostAddress()));
        query.addCriteria(Criteria.where("port").is(brokerServer.getPort()));
        Update update = new Update();
        update.set("hostName", brokerServer.getHostName());
        update.set("hostAddress", brokerServer.getHostAddress());
        update.set("port", brokerServer.getPort());
        update.set("updateTime", brokerServer.getUpdateTime());
        update.set("enable", brokerServer.isEnable());
        update.pull("serviceFlows", serviceFlow);
        mongoOperations.upsert(query, update, BrokerServer.class);
    }
    public void login(BrokerServer brokerServer) {
        Query query = new Query();
        query.addCriteria(Criteria.where("hostName").is(brokerServer.getHostName()));
        query.addCriteria(Criteria.where("hostAddress").is(brokerServer.getHostAddress()));
        query.addCriteria(Criteria.where("port").is(brokerServer.getPort()));
        Update update = new Update();
        update.set("tenant", brokerServer.getTenant());
        update.set("hostName", brokerServer.getHostName());
        update.set("hostAddress", brokerServer.getHostAddress());
        update.set("port", brokerServer.getPort());
        Date updateTime = brokerServer.getUpdateTime() == null ? new Date() : brokerServer.getUpdateTime();
        update.set("updateTime", updateTime);
        update.set("createTime", new Date());
        update.set("enable", brokerServer.isEnable());
        mongoOperations.upsert(query, update, BrokerServer.class);
        try {
            String msg = objectMapper.writeValueAsString(brokerServer);
            ProducerTemplate producerTemplate = createProducerTemplate();
            Map<String, Object> header = new HashMap<>();
            header.put("event", ServiceFlowConstant.BROKER_SERVER_ON);
            producerTemplate.sendBodyAndHeaders(ServiceFlowConstant.CAMEL_ENDPOINT, msg, header);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    public void logout(BrokerServer brokerServer) {
        Query query = new Query();
        query.addCriteria(Criteria.where("hostName").is(brokerServer.getHostName()));
        query.addCriteria(Criteria.where("hostAddress").is(brokerServer.getHostAddress()));
        query.addCriteria(Criteria.where("port").is(brokerServer.getPort()));
        BrokerServer server = mongoOperations.findOne(query, BrokerServer.class);
        mongoOperations.remove(server);
    }
    public void heartbeat(BrokerServer brokerServer) {
        Query query = new Query();
        query.addCriteria(Criteria.where("hostName").is(brokerServer.getHostName()));
        query.addCriteria(Criteria.where("hostAddress").is(brokerServer.getHostAddress()));
        query.addCriteria(Criteria.where("port").is(brokerServer.getPort()));
        BrokerServer server = mongoOperations.findOne(query, BrokerServer.class);
        if (server != null) {
            server.setUpdateTime(new Date());
            mongoOperations.save(server);
        }
    }
    private ProducerTemplate createProducerTemplate() {
        if (producerTemplate == null) {
            producerTemplate = camelContext.createProducerTemplate();
        }
        return producerTemplate;
    }
}

src/main/java/com/yihu/hos/arbiter/services/EndpointService.java

@@ -0,0 +1,118 @@
package com.yihu.hos.arbiter.services;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.core.datatype.StringUtil;
import com.yihu.hos.core.http.HTTPResponse;
import com.yihu.hos.core.http.HttpClientKit;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import com.yihu.hos.web.framework.model.bo.Endpoint;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.List;
/**
 * @created Airhead 2016/7/27.
 */
@Service("endpointService")
public class EndpointService {
    private static final Logger logger = LoggerFactory.getLogger(BrokerServerService.class);
    @Autowired
    private MongoOperations mongoOperations;
    @Autowired
    private ObjectMapper objectMapper;
    public void save(Endpoint endpoint) {
        Query query = new Query();
        query.addCriteria(Criteria.where("code").is(endpoint.getCode()));
        Update update = new Update();
        update.set("code", endpoint.getCode());
        update.set("createTime", endpoint.getCreateTime());
        update.set("endpoint", endpoint.getEndpoint());
        update.set("event", endpoint.getEvent());
        update.set("healthCheckType", endpoint.getHealthCheckType());
        update.set("healthCheckURL", endpoint.getHealthCheckURL());
        update.set("metricsType", endpoint.getMetricsType());
        update.set("metricsURL", endpoint.getMetricsURL());
        update.set("name", endpoint.getName());
        update.set("updateTime", endpoint.getUpdateTime());
        update.set("active", endpoint.isActive());
        mongoOperations.upsert(query, update, Endpoint.class);
    }
    public void delete(Endpoint endpoint) {
        mongoOperations.remove(endpoint);
    }
    public Endpoint get(String serviceName) {
        return mongoOperations.findOne(new Query(), Endpoint.class);
    }
    public List<Endpoint> getEndpointList() {
        return mongoOperations.findAll(Endpoint.class);
    }
    public void check() {
        List<Endpoint> endpointList = getEndpointList();
        endpointList.forEach(this::remoteCheck);
    }
    public void addEndpoint(String msg) {
        try {
            Endpoint endpoint = objectMapper.readValue(msg, Endpoint.class);
            endpoint.setActive(true);
            this.save(endpoint);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public void deleteEndpoint(String msg) {
        try {
            Endpoint endpoint = objectMapper.readValue(msg, Endpoint.class);
            this.delete(endpoint);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public void offServer(String msg) {
        try {
            Endpoint endpoint = objectMapper.readValue(msg, Endpoint.class);
            endpoint.setActive(false);
            this.save(endpoint);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * 调用实际的检查地址
     *
     * @param endpoint 通讯点信息
     */
    private void remoteCheck(Endpoint endpoint) {
        String url = endpoint.getHealthCheckURL();
        if (StringUtil.isEmpty(url)) {
            return;
        }
        HTTPResponse response = HttpClientKit.get(url);
        if (response.getStatusCode() == 200) {
            return;
        }
        endpoint.setActive(false);
        this.save(endpoint);
    }
}

src/main/java/com/yihu/hos/arbiter/services/MycatService.java

@@ -0,0 +1,89 @@
package com.yihu.hos.arbiter.services;
import com.yihu.hos.core.http.HTTPResponse;
import com.yihu.hos.core.http.HttpClientKit;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import com.yihu.hos.web.framework.model.bo.BrokerServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
 * @author HZY
 * @vsrsion 1.0
 * Created at 2017/1/5.
 */
@Service("mycatService")
public class MycatService {
    private static final Logger logger = LoggerFactory.getLogger(MycatService.class);
    @Autowired
    private BrokerServerService brokerServerService;
    /**
     * 发送shell命令请求到broker
     *
     * @param msg 命令消息内容
     */
    public void updateMycat(String msg) {
        try {
            List<BrokerServer> brokerServerList = brokerServerService.get(false);
            for (BrokerServer broker : brokerServerList) {
                String result = sendMessage(broker, "post", "/esb/serviceMycat/update", msg);
                if (result == null) {
                    logger.error("sendMessage to broker start failed, broker:" + broker.getURL() + ", msg:" + msg);
                    continue;
                } else {
                    logger.debug("发送mycat修改请求到broker成功");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    private String sendMessage(BrokerServer brokerServer, String method, String path, String msg) {
        if (brokerServer == null) {
            return null;
        }
        switch (method) {
            case "post": {
                HTTPResponse response = HttpClientKit.post(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    return body;
                }
                return null;
            }
            case "put": {
                HTTPResponse response = HttpClientKit.put(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    logger.debug(body);
                    return body;
                }
                return null;
            }
            case "delete": {
                HTTPResponse response = HttpClientKit.delete(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    logger.debug(body);
                    return body;
                }
                return null;
            }
            default:
                break;
        }
        return null;
    }
}

src/main/java/com/yihu/hos/arbiter/services/ProxyService.java

@@ -0,0 +1,213 @@
package com.yihu.hos.arbiter.services;
import com.yihu.hos.arbiter.configuration.ArbiterServerConfiguration;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import com.yihu.hos.web.framework.constant.EndPointConstant;
import com.yihu.hos.web.framework.constant.MycatConstant;
import com.yihu.hos.web.framework.constant.SSHConstant;
import com.yihu.hos.web.framework.constant.ServiceFlowConstant;
import io.zbus.mq.*;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
/**
 * @author Airhead
 * @since 2016/12/22.
 */
@Service
public class ProxyService {
    private static final Logger logger = LoggerFactory.getLogger(ProxyService.class);
    private CamelContext camelContext;
    private ArbiterServerConfiguration configuration;
    private Broker broker;
    @Autowired
    public void setBroker(Broker broker) {
        this.broker = broker;
    }
    @Autowired
    public void setCamelContext(CamelContext camelContext) {
        this.camelContext = camelContext;
    }
    @Autowired
    public void setConfiguration(ArbiterServerConfiguration configuration) {
        this.configuration = configuration;
    }
    public void start() {
        new ServiceFlow().start();
        new Mycat().start();
        new Shell().start();
        new EndPoint().start();
    }
    public class ServiceFlow {
        private Consumer consumer;
        public void start() {
            if (broker == null) {
                logger.error("broker is null");
                return;
            }
            try {
                ConsumerConfig config = new ConsumerConfig(broker);
                config.setTopic(ServiceFlowConstant.ZBUS_MQ + "@" + configuration.getTenant());
                ConsumeGroup consumeGroup = ConsumeGroup.createTempBroadcastGroup();
                config.setConsumeGroup(consumeGroup);
                config.setMessageHandler((message, mqClient) -> {
                    Map<String, Object> header = new HashMap<>();
                    header.put("event", message.getHeader("event"));
                    ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
                    producerTemplate.sendBodyAndHeaders(ServiceFlowConstant.CAMEL_ENDPOINT, message.getBodyString(), header);
                });
                consumer = new Consumer(config);
                consumer.start();
            } catch (IOException e) {
                e.printStackTrace();
                logger.error(e.getMessage());
            }
        }
        @Override
        protected void finalize() throws Throwable {
            if (consumer != null) {
                consumer.close();
            }
            super.finalize();
        }
    }
    public class Mycat {
        private Consumer consumer;
        public void start() {
            if (broker == null) {
                logger.error("broker is null");
                return;
            }
            try {
                ConsumerConfig config = new ConsumerConfig(broker);
                config.setTopic(MycatConstant.ZBUS_MQ + "@" + configuration.getMycatName());
                ConsumeGroup consumeGroup = ConsumeGroup.createTempBroadcastGroup();
                config.setConsumeGroup(consumeGroup);
                config.setMessageHandler((message, mqClient) -> {
                    Map<String, Object> header = new HashMap<>();
                    header.put("event", message.getHeader("event"));
                    ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
                    producerTemplate.sendBodyAndHeaders(MycatConstant.CAMEL_ENDPOINT, message.getBodyString(), header);
                });
                consumer = new Consumer(config);
                consumer.start();
            } catch (IOException e) {
                e.printStackTrace();
                logger.error(e.getMessage());
            }
        }
        @Override
        protected void finalize() throws Throwable {
            if (consumer != null) {
                consumer.close();
            }
            super.finalize();
        }
    }
    public class Shell {
        private Consumer consumer;
        public void start() {
            if (broker == null) {
                logger.error("broker is null");
                return;
            }
            try {
                ConsumerConfig config = new ConsumerConfig(broker);
                config.setTopic(SSHConstant.SHELL_REQUEST + "@" + configuration.getTenant());
                ConsumeGroup consumeGroup = ConsumeGroup.createTempBroadcastGroup();
                config.setConsumeGroup(consumeGroup);
                config.setMessageHandler((message, mqClient) -> {
                    Map<String, Object> header = new HashMap<>();
                    header.put("event", message.getHeader("event"));
                    ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
                    producerTemplate.sendBodyAndHeaders(SSHConstant.CAMEL_ENDPOINT, message.getBodyString(), header);
                });
                consumer = new Consumer(config);
                consumer.start();
            } catch (IOException e) {
                e.printStackTrace();
                logger.error(e.getMessage());
            }
        }
        @Override
        protected void finalize() throws Throwable {
            if (consumer != null) {
                consumer.close();
            }
            super.finalize();
        }
    }
    public class EndPoint {
        private Consumer consumer;
        public void start() {
            if (broker == null) {
                logger.error("broker is null");
                return;
            }
            try {
                ConsumerConfig config = new ConsumerConfig(broker);
                config.setTopic(EndPointConstant.ZBUS_MQ + "@" + configuration.getTenant());
                ConsumeGroup consumeGroup = ConsumeGroup.createTempBroadcastGroup();
                config.setConsumeGroup(consumeGroup);
                config.setMessageHandler((message, mqClient) -> {
                    Map<String, Object> header = new HashMap<>();
                    header.put("event", message.getHeader("event"));
                    ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
                    producerTemplate.sendBodyAndHeaders(EndPointConstant.CAMEL_ENDPOINT, message.getBodyString(), header);
                });
                consumer = new Consumer(config);
                consumer.start();
            } catch (IOException e) {
                e.printStackTrace();
                logger.error(e.getMessage());
            }
        }
        @Override
        protected void finalize() throws Throwable {
            if (consumer != null) {
                consumer.close();
            }
            super.finalize();
        }
    }
}

src/main/java/com/yihu/hos/arbiter/services/ServiceFlowService.java

@@ -0,0 +1,408 @@
package com.yihu.hos.arbiter.services;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.core.http.HTTPResponse;
import com.yihu.hos.core.http.HttpClientKit;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import com.yihu.hos.web.framework.constant.ServiceFlowConstant;
import com.yihu.hos.web.framework.model.bo.BrokerServer;
import com.yihu.hos.web.framework.model.bo.ServiceFlow;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.List;
/**
 * Broker原则上具有等同性，这样Arbiter无论选择了哪个Broker能提供的服务都是一样的。
 * 但是因为Broker上还是会运行一些定时的采集任务，这些采集任务如果是多台Broker运行的话，可能会引起数据重复问题。
 * 所以在事件触发时需要做一些策略的调整：
 * 1.实时任务，通知所有的Broker进行更新路由
 * 2.采集任务，只通知其中的一台进行更新路由
 *
 * @created Airhead 2016/8/16.
 */
@Service("serviceFlowService")
public class ServiceFlowService {
    private static final Logger logger = LoggerFactory.getLogger(BrokerServerService.class);
    @Autowired
    private MongoOperations mongoOperations;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private BrokerServerService brokerServerService;
    public ServiceFlow save(ServiceFlow serviceFlow) {
        if (serviceFlow == null) {
            logger.error("ServiceFlow is null");
            return null;
        }
        Query query = new Query();
        query.addCriteria(Criteria.where("routeCode").is(serviceFlow.getRouteCode()));
        Update update = new Update();
        update.set("routeCode", serviceFlow.getRouteCode());
        update.set("updated", serviceFlow.getUpdated());
        update.set("flowType", serviceFlow.getFlowType());
        update.set("tenant", serviceFlow.getTenant());
        //删除旧记录
        update.set("handleFiles", serviceFlow.getHandleFiles());
//        update1.set("handleFiles.$.className", handleFile.getClassName());
//        Query query1 = Query.query(new Criteria().andOperator(Criteria.where("className").is( handleFile.getClassName()),Criteria.where("handleFiles").elemMatch(Criteria.where("className").is( handleFile.getClassName()))));
//        mongoOperations.updateFirst(query1, update1, ServiceFlow.class);
        mongoOperations.upsert(query, update, ServiceFlow.class);
        return mongoOperations.findOne(query, ServiceFlow.class);
    }
    public String get(String serviceName) {
        return null;
    }
    public void delete(ServiceFlow serviceFlow) {
        mongoOperations.remove(serviceFlow);
    }
    public List<ServiceFlow> getAll() {
        return mongoOperations.findAll(ServiceFlow.class);
    }
    /**
     * admin发过来的服务流程启动事件处理。
     *
     * @param msg serviceFlow
     */
    public void serviceFlowStarted(String msg) {
        try {
            ServiceFlow serviceFlow = getServiceFlow(msg);
            serviceFlow = this.save(serviceFlow);
            boolean one = ServiceFlowConstant.JAVA.equals(serviceFlow.getFlowType());//java类型为采集任务
            if (one && isStarted(serviceFlow)) {
                return;
            }
            List<BrokerServer> brokerServerList;
            brokerServerList = brokerServerService.get(one);
            for (BrokerServer broker : brokerServerList) {
                boolean result = sendMessage(broker, "post", "/esb/serviceFlow/serverServiceFlow", msg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + broker.getURL() + ", msg:" + msg);
                    continue;
                }
                brokerServerService.addServiceFlow(broker, serviceFlow);
            }
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    public void serviceFlowStopped(String msg) {
        try {
            ServiceFlow serviceFlow = getServiceFlow(msg);
            this.delete(serviceFlow);
            List<BrokerServer> brokerServerList;
            brokerServerList = brokerServerService.get(false);
            for (BrokerServer broker : brokerServerList) {
                HTTPResponse response = HttpClientKit.post(broker.getURL() + "", msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    logger.debug(body);
                }
                boolean result = sendMessage(broker, "post", "/esb/serviceFlow/stop", msg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + broker.getURL() + ", msg:" + msg);
                    continue;
                }
                brokerServerService.removeServiceFlow(broker, serviceFlow);
            }
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    public void serviceFlowAdd(String msg) {
        try {
            ServiceFlow serviceFlow = getServiceFlow(msg);
            serviceFlow = this.save(serviceFlow);
            System.out.println("流程添加serviceFlowAdd开始！");
            boolean one = ServiceFlowConstant.JAVA.equals(serviceFlow.getFlowType());//java类型为采集任务
            if (one) {
                BrokerServer brokerServer = brokerServerService.get();
                boolean result = sendMessage(brokerServer, "post", "/esb/serviceFlow", msg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + brokerServer.getURL() + ", msg:" + msg);
                    return;
                }
                brokerServerService.addServiceFlow(brokerServer, serviceFlow);
                return;
            }
            List<BrokerServer> brokerServerList;
            brokerServerList = brokerServerService.get(one);
            for (BrokerServer broker : brokerServerList) {
                boolean result = sendMessage(broker, "post", "/esb/serviceFlow", msg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + broker.getURL() + ", msg:" + msg);
                    continue;
                }
                brokerServerService.addServiceFlow(broker, serviceFlow);
            }
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    public void serviceFlowModifyAdd(String msg) {
        try {
            System.out.println("arbiter'route  serviceFlowModifyAdd ================ 1 ");
            ServiceFlow serviceFlow = getServiceFlow(msg);
            serviceFlow = this.save(serviceFlow);
            boolean one = ServiceFlowConstant.JAVA.equals(serviceFlow.getFlowType());//java类型为采集任务
            if (one) {
                System.out.println("arbiter'route  is java ================ 2, cdoe: " + serviceFlow.getRouteCode());
                List<BrokerServer> brokerList = brokerServerService.getBrokerList(serviceFlow.getRouteCode());
                if (brokerList == null || brokerList.size() == 0) {
                    logger.error("service flow stopped unexpected.");
                    return;
                }
                boolean result = sendMessage(brokerList.get(0), "post", "/esb/serviceFlow/add", msg);
                if (!result) {
                    logger.error("serviceFlowModifyAdd11 sendMessage to broker start failed, broker:" + brokerList.get(0).getURL() + ", msg:" + msg);
                    return;
                }
                brokerServerService.addServiceFlow(brokerList.get(0), serviceFlow);
                return;
            }
            List<BrokerServer> brokerServerList;
            brokerServerList = brokerServerService.get(one);
            for (BrokerServer broker : brokerServerList) {
                boolean result = sendMessage(broker, "post", "/esb/serviceFlow/add", msg);
                if (!result) {
                    logger.error("serviceFlowModifyAdd22 sendMessage to broker start failed, broker:" + broker.getURL() + ", msg:" + msg);
                    continue;
                } else {
                    System.out.println("arbiter'route susscess ================ 3");
                }
                brokerServerService.addServiceFlow(broker, serviceFlow);
            }
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    public void serviceFlowModifyReduce(String msg) {
        try {
            ServiceFlow serviceFlow = getServiceFlow(msg);
            serviceFlow = this.save(serviceFlow);
            boolean one = ServiceFlowConstant.JAVA.equals(serviceFlow.getFlowType());//java类型为采集任务
            if (one) {
                List<BrokerServer> brokerList = brokerServerService.getBrokerList(serviceFlow.getRouteCode());
                if (brokerList == null || brokerList.size() == 0) {
                    logger.error("service flow stopped unexpected.");
                    return;
                }
                boolean result = sendMessage(brokerList.get(0), "put", "/esb/serviceFlow/reduce", msg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + brokerList.get(0).getURL() + ", msg:" + msg);
                    return;
                }
                brokerServerService.addServiceFlow(brokerList.get(0), serviceFlow);
                return;
            }
            List<BrokerServer> brokerServerList;
            brokerServerList = brokerServerService.get(one);
            for (BrokerServer broker : brokerServerList) {
                boolean result = sendMessage(broker, "put", "/esb/serviceFlow/reduce", msg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + broker.getURL() + ", msg:" + msg);
                    continue;
                }
                brokerServerService.addServiceFlow(broker, serviceFlow);
            }
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    public void serviceFlowDelete(String msg) {
        try {
            ServiceFlow serviceFlow = getServiceFlow(msg);
            this.delete(serviceFlow);
            boolean one = ServiceFlowConstant.JAVA.equals(serviceFlow.getFlowType());//java类型为采集任务
            if (one) {
                List<BrokerServer> brokerList = brokerServerService.getBrokerList(serviceFlow.getRouteCode());
                if (brokerList == null || brokerList.size() == 0) {
                    logger.error("service flow stopped unexpected.");
                    return;
                }
                boolean result = sendMessage(brokerList.get(0), "delete", "/esb/serviceFlow", msg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + brokerList.get(0).getURL() + ", msg:" + msg);
                    return;
                }
                brokerServerService.removeServiceFlow(brokerList.get(0), serviceFlow);
                return;
            }
            List<BrokerServer> brokerServerList;
            brokerServerList = brokerServerService.get(one);
            for (BrokerServer broker : brokerServerList) {
                boolean result = sendMessage(broker, "delete", "/esb/serviceFlow", msg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + broker.getURL() + ", msg:" + msg);
                    continue;
                }
                brokerServerService.removeServiceFlow(broker, serviceFlow);
            }
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    public void brokerServerOn(String msg) {
        List<ServiceFlow> serviceFlowList = getAll();
        serviceFlowList.forEach(serviceFlow -> {
            try {
                serviceFlow = this.save(serviceFlow);
                boolean one = ServiceFlowConstant.JAVA.equals(serviceFlow.getFlowType());//java类型为采集任务
                if (one && isStarted(serviceFlow)) {
                    return;
                }
                BrokerServer brokerServer = objectMapper.readValue(msg, BrokerServer.class);
                String serviceFlowMsg = objectMapper.writeValueAsString(serviceFlow);
                boolean result = sendMessage(brokerServer, "post", "/esb/serviceFlow/start", serviceFlowMsg);
                if (!result) {
                    logger.error("sendMessage to broker start failed, broker:" + brokerServer.getURL() + ", msg:" + serviceFlowMsg);
                    return;
                }
                brokerServerService.addServiceFlow(brokerServer, serviceFlow);
            } catch (IOException e) {
                e.printStackTrace();
                logger.error(e.getMessage());
            }
        });
    }
    public void brokerServerOff(String msg) {
        //可以不用处理。
    }
    private boolean sendMessage(BrokerServer brokerServer, String method, String path, String msg) {
        if (brokerServer == null) {
            return false;
        }
        switch (method) {
            case "post": {
                HTTPResponse response = HttpClientKit.post(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    logger.debug(body);
                    return true;
                }
                logger.error("post error,url: " + brokerServer.getURL() + path);
                return false;
            }
            case "put": {
                HTTPResponse response = HttpClientKit.put(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    logger.debug(body);
                    return true;
                }
                logger.error("put error,url: " + brokerServer.getURL() + path);
                return false;
            }
            case "delete": {
                HTTPResponse response = HttpClientKit.delete(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    logger.debug(body);
                    return true;
                }
                logger.error("delete error,url: " + brokerServer.getURL() + path);
                return false;
            }
            default:
                break;
        }
        return false;
    }
    private ServiceFlow getServiceFlow(String msg) throws IOException {
        return objectMapper.readValue(msg, ServiceFlow.class);
    }
    private boolean isStarted(ServiceFlow serviceFlow) {
        List<BrokerServer> brokerList = brokerServerService.getBrokerList(serviceFlow.getRouteCode());
        if (brokerList != null && brokerList.size() != 0) {
            logger.debug("service flow is already started on the broker");
            return true;
        }
        return false;
    }
}

src/main/java/com/yihu/hos/arbiter/services/ShellService.java

@@ -0,0 +1,124 @@
package com.yihu.hos.arbiter.services;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.core.http.HTTPResponse;
import com.yihu.hos.core.http.HttpClientKit;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import com.yihu.hos.web.framework.constant.SSHConstant;
import com.yihu.hos.web.framework.model.bo.BrokerServer;
import com.yihu.hos.web.framework.model.bo.ServiceShell;
import io.zbus.mq.Broker;
import io.zbus.mq.Message;
import io.zbus.mq.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
 * @author HZY
 * @vsrsion 1.0
 * Created at 2017/1/5.
 */
@Service("linuxShellService")
public class ShellService {
    private static final Logger logger = LoggerFactory.getLogger(ShellService.class);
    @Autowired
    private BrokerServerService brokerServerService;
    private Broker broker;
    @Autowired
    private ObjectMapper objectMapper;
    /**
     * 发送shell命令请求到broker
     *
     * @param msg 命令消息内容
     */
    public void sendShell(String msg) {
        try {
            List<BrokerServer> brokerServerList = brokerServerService.get(false);
            for (BrokerServer broker : brokerServerList) {
                String result = sendMessage(broker, "post", "/esb/serviceShell/send", msg);
                if (result == null) {
                    logger.error("sendMessage to broker start failed, broker:" + broker.getURL() + ", msg:" + msg);
                    continue;
                } else {
                    logger.debug("发送shell请求到broker成功");
                    // shell执行成功，见执行结果返回到中心zbus显示。
                    ServiceShell serviceShell = objectMapper.readValue(msg, ServiceShell.class);
                    String topic = SSHConstant.SHELL_RESPONSE + "@" + serviceShell.getTenant();
                    Producer producer = new Producer(this.broker);
                    producer.declareTopic(topic);
                    Message message = new Message();
                    message.setTopic(topic);
                    message.setHeader("event", serviceShell.getType());
                    message.setHeader("tenant", serviceShell.getTenant());
                    message.setMethod("POST");
                    message.setBody(result);
                    Message res = producer.publish(message);
                    if (logger.isInfoEnabled()) {
                        logger.info(res.toString());
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
    @Autowired
    public void setBroker(Broker broker) {
        this.broker = broker;
    }
    private String sendMessage(BrokerServer brokerServer, String method, String path, String msg) {
        if (brokerServer == null) {
            return null;
        }
        switch (method) {
            case "post": {
                HTTPResponse response = HttpClientKit.post(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    return body;
                }
                return null;
            }
            case "put": {
                HTTPResponse response = HttpClientKit.put(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    logger.debug(body);
                    return body;
                }
                return null;
            }
            case "delete": {
                HTTPResponse response = HttpClientKit.delete(brokerServer.getURL() + path, msg);
                if (response.getStatusCode() == 200) {
                    String body = response.getBody();
                    logger.debug(body);
                    return body;
                }
                return null;
            }
            default:
                break;
        }
        return null;
    }
}

src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -0,0 +1,19 @@
{
  "properties": [
    {
      "name": "arbiter.timer.period",
      "type": "java.lang.String",
      "description": "Description for arbiter.timer.period."
    },
    {
      "name": "arbiter.central.url",
      "type": "java.lang.String",
      "description": "Description for arbiter.central.url."
    },
    {
      "name": "arbiter.tenant.name",
      "type": "java.lang.String",
      "description": "Description for arbiter.tenant.name."
    }
  ]
}

src/main/resources/application.yml

@@ -0,0 +1,78 @@
server:
  port: 10135
---
spring:
  profiles: dev
  data:
    mongodb:
      host: 172.19.103.42
      port: 27017
      username: esb
      password: esb
      authenticationDatabase: admin
      database: runtime
  activemq:
    broker-url: tcp://172.19.103.57:61616
    user: admin
    password: admin
arbiter:
  timer:
    period: 10000
  central:
    url: 172.19.103.57:9020
  tenant:
    name: jkzl
  mycat:
    name: mycat172.19.103.42
---
spring:
  profiles: test
  data:
    mongodb:
      host: 172.19.103.57
      port: 27017
      username: esb
      password: esb
      authenticationDatabase: admin
      database: runtime
  activemq:
    broker-url: tcp://172.19.103.57:61616
    user: admin
    password: admin
arbiter:
  timer:
    period: 10000
  central:
    url: 172.19.103.58:9020
  tenant:
    name: yichang
  mycat:
    name: mycat172.19.103.57
---
spring:
  profiles: prod
  data:
    mongodb:
      host: 192.168.50.26
      port: 20000
      username: jkzlesb
      password: fgu$^ezl
      authenticationDatabase: admin
      database: runtime
  activemq:
    broker-url: tcp://192.168.200.60:61616
    user: admin
    password: admin
arbiter:
  timer:
      period: 10000
  central:
    url: 192.168.200.62:9020
  tenant:
    name: jkzl
  mycat:
    name: mycat10.176.97.14

src/main/resources/bootstrap.yml

@@ -0,0 +1,38 @@
spring:
  application:
    name: hos-arbiter
security:
  basic:
    enabled: false
---
spring:
  profiles: dev
  cloud:
    config:
      username: user
      password: configuration
      uri: ${spring.config.uri:http://172.19.103.73:1221}
      label: ${spring.config.label:dev}
---
spring:
  profiles: test
  cloud:
    config:
      username: user
      password: configuration
      uri: ${spring.config.uri:http://172.19.103.73:1221}
      label: ${spring.config.label:dev}
---
spring:
  profiles: prod
  cloud:
    config:
      username: user
      password: configuration
      uri: ${spring.config.uri}
      label: ${spring.config.label}

src/main/resources/log4j2.xml

@@ -0,0 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration status ="ALL"  monitorInterval="1800">
    <Appenders>
        <Console name="Console" target="SYSTEM_OUT">
            <PatternLayout charset="UTF-8" pattern="%d{HH:mm:ss}[%-5p][%t][%c]: %m%n"/>
        </Console>
        <RollingFile name="RollingFile" filename="/usr/local/esb/log/CrunchifyTest.log"
                     filepattern="/usr/local/esb/log/rolling/%d{yyyyMMddHHmmss}-fargo.log">
            <PatternLayout charset="UTF-8" pattern="%d{yyyy-MM-dd HH:mm:ss} [%t] %-5p %c{1}:%L - %msg%n"/>
            <Policies>
                <SizeBasedTriggeringPolicy size="100 MB"/>
            </Policies>
            <DefaultRolloverStrategy max="20"/>
        </RollingFile>
    </Appenders>
    <Loggers>
        <Root level="debug">
            <AppenderRef ref="Console"/>
            <AppenderRef ref="RollingFile" />
        </Root>
    </Loggers>
</configuration>

src/main/resources/logback.xml

@@ -0,0 +1,88 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration debug="false">
    <!-- 日志文件基础目录-->
    <property name="log_dir" value="/usr/local/sip/logs/" />
    <!-- 应用名称 -->
    <property name="APP_NAME" value="hos-arbiter" />
    <!--日志文件的保存路径,首先查找系统属性-Dlog.dir,如果存在就使用其；否则，在当前目录下创建名为logs目录做日志存放的目录 -->
    <property name="LOG_HOME" value="${log_dir:-logs}/${APP_NAME}" />
    <!-- 日志输出格式 -->
    <property name="ENCODER_PATTERN"
              value="%d{yyyy-MM-dd  HH:mm:ss.SSS} [%thread] %-5level %logger{80} - %msg%n" />
    <contextName>${APP_NAME}</contextName>
    <!-- 控制台日志：输出全部日志到控制台 -->
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <Pattern>${ENCODER_PATTERN}</Pattern>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>DEBUG</level>
        </filter>
    </appender>
    <!-- 文件日志：输出全部日志到文件 根据时间和文件大小滚动 -->
    <!--<appender name="FILE_SizeAndTime" class="ch.qos.logback.core.rolling.RollingFileAppender">-->
    <!--<file>${LOG_HOME}/output.log</file>-->
    <!--<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">-->
    <!--<fileNamePattern>${LOG_HOME}/output.%d{yyyy-MM-dd_HH_mm}.%i.log</fileNamePattern>-->
    <!--<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">-->
    <!--<maxFileSize>100KB</maxFileSize>-->
    <!--</timeBasedFileNamingAndTriggeringPolicy>-->
    <!--&lt;!&ndash; keep X days' worth of history &ndash;&gt;-->
    <!--<maxHistory>3</maxHistory>-->
    <!--<cleanHistoryOnStart>true</cleanHistoryOnStart>-->
    <!--</rollingPolicy>-->
    <!--<append>true</append>-->
    <!--<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">-->
    <!--<pattern>${ENCODER_PATTERN}</pattern>-->
    <!--<charset>utf-8</charset>-->
    <!--</encoder>-->
    <!--</appender>-->
    <!-- 文件日志：输出全部日志到文件 根据日期滚动 -->
    <appender name="FILE"  class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_HOME}/output.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${LOG_HOME}/output.%d{yyyy-MM-dd_HH}.log</fileNamePattern>
            <maxHistory>10</maxHistory>
        </rollingPolicy>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>${ENCODER_PATTERN}</pattern>
            <charset>utf-8</charset>
        </encoder>
    </appender>
    <!-- 错误日志：用于将错误日志输出到独立文件 -->
    <appender name="ERROR_FILE"
              class="ch.qos.logback.core.rolling.RollingFileAppender">
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${LOG_HOME}/error/error.%d{yyyy-MM-dd}.log</fileNamePattern>
            <maxHistory>7</maxHistory>
        </rollingPolicy>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>${ENCODER_PATTERN}</pattern>
            <charset>utf-8</charset>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>WARN</level>
        </filter>
    </appender>
    <logger name="org.hibernate" level="WARN"/>
    <logger name="org.springframework" level="WARN"/>
    <logger name="springfox.documentation" level="WARN"/>
    <root>
        <level value="DEBUG" />
        <appender-ref ref="STDOUT" />
        <appender-ref ref="FILE" />
        <appender-ref ref="ERROR_FILE" />
    </root>
</configuration>

src/test/java/com/yihu/hos/HosArbiterApplicationTests.java

@@ -0,0 +1,17 @@
package com.yihu.hos;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;
@RunWith(SpringJUnit4ClassRunner.class)
//@SpringApplicationConfiguration(classes = HosArbiterApplication.class)
@WebAppConfiguration
public class HosArbiterApplicationTests {
	@Test
	public void contextLoads() {
	}
}

src/test/java/com/yihu/hos/arbiter/services/ProxyServiceTest.java

@@ -0,0 +1,29 @@
package com.yihu.hos.arbiter.services;
import io.zbus.mq.Broker;
import io.zbus.mq.Message;
import io.zbus.mq.Producer;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
/**
 * @author Airhead
 * @since 2016/12/22.
 */
@SpringBootTest
public class ProxyServiceTest {
    @Autowired
    private Broker broker;
    @Test
    public void proxy() throws Exception {
        Producer producer = new Producer(broker);
        producer.declareTopic("RemoteControl");//确定为创建消息队列需要显示调用
        Message msg = new Message();
        msg.setTopic("RemoteControl");
        msg.setBody("hello world"); //消息体底层是byte[]
        msg = producer.publish(msg);
    }
}