kafka消息总线功能

 server:
 server:
-  port: 9030
+  port: 19031
 
 
 # spring 配置
 # spring 配置
 spring:
 spring:
@@ -22,9 +22,9 @@ spring:
   # redis设置
   # redis设置
   redis:
   redis:
     database: 0 # Redis 数据库索引（默认为 0）
     database: 0 # Redis 数据库索引（默认为 0）
-    host: 192.168.37.139 # Redis 服务器地址
+    host: 192.168.1.53 # Redis 服务器地址
     port: 6379 # Redis 服务器连接端口
     port: 6379 # Redis 服务器连接端口
-    password: 123456 # Redis 服务器连接密码（默认为空）
+    password:  # Redis 服务器连接密码（默认为空）
     lettuce:
     lettuce:
       pool:
       pool:
         max-active: 8 # 连接池最大连接数（使用负值表示没有限制） 默认 8
         max-active: 8 # 连接池最大连接数（使用负值表示没有限制） 默认 8
@@ -34,10 +34,10 @@ spring:
 
 
   # rabbitmq配置
   # rabbitmq配置
   rabbitmq:
   rabbitmq:
-    host: 192.168.37.139
+    host: 192.168.1.63
     port: 5672
     port: 5672
-    username: rabbit
-    password: 123456
+    username: mrz
+    password: vmvnv1v2
     virtual-host: /
     virtual-host: /
 
 
   # 多环境配置
   # 多环境配置
@@ -50,18 +50,12 @@ spring:
     time-zone: GMT+8
     time-zone: GMT+8
     date-format: yyyy-MM-dd HH:mm:ss
     date-format: yyyy-MM-dd HH:mm:ss
 
 
-  # zipkin 链路追踪配置
-  zipkin:
-    base-url: http://192.168.1.63:9411
-    sleuth:
-      sampler:
-        probability: 1
 
 
 # mybatis 配置
 # mybatis 配置
 mybatis:
 mybatis:
   mapper-locations: classpath:mapper/*.xml
   mapper-locations: classpath:mapper/*.xml
   type-aliases-package: com.sunyo.wlpt.message.bus.service.domain
   type-aliases-package: com.sunyo.wlpt.message.bus.service.domain
-
+debug: false
 # 日志配置
 # 日志配置
 logging:
 logging:
   config: config/logback-dev.xml
   config: config/logback-dev.xml

@@ -19,8 +19,9 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
         <spring-cloud.version>Hoxton.RELEASE</spring-cloud.version>
         <spring-cloud.version>Hoxton.RELEASE</spring-cloud.version>
+        <swagger2_version>2.9.2</swagger2_version>
         <!-- springboot 2.2.1默认的es版本是6.8.4，下面的es的版本要和ES的版本一致 -->
         <!-- springboot 2.2.1默认的es版本是6.8.4，下面的es的版本要和ES的版本一致 -->
-        <elasticsearch.version>7.4.0</elasticsearch.version>
+        <elasticsearch.version>7.5.1</elasticsearch.version>
     </properties>
     </properties>
 
 
     <dependencies>
     <dependencies>
@@ -31,6 +32,17 @@
         </dependency>
         </dependency>
 
 
         <dependency>
         <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_2.13</artifactId>
+            <version>2.6.0</version>
+        </dependency>
+
+        <dependency>
             <groupId>org.springframework.boot</groupId>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter</artifactId>
             <artifactId>spring-boot-starter</artifactId>
         </dependency>
         </dependency>
@@ -59,6 +71,41 @@
          <dependency>
          <dependency>
             <groupId>org.springframework.boot</groupId>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
             <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
+             <exclusions>
+                 <exclusion>
+                     <groupId>org.elasticsearch</groupId>
+                     <artifactId>elasticsearch</artifactId>
+                 </exclusion>
+                 <exclusion>
+                     <groupId>org.elasticsearch.client</groupId>
+                     <artifactId>elasticsearch-rest-high-level-client</artifactId>
+                 </exclusion>
+             </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+            <version>${elasticsearch.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.elasticsearch</groupId>
+                    <artifactId>elasticsearch</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.elasticsearch.client</groupId>
+                    <artifactId>elasticsearch-rest-client</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-client</artifactId>
+            <version>${elasticsearch.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>${elasticsearch.version}</version>
         </dependency>
         </dependency>
         <!-- SpringBoot end -->
         <!-- SpringBoot end -->
 
 
@@ -79,11 +126,6 @@
         </dependency>
         </dependency>
 
 
         <dependency>
         <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-zipkin</artifactId>
-        </dependency>
-
-        <dependency>
             <groupId>de.codecentric</groupId>
             <groupId>de.codecentric</groupId>
             <artifactId>spring-boot-admin-starter-client</artifactId>
             <artifactId>spring-boot-admin-starter-client</artifactId>
             <version>2.2.0</version>
             <version>2.2.0</version>
@@ -148,16 +190,21 @@
             <artifactId>log4j</artifactId>
             <artifactId>log4j</artifactId>
             <version>1.2.17</version>
             <version>1.2.17</version>
         </dependency>
         </dependency>
-
+        <!-- https://mvnrepository.com/artifact/com.github.xiaoymin/swagger-bootstrap-ui -->
+        <dependency>
+            <groupId>com.github.xiaoymin</groupId>
+            <artifactId>swagger-bootstrap-ui</artifactId>
+            <version>1.9.6</version>
+        </dependency>
         <dependency>
         <dependency>
             <groupId>io.springfox</groupId>
             <groupId>io.springfox</groupId>
             <artifactId>springfox-swagger2</artifactId>
             <artifactId>springfox-swagger2</artifactId>
             <version>2.9.2</version>
             <version>2.9.2</version>
         </dependency>
         </dependency>
         <dependency>
         <dependency>
-          <groupId>com.github.xiaoymin</groupId>
-          <artifactId>swagger-bootstrap-ui</artifactId>
-          <version>1.9.6</version>
+            <groupId>io.springfox</groupId>
+            <artifactId>springfox-swagger-ui</artifactId>
+            <version>2.9.2</version>
         </dependency>
         </dependency>
         <dependency>
         <dependency>
             <groupId>joda-time</groupId>
             <groupId>joda-time</groupId>
@@ -225,6 +272,17 @@
                 <groupId>org.springframework.boot</groupId>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
             </plugin>
             </plugin>
+            <!-- mybatis generator 自动生成代码插件 -->
+            <plugin>
+                <groupId>org.mybatis.generator</groupId>
+                <artifactId>mybatis-generator-maven-plugin</artifactId>
+                <version>1.3.2</version>
+                <configuration>
+                    <configurationFile>${basedir}/src/main/resources/generator/generatorConfig.xml</configurationFile>
+                    <overwrite>true</overwrite>
+                    <verbose>true</verbose>
+                </configuration>
+            </plugin>
         </plugins>
         </plugins>
     </build>
     </build>
     <repositories>
     <repositories>

@@ -2,7 +2,9 @@ package com.sunyo.wlpt.message.bus.service;
 
 
 import org.mybatis.spring.annotation.MapperScan;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.data.elasticsearch.ElasticsearchAutoConfiguration;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
 import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
 import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.cloud.openfeign.EnableFeignClients;
@@ -21,6 +23,7 @@ import org.springframework.transaction.annotation.EnableTransactionManagement;
 @EnableTransactionManagement
 @EnableTransactionManagement
 @EnableScheduling
 @EnableScheduling
 @EnableAsync
 @EnableAsync
+
 public class MessageBusServiceApplication {
 public class MessageBusServiceApplication {
 
 
     public static void main(String[] args) {
     public static void main(String[] args) {

+package com.sunyo.wlpt.message.bus.service;
+
+import com.google.common.base.Predicates;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import springfox.documentation.builders.ApiInfoBuilder;
+import springfox.documentation.builders.PathSelectors;
+import springfox.documentation.builders.RequestHandlerSelectors;
+import springfox.documentation.service.ApiInfo;
+import springfox.documentation.service.Contact;
+import springfox.documentation.spi.DocumentationType;
+import springfox.documentation.spring.web.plugins.Docket;
+import springfox.documentation.swagger2.annotations.EnableSwagger2;
+
+@Configuration
+@EnableSwagger2 // Swagger的开关，表示已经启用Swagger
+public class SwaggerConfig {
+    @Bean
+    public Docket api() {
+        Docket docket = new Docket(DocumentationType.SWAGGER_2)
+                .apiInfo(apiInfo())
+                .pathMapping("/")
+                .select() // 选择哪些路径和api会生成document
+                .apis(RequestHandlerSelectors.any())// 对所有api进行监控
+//                .apis(RequestHandlerSelectors.basePackage("com.hanstrovsky.controller"))// 选择监控的package
+//                .apis(RequestHandlerSelectors.withMethodAnnotation(ApiOperation.class))// 只监控有ApiOperation注解的接口
+                //不显示错误的接口地址
+                .paths(Predicates.not(PathSelectors.regex("/error.*")))//错误路径不监控
+                .paths(PathSelectors.regex("/.*"))// 对根下所有路径进行监控
+                .build();
+        return docket;
+    }
+
+    private ApiInfo apiInfo() {
+        return new ApiInfoBuilder().title("消息总线项目接口文档")
+                .contact(new Contact("Hanstrovsky", "www.hanstrovsky.com", "Hanstrovsky@gmail.com"))
+                .description("这是用Swagger动态生成的接口文档")
+                .termsOfServiceUrl("NO terms of service")
+                .license("The Apache License, Version 2.0")
+                .licenseUrl("http://www.apache.org/licenses/LICENSE-2.0.html")
+                .version("1.0")
+                .build();
+    }
+}

+package com.sunyo.wlpt.message.bus.service.bean;
+
+
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.elasticsearch.client.ClientConfiguration;
+import org.springframework.data.elasticsearch.client.RestClients;
+
+@Configuration
+public class EsRestClientConfig {
+
+    @Bean(name = "EsHighLevelClient")
+    public RestHighLevelClient highLevelClient() {
+        final ClientConfiguration clientConfiguration= ClientConfiguration.builder().connectedTo("192.168.1.73:9200").build();
+        return RestClients.create(clientConfiguration).rest();
+    }
+}

+package com.sunyo.wlpt.message.bus.service.bean;
+
+import com.sunyo.wlpt.message.bus.service.domain.BusServer;
+import com.sunyo.wlpt.message.bus.service.mapper.BusServerMapper;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.KafkaAdmin;
+
+import javax.annotation.Resource;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+//@Configuration
+public class KafkaInitialConfiguration {
+
+//    @Resource
+    private BusServerMapper busServerMapper;
+
+//    @Bean
+    public AdminClient adminClient() {
+        Map<String, Object> configs = new HashMap<>();
+        String serverMap = ServerListForMap();
+        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+                serverMap);
+        KafkaAdmin admin = new KafkaAdmin(configs);
+        return AdminClient.create(admin.getConfig());
+    }
+
+    public String ServerListForMap(){
+        List<BusServer> serverList = busServerMapper.selectBusServerList(new BusServer());
+
+        String KAFKA_SERVERS = serverList.stream().map(item -> {
+            return item.getServerIp()+":"+item.getServerPort();
+        })
+                .collect(Collectors.joining(","));
+
+        return KAFKA_SERVERS;
+    }
+}

@@ -13,6 +13,7 @@ import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;
 import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;
@@ -56,7 +57,7 @@ public class ElasticSearchConfig extends AbstractElasticsearchConfiguration {
 
 
 
 
     @Override
     @Override
-    @Bean
+//    @Bean
     public RestHighLevelClient elasticsearchClient()
     public RestHighLevelClient elasticsearchClient()
     {
     {
         final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
         final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();

-package com.sunyo.wlpt.message.bus.service.config;
-
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import springfox.documentation.builders.ApiInfoBuilder;
-import springfox.documentation.builders.PathSelectors;
-import springfox.documentation.builders.RequestHandlerSelectors;
-import springfox.documentation.service.ApiInfo;
-import springfox.documentation.spi.DocumentationType;
-import springfox.documentation.spring.web.plugins.Docket;
-import springfox.documentation.swagger2.annotations.EnableSwagger2;
-
-/**
- * @author 子诚
- * Description：swagger-knife4j 的配置文件
- * 时间：2020/7/10 11:48
- */
-@Configuration
-@EnableSwagger2
-public class SwaggerConfig {
-    @Bean
-    public Docket createRestApi()
-    {
-        return new Docket(DocumentationType.SWAGGER_2)
-                .apiInfo(apiInfo())
-                .select()
-                .apis(RequestHandlerSelectors.basePackage("com.sunyo.wlpt.message.bus.service.controller"))
-                .paths(PathSelectors.any())
-                .build();
-    }
-
-    private ApiInfo apiInfo()
-    {
-        return new ApiInfoBuilder()
-                .title("消息总线平台--后台管理服务")
-                .description("消息总线平台--后台管理服务")
-                .termsOfServiceUrl("http://localhost:9030/")
-                .contact("子诚")
-                .version("1.0.0")
-                .build();
-    }
-}

@@ -2,9 +2,14 @@ package com.sunyo.wlpt.message.bus.service.controller;
 
 
 import com.github.pagehelper.PageInfo;
 import com.github.pagehelper.PageInfo;
 import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
 import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
+import com.sunyo.wlpt.message.bus.service.model.ConsumerGroupOffsets;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.service.BusQueueService;
 import com.sunyo.wlpt.message.bus.service.service.BusQueueService;
+import com.sunyo.wlpt.message.bus.service.service.KafkaService;
 import com.sunyo.wlpt.message.bus.service.service.UserInfoService;
 import com.sunyo.wlpt.message.bus.service.service.UserInfoService;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.web.bind.annotation.*;
 
 
 import javax.annotation.Resource;
 import javax.annotation.Resource;
@@ -30,6 +35,10 @@ public class BusQueueController {
     @Resource
     @Resource
     private BusQueueService busQueueService;
     private BusQueueService busQueueService;
 
 
+    @Resource
+    private KafkaService kafkaService;
+
+
     /**
     /**
      * 分页查询，消息队列-列表
      * 分页查询，消息队列-列表
      *
      *
@@ -182,4 +191,12 @@ public class BusQueueController {
         }
         }
     }
     }
 
 
+    @GetMapping("monitor")
+    public ResultJson<List<ConsumerGroupOffsets>> queueMonitor(){
+
+         List<ConsumerGroupOffsets> result = kafkaService.queueMonitor();
+
+         return new ResultJson<List<ConsumerGroupOffsets>>("200","success",result);
+    }
+
 }
 }

+package com.sunyo.wlpt.message.bus.service.controller;
+
+
+import com.github.pagehelper.PageInfo;
+import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
+import com.sunyo.wlpt.message.bus.service.model.MessageType;
+import com.sunyo.wlpt.message.bus.service.response.ResultJson;
+import com.sunyo.wlpt.message.bus.service.service.MessageTypeService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.web.bind.annotation.*;
+
+import javax.annotation.Resource;
+import java.util.List;
+
+@RestController
+@RequestMapping("bus/message/type")
+public class MessageTypesController {
+
+    @Autowired
+    MessageTypeService messageTypeService;
+
+    @GetMapping("list")
+    public ResultJson<List<MessageType>> typeList()
+    {
+        MessageType messageType = new MessageType();
+        messageType.setPageNum(0);
+        messageType.setPageSize(1000000);
+        List<MessageType> list = messageTypeService.list(messageType);
+        return new ResultJson("200","success",list);
+    }
+
+    @PostMapping("list")
+    public ResultJson<PageInfo<MessageType>> typeListPage(@RequestBody MessageType messageType)
+    {
+        PageInfo<MessageType> list = messageTypeService.pageList(messageType);
+        return new ResultJson("200","success",list);
+    }
+}

+package com.sunyo.wlpt.message.bus.service.controller;
+
+
+import io.swagger.annotations.ApiOperation;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RequestMapping("router")
+@RestController
+public class RouterController {
+
+    @ApiOperation(value = "批量添加消息路由", notes = "超级管理修改其他用户密码")
+    @PostMapping
+    public void batchAddRouter(){
+
+
+
+    }
+}

 package com.sunyo.wlpt.message.bus.service.controller.es;
 package com.sunyo.wlpt.message.bus.service.controller.es;
 
 
 import com.sunyo.wlpt.message.bus.service.domain.es.ElasticSearchInfo;
 import com.sunyo.wlpt.message.bus.service.domain.es.ElasticSearchInfo;
+import com.sunyo.wlpt.message.bus.service.exception.CustomException;
+import com.sunyo.wlpt.message.bus.service.exception.CustomExceptionType;
+import com.sunyo.wlpt.message.bus.service.model.ESPage;
+import com.sunyo.wlpt.message.bus.service.model.MessageBusMsg;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.service.ElasticSearchInfoService;
 import com.sunyo.wlpt.message.bus.service.service.ElasticSearchInfoService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.StringUtils;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.MatchQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.query.WildcardQueryBuilder;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.web.bind.annotation.*;
 
 
 import javax.annotation.Resource;
 import javax.annotation.Resource;
+import java.util.Date;
+import java.util.List;
 
 
 /**
 /**
  * @author 子诚
  * @author 子诚
  * Description：
  * Description：
  * 时间：2020/9/10 17:09
  * 时间：2020/9/10 17:09
  */
  */
+@Slf4j
 @CrossOrigin
 @CrossOrigin
 @RequestMapping("bus/es")
 @RequestMapping("bus/es")
 @RestController
 @RestController
@@ -19,6 +38,7 @@ public class ElasticSearchInfoController {
     @Resource
     @Resource
     private ElasticSearchInfoService elasticSearchInfoService;
     private ElasticSearchInfoService elasticSearchInfoService;
 
 
+
     /**
     /**
      * 分页查询，ES信息列表
      * 分页查询，ES信息列表
      *
      *
@@ -82,4 +102,90 @@ public class ElasticSearchInfoController {
     {
     {
         return elasticSearchInfoService.updateByPrimaryKeySelective(elasticSearchInfo);
         return elasticSearchInfoService.updateByPrimaryKeySelective(elasticSearchInfo);
     }
     }
+
+
+    @RequestMapping("/search/wildmsg")
+    public ResultJson searchMsg(@RequestBody MessageBusMsg messageBusMsg)
+    {
+        Sort sort = Sort.by(Sort.Direction.DESC, "ddtm","creatime");
+        //前端提交的起始页从1开始,ES是从0开始
+        ESPage page = ESPage.of(messageBusMsg.getPageNum()-1,messageBusMsg.getPageSize(),sort);
+        WildcardQueryBuilder queryBuilders=null;
+        if(StringUtils.isNotBlank(messageBusMsg.getAlias_sendContent())){
+            queryBuilders = QueryBuilders.wildcardQuery("msg", "*"+messageBusMsg.getAlias_sendContent().toLowerCase()+"*");
+        }else {
+            return  ResultJson.error(new CustomException(CustomExceptionType.SEARCH_EXCEPTION));
+        }
+
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+        searchSourceBuilder.query(queryBuilders);
+        Page<MessageBusMsg> data = elasticSearchInfoService.search(searchSourceBuilder,MessageBusMsg.class,page);
+        ResultJson resultJson = new ResultJson("200","success",data);
+        return  resultJson;
+    }
+
+    @PostMapping("/search/term")
+    public ResultJson search(@RequestBody MessageBusMsg messageBusMsg)
+    {
+
+        Sort sort = Sort.by(Sort.Direction.DESC, "ddtm","creatime");
+        ESPage page = ESPage.of(messageBusMsg.getPageNum()-1,messageBusMsg.getPageSize(),sort);
+        /**
+         * term精确字段检索不要与matchQuery检索混用
+         */
+        BoolQueryBuilder queryBuilder =  QueryBuilders.boolQuery();
+
+        queryBuilder = esQueryFilter(queryBuilder,messageBusMsg);
+        log.info(new Date().toString());
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+        searchSourceBuilder.query(queryBuilder);
+        Page<MessageBusMsg> data = elasticSearchInfoService.search(searchSourceBuilder,MessageBusMsg.class,page);
+        ResultJson resultJson = new ResultJson("200","success",data);
+        return  resultJson;
+    }
+
+    public BoolQueryBuilder esQueryFilter(BoolQueryBuilder queryBuilder, MessageBusMsg messageBusMsg){
+
+        //and条件
+        if (StringUtils.isNotBlank(messageBusMsg.getSndr())){
+            queryBuilder =  queryBuilder.must(QueryBuilders.termQuery("sndr",messageBusMsg.getSndr()));
+        }
+        if (StringUtils.isNotBlank(messageBusMsg.getBtype())){
+            queryBuilder =  queryBuilder.must(QueryBuilders.termQuery("btype",messageBusMsg.getBtype()));
+        }
+        if (StringUtils.isNotBlank(messageBusMsg.getStype())){
+            queryBuilder =  queryBuilder.must(QueryBuilders.termQuery("stype",messageBusMsg.getStype()));
+        }
+
+        if (StringUtils.isNotBlank(messageBusMsg.getSeqn())){
+            queryBuilder =  queryBuilder.must(QueryBuilders.termQuery("seqn",messageBusMsg.getSeqn()));
+        }
+        if (StringUtils.isNotBlank(messageBusMsg.getMsgid())){
+            queryBuilder =  queryBuilder.must(QueryBuilders.termQuery("msgid",messageBusMsg.getMsgid()));
+        }
+
+        if (StringUtils.isNotBlank(messageBusMsg.getRcvrsUserName())){
+            queryBuilder =  queryBuilder.should(QueryBuilders.termQuery("rcvrs.username",messageBusMsg.getRcvrsUserName()));
+        }
+
+        if (StringUtils.isNotBlank(messageBusMsg.getRcvlogUsername())){
+            queryBuilder =  queryBuilder.should(QueryBuilders.termQuery("rcvlog.username",messageBusMsg.getRcvlogUsername()));
+        }
+        if (messageBusMsg.getCreatimeSearch() !=null && !messageBusMsg.getCreatimeSearch().isEmpty()  && StringUtils.isNotBlank(messageBusMsg.getCreatimeSearch().get(0)) && StringUtils.isNotBlank(messageBusMsg.getCreatimeSearch().get(1))){
+            queryBuilder =  queryBuilder.must(QueryBuilders.rangeQuery("ddtm").from(messageBusMsg.getCreatimeSearch().get(0)).to(messageBusMsg.getCreatimeSearch().get(1)));
+            //or条件
+            queryBuilder =  queryBuilder.should(QueryBuilders.rangeQuery("creatime").from(messageBusMsg.getCreatimeSearch().get(0)).to(messageBusMsg.getCreatimeSearch().get(1)));
+        }
+
+        if (StringUtils.isNotBlank(messageBusMsg.getMsgid())){
+            queryBuilder =  queryBuilder.should(QueryBuilders.termQuery("_id",messageBusMsg.getMsgid()));
+        }
+        //wild搜索大写搜不到小写也搜不到大写的包含，要转成小写
+        if (StringUtils.isNotBlank(messageBusMsg.getAlias_sendContent())){
+            queryBuilder =  queryBuilder.must(QueryBuilders.wildcardQuery("msg", "*"+messageBusMsg.getAlias_sendContent().toLowerCase()+"*"));
+        }
+        log.info("bool查询语句为:{}",queryBuilder);
+        return queryBuilder;
+    }
+
 }
 }

@@ -86,4 +86,20 @@ public class BusQueue implements Serializable {
      * 一个队列对应一个虚拟机
      * 一个队列对应一个虚拟机
      */
      */
     private VirtualHost virtualHost;
     private VirtualHost virtualHost;
+
+    /**
+     * topic partition数量，默认3
+     */
+    private int partitionCount;
+
+    /**
+     * 队列类型，默认1 kafka
+     */
+    private int queueType;
+    private int serverType;
+
+    /**
+     * 消费者组名
+     */
+    private String consumerGroupName;
 }
 }

@@ -90,4 +90,8 @@ public class BusServer implements Serializable {
      */
      */
     private String aliasName;
     private String aliasName;
 
 
+    /**
+     * 服务类型 1kafka 2rabbit
+     */
+    private String serverType;
 }
 }

@@ -14,6 +14,8 @@ public enum CustomExceptionType {
     SERVER_EXCEPTION("10500", "服务器异常，发送消息失败！"),
     SERVER_EXCEPTION("10500", "服务器异常，发送消息失败！"),
     CLIENT_EXCEPTION("10400", "报文格式错误，请检查报文格式！"),
     CLIENT_EXCEPTION("10400", "报文格式错误，请检查报文格式！"),
 
 
+    SEARCH_EXCEPTION("10600", "搜索内容不能为空"),
+
     BINDING_ERROR("10501", "配置信息，未进行绑定！"),
     BINDING_ERROR("10501", "配置信息，未进行绑定！"),
     SENDER_ERROR("10401", "报文格式错误，发送者不能为空！"),
     SENDER_ERROR("10401", "报文格式错误，发送者不能为空！"),
     CONTENT_ERROR("10402", "报文格式错误，消息内容不能为空！"),
     CONTENT_ERROR("10402", "报文格式错误，消息内容不能为空！"),
@@ -45,6 +47,8 @@ public enum CustomExceptionType {
     RECEIVE_HOST_NO_EXIST("30414", "参数错误，虚拟主机名称不能为空！"),
     RECEIVE_HOST_NO_EXIST("30414", "参数错误，虚拟主机名称不能为空！"),
     RECEIVE_QUEUE_NO_EXIST("30415", "参数错误，队列名称不能为空！"),
     RECEIVE_QUEUE_NO_EXIST("30415", "参数错误，队列名称不能为空！"),
 
 
+    KAFKA_QUEUE_ADD_ERR("40000","添加kafka队列失败"),
+
 
 
     CLIENT_ERROR("400", "客户端异常"),
     CLIENT_ERROR("400", "客户端异常"),
     SYSTEM_ERROR("500", "系统服务异常"),
     SYSTEM_ERROR("500", "系统服务异常"),

+package com.sunyo.wlpt.message.bus.service.mapper;
+
+import com.sunyo.wlpt.message.bus.service.model.Btype;
+
+public interface BtypeMapper {
+    int deleteByPrimaryKey(String id);
+
+    int insert(Btype record);
+
+    int insertSelective(Btype record);
+
+    Btype selectByPrimaryKey(String id);
+
+    int updateByPrimaryKeySelective(Btype record);
+
+    int updateByPrimaryKey(Btype record);
+}

+package com.sunyo.wlpt.message.bus.service.mapper;
+
+import com.sunyo.wlpt.message.bus.service.model.ConsumerGroup;
+
+import java.util.List;
+
+public interface ConsumerGroupMapper {
+    int deleteByPrimaryKey(String id);
+
+    int insert(ConsumerGroup record);
+
+    int insertSelective(ConsumerGroup record);
+
+    ConsumerGroup selectByPrimaryKey(String id);
+
+    List<String> groups();
+
+    int updateByPrimaryKeySelective(ConsumerGroup record);
+
+    int updateByPrimaryKey(ConsumerGroup record);
+}

+package com.sunyo.wlpt.message.bus.service.mapper;
+
+import com.sunyo.wlpt.message.bus.service.model.MessageRouter;
+
+public interface MessageRouterMapper {
+    int deleteByPrimaryKey(String id);
+
+    int insert(MessageRouter record);
+
+    int insertSelective(MessageRouter record);
+
+    MessageRouter selectByPrimaryKey(String id);
+
+    int updateByPrimaryKeySelective(MessageRouter record);
+
+    int updateByPrimaryKey(MessageRouter record);
+}

+package com.sunyo.wlpt.message.bus.service.mapper;
+
+import com.sunyo.wlpt.message.bus.service.model.MessageRouterReciverFilter;
+
+public interface MessageRouterReciverFilterMapper {
+    int deleteByPrimaryKey(String id);
+
+    int insert(MessageRouterReciverFilter record);
+
+    int insertSelective(MessageRouterReciverFilter record);
+
+    MessageRouterReciverFilter selectByPrimaryKey(String id);
+
+    int updateByPrimaryKeySelective(MessageRouterReciverFilter record);
+
+    int updateByPrimaryKey(MessageRouterReciverFilter record);
+}

+package com.sunyo.wlpt.message.bus.service.mapper;
+
+import com.sunyo.wlpt.message.bus.service.model.MessageRouterReciver;
+
+public interface MessageRouterReciverMapper {
+    int deleteByPrimaryKey(String id);
+
+    int insert(MessageRouterReciver record);
+
+    int insertSelective(MessageRouterReciver record);
+
+    MessageRouterReciver selectByPrimaryKey(String id);
+
+    int updateByPrimaryKeySelective(MessageRouterReciver record);
+
+    int updateByPrimaryKey(MessageRouterReciver record);
+}

+package com.sunyo.wlpt.message.bus.service.mapper;
+
+import com.sunyo.wlpt.message.bus.service.model.MessageType;
+
+import java.util.List;
+
+public interface MessageTypeMapper {
+    int deleteByPrimaryKey(String id);
+
+    int insert(MessageType record);
+
+    int insertSelective(MessageType record);
+
+    MessageType selectByPrimaryKey(String id);
+
+    List<MessageType> selectByParentId(String id);
+
+    List<MessageType> selectAll(MessageType record);
+
+    int updateByPrimaryKeySelective(MessageType record);
+
+    int updateByPrimaryKey(MessageType record);
+}

+package com.sunyo.wlpt.message.bus.service.mapper;
+
+import com.sunyo.wlpt.message.bus.service.model.Stype;
+
+public interface StypeMapper {
+    int deleteByPrimaryKey(String id);
+
+    int insert(Stype record);
+
+    int insertSelective(Stype record);
+
+    Stype selectByPrimaryKey(String id);
+
+    int updateByPrimaryKeySelective(Stype record);
+
+    int updateByPrimaryKey(Stype record);
+}

+package com.sunyo.wlpt.message.bus.service.mapper;
+
+import com.sunyo.wlpt.message.bus.service.model.UserTopic;
+
+public interface UserTopicMapper {
+    int deleteByPrimaryKey(String id);
+
+    int insert(UserTopic record);
+
+    int insertSelective(UserTopic record);
+
+    UserTopic selectByPrimaryKey(String id);
+
+    int updateByPrimaryKeySelective(UserTopic record);
+
+    int updateByPrimaryKey(UserTopic record);
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import lombok.Data;
+
+@Data
+public class BasePage {
+    /**
+     * //开始页
+     */
+    private int pageNum =0;
+
+    /**
+     * 每页多少条数据
+     */
+    private int pageSize=10;
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import java.util.List;
+
+public class Btype {
+    private String id;
+
+    private String name;
+
+    private String parentId;
+
+    private String des;
+
+    private List<Stype> children;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id == null ? null : id.trim();
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name == null ? null : name.trim();
+    }
+
+    public String getParentId() {
+        return parentId;
+    }
+
+    public void setParentId(String parentId) {
+        this.parentId = parentId == null ? null : parentId.trim();
+    }
+
+    public String getDes() {
+        return des;
+    }
+
+    public void setDes(String des) {
+        this.des = des == null ? null : des.trim();
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+public class ConsumerGroup {
+    private String id;
+
+    private String name;
+
+    private String des;
+
+    private String userId;
+
+    private String username;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id == null ? null : id.trim();
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name == null ? null : name.trim();
+    }
+
+    public String getDes() {
+        return des;
+    }
+
+    public void setDes(String des) {
+        this.des = des == null ? null : des.trim();
+    }
+
+    public String getUserId() {
+        return userId;
+    }
+
+    public void setUserId(String userId) {
+        this.userId = userId == null ? null : userId.trim();
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username == null ? null : username.trim();
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import lombok.Data;
+
+/**
+ * @author mrz
+ */
+@Data
+public class ConsumerGroupOffsets  {
+    private String groupName;
+    private String topic;
+    private int partition;
+    private long offset;
+    private long endoffset;
+    private long lag;
+
+    public long getLag() {
+        return this.lag;
+    }
+    public void setLag(){
+        this.lag = endoffset - offset;
+    }
+
+    public ConsumerGroupOffsets(String consumerGroupName, String topic, int partition, long offset, long endoffset) {
+        this.groupName = consumerGroupName;
+        this.topic = topic;
+        this.partition = partition;
+        this.offset = offset;
+        this.endoffset = endoffset;
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+public class ESMessageModel {
+    private String _index;
+    private String _type;
+    private String _id;
+    private int _score;
+    private Object _source;
+
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import lombok.Data;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.util.Assert;
+
+
+public class ESPage  implements Pageable {
+    private static final long serialVersionUID = -4541509938956089563L;
+
+    /**
+     * //开始页
+     */
+    private int pageNum;
+
+    /**
+     * 每页多少条数据
+     */
+    private int pageSize;
+
+    /**
+     * 排序字段
+     */
+    final Sort sort;
+
+    protected ESPage(int pageNumber, int pageSize, Sort sort) {
+        if (pageNumber < 0) {
+            pageNumber = 0;
+        } else if (pageSize < 1) {
+            pageSize = 10;
+        }
+
+        this.pageNum = pageNumber;
+        this.pageSize = pageSize;
+
+        Assert.notNull(sort, "Sort must not be null!");
+        this.sort = sort;
+
+    }
+
+    public ESPage(Sort sort) {
+        this.sort=sort;
+    }
+
+    public static ESPage of(int page, int size) {
+        return of(page, size, Sort.unsorted());
+    }
+
+    public static ESPage of(int page, int size, Sort sort) {
+        return new ESPage(page, size, sort);
+    }
+
+    public static ESPage of(int page, int size, Sort.Direction direction, String... properties) {
+        return of(page, size, Sort.by(direction, properties));
+    }
+
+    @Override
+    public int getPageNumber() {
+        return  this.pageNum;
+    }
+
+    @Override
+    public long getOffset() {
+        return (long)this.pageNum * (long)this.pageSize;
+    }
+
+    @Override
+    public Pageable next() {
+        return new ESPage(this.getPageNumber() + 1, this.getPageSize(), this.getSort());
+    }
+
+    @Override
+    public Pageable previousOrFirst() {
+        return this.hasPrevious() ? this.previous() : this.first();
+    }
+
+    @Override
+    public Pageable first() {
+        return new ESPage(0, this.getPageSize(), this.getSort());
+    }
+
+    @Override
+    public boolean hasPrevious() {
+        return this.pageNum > 0;
+    }
+
+    @Override
+    public int getPageSize() {
+        return this.pageSize;
+    }
+
+    @Override
+    public Sort getSort() {
+        return this.sort;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("Page request [number: %d, size %d, sort: %s]", this.getPageNumber(), this.getPageSize(), this.sort);
+    }
+
+    public Pageable previous() {
+        return this.getPageNumber() == 0 ? this : new ESPage(this.getPageNumber() - 1, this.getPageSize(), this.getSort());
+    }
+
+
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import lombok.Data;
+import org.springframework.data.domain.Sort;
+
+import java.util.Date;
+import java.util.List;
+
+@Data
+public class MessageBusMsg{
+    private String sndr;
+    private String rcvr;
+    private String btype;
+    private String stype;
+    private Date   ddtm;
+    private String    seqn;
+    private String    msgid;
+    private String    msg;
+    private Date    creatime;
+
+    /**
+     * 查询搜索字段
+     */
+    private List<String>  creatimeSearch;
+    //消息接收者
+    private String  rcvrsUserName;
+    //路由接收者
+    private String  rcvlogUsername;
+    //搜索内容
+    private String  alias_sendContent;
+    /**
+     * //开始页
+     */
+    private int pageNum;
+
+    /**
+     * 每页多少条数据
+     */
+    private int pageSize;
+
+
+    private List<MessageBusMsgTarget> target;
+    private List<MessageBusMsgRcvrs> rcvrs;
+    private List<MessageBusMsgRcvlog> rcvlog;
+
+
+    public void setAlias_sendContent(String alias_sendContent) {
+        this.msg = alias_sendContent;
+        this.alias_sendContent = alias_sendContent;
+    }
+
+
+    public void setMsg(String msg) {
+        this.msg = msg;
+        this.alias_sendContent = msg;
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import lombok.Data;
+
+import java.util.Date;
+
+@Data
+public class MessageBusMsgRcvlog {
+    private String username;
+    private Date rvtm;
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import lombok.Data;
+
+@Data
+public class MessageBusMsgRcvrs {
+
+    private String username;
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import lombok.Data;
+
+@Data
+public class MessageBusMsgTarget {
+    private String filter;
+    private String value;
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import java.util.Date;
+
+public class MessageRouter {
+    private String id;
+
+    private String sndr;
+
+    private String btype;
+
+    private String stype;
+
+    private String optype;
+
+    private Byte msgLimit;
+
+    private String character;
+
+    private Boolean status;
+
+    private Boolean usage;
+
+    private String des;
+
+    private String ver;
+
+    private Date creatTime;
+
+    private Date updateTime;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id == null ? null : id.trim();
+    }
+
+    public String getSndr() {
+        return sndr;
+    }
+
+    public void setSndr(String sndr) {
+        this.sndr = sndr == null ? null : sndr.trim();
+    }
+
+    public String getBtype() {
+        return btype;
+    }
+
+    public void setBtype(String btype) {
+        this.btype = btype == null ? null : btype.trim();
+    }
+
+    public String getStype() {
+        return stype;
+    }
+
+    public void setStype(String stype) {
+        this.stype = stype == null ? null : stype.trim();
+    }
+
+    public String getOptype() {
+        return optype;
+    }
+
+    public void setOptype(String optype) {
+        this.optype = optype == null ? null : optype.trim();
+    }
+
+    public Byte getMsgLimit() {
+        return msgLimit;
+    }
+
+    public void setMsgLimit(Byte msgLimit) {
+        this.msgLimit = msgLimit;
+    }
+
+    public String getCharacter() {
+        return character;
+    }
+
+    public void setCharacter(String character) {
+        this.character = character == null ? null : character.trim();
+    }
+
+    public Boolean getStatus() {
+        return status;
+    }
+
+    public void setStatus(Boolean status) {
+        this.status = status;
+    }
+
+    public Boolean getUsage() {
+        return usage;
+    }
+
+    public void setUsage(Boolean usage) {
+        this.usage = usage;
+    }
+
+    public String getDes() {
+        return des;
+    }
+
+    public void setDes(String des) {
+        this.des = des == null ? null : des.trim();
+    }
+
+    public String getVer() {
+        return ver;
+    }
+
+    public void setVer(String ver) {
+        this.ver = ver == null ? null : ver.trim();
+    }
+
+    public Date getCreatTime() {
+        return creatTime;
+    }
+
+    public void setCreatTime(Date creatTime) {
+        this.creatTime = creatTime;
+    }
+
+    public Date getUpdateTime() {
+        return updateTime;
+    }
+
+    public void setUpdateTime(Date updateTime) {
+        this.updateTime = updateTime;
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import java.util.Date;
+
+public class MessageRouterReciver {
+    private String id;
+
+    private String rcvrTopic;
+
+    private String messageRouterId;
+
+    private Date creatTime;
+
+    private Date updateTime;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id == null ? null : id.trim();
+    }
+
+    public String getRcvrTopic() {
+        return rcvrTopic;
+    }
+
+    public void setRcvrTopic(String rcvrTopic) {
+        this.rcvrTopic = rcvrTopic == null ? null : rcvrTopic.trim();
+    }
+
+    public String getMessageRouterId() {
+        return messageRouterId;
+    }
+
+    public void setMessageRouterId(String messageRouterId) {
+        this.messageRouterId = messageRouterId == null ? null : messageRouterId.trim();
+    }
+
+    public Date getCreatTime() {
+        return creatTime;
+    }
+
+    public void setCreatTime(Date creatTime) {
+        this.creatTime = creatTime;
+    }
+
+    public Date getUpdateTime() {
+        return updateTime;
+    }
+
+    public void setUpdateTime(Date updateTime) {
+        this.updateTime = updateTime;
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import java.util.Date;
+
+public class MessageRouterReciverFilter {
+    private String id;
+
+    private String filter;
+
+    private String filterValue;
+
+    private String type;
+
+    private Boolean status;
+
+    private String messageRouterReciverId;
+
+    private Date creatTime;
+
+    private Date updateTime;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id == null ? null : id.trim();
+    }
+
+    public String getFilter() {
+        return filter;
+    }
+
+    public void setFilter(String filter) {
+        this.filter = filter == null ? null : filter.trim();
+    }
+
+    public String getFilterValue() {
+        return filterValue;
+    }
+
+    public void setFilterValue(String filterValue) {
+        this.filterValue = filterValue == null ? null : filterValue.trim();
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type == null ? null : type.trim();
+    }
+
+    public Boolean getStatus() {
+        return status;
+    }
+
+    public void setStatus(Boolean status) {
+        this.status = status;
+    }
+
+    public String getMessageRouterReciverId() {
+        return messageRouterReciverId;
+    }
+
+    public void setMessageRouterReciverId(String messageRouterReciverId) {
+        this.messageRouterReciverId = messageRouterReciverId == null ? null : messageRouterReciverId.trim();
+    }
+
+    public Date getCreatTime() {
+        return creatTime;
+    }
+
+    public void setCreatTime(Date creatTime) {
+        this.creatTime = creatTime;
+    }
+
+    public Date getUpdateTime() {
+        return updateTime;
+    }
+
+    public void setUpdateTime(Date updateTime) {
+        this.updateTime = updateTime;
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+import java.util.List;
+
+/**
+ * @author mrz
+ */
+public class MessageType extends BasePage{
+    private String id;
+
+    private String name;
+
+    private String parentId;
+
+    private String des;
+
+    private Integer type;
+
+    private List<MessageType> children;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id == null ? null : id.trim();
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name == null ? null : name.trim();
+    }
+
+    public String getParentId() {
+        return parentId;
+    }
+
+    public void setParentId(String parentId) {
+        this.parentId = parentId == null ? null : parentId.trim();
+    }
+
+    public String getDes() {
+        return des;
+    }
+
+    public void setDes(String des) {
+        this.des = des == null ? null : des.trim();
+    }
+
+    public Integer getType() {
+        return type;
+    }
+
+    public void setType(Integer type) {
+        this.type = type;
+    }
+
+    public List<MessageType> getChildren() {
+        return children;
+    }
+
+    public void setChildren(List<MessageType> children) {
+        this.children = children;
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+public class Stype {
+    private String id;
+
+    private String name;
+
+    private String des;
+
+    private String parentId;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id == null ? null : id.trim();
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name == null ? null : name.trim();
+    }
+
+    public String getDes() {
+        return des;
+    }
+
+    public void setDes(String des) {
+        this.des = des == null ? null : des.trim();
+    }
+
+    public String getParentId() {
+        return parentId;
+    }
+
+    public void setParentId(String parentId) {
+        this.parentId = parentId == null ? null : parentId.trim();
+    }
+}

+package com.sunyo.wlpt.message.bus.service.model;
+
+public class UserTopic {
+    private String id;
+
+    private String userId;
+
+    private String busQueueId;
+
+    private String username;
+
+    private String topic;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id == null ? null : id.trim();
+    }
+
+    public String getUserId() {
+        return userId;
+    }
+
+    public void setUserId(String userId) {
+        this.userId = userId == null ? null : userId.trim();
+    }
+
+    public String getBusQueueId() {
+        return busQueueId;
+    }
+
+    public void setBusQueueId(String busQueueId) {
+        this.busQueueId = busQueueId == null ? null : busQueueId.trim();
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username == null ? null : username.trim();
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic == null ? null : topic.trim();
+    }
+
+    public UserTopic(String id, String userId, String busQueueId, String username, String topic) {
+        this.id = id;
+        this.userId = userId;
+        this.busQueueId = busQueueId;
+        this.username = username;
+        this.topic = topic;
+    }
+    public UserTopic(){
+
+    }
+}

@@ -150,7 +150,7 @@ public class ResultJson<T> implements Serializable {
         } else if (e.getCode() == CustomExceptionType.SYSTEM_ERROR.getCode()) {
         } else if (e.getCode() == CustomExceptionType.SYSTEM_ERROR.getCode()) {
             result.setMsg(e.getMessage() + "；请将该异常发送给管理员");
             result.setMsg(e.getMessage() + "；请将该异常发送给管理员");
         } else {
         } else {
-            result.setMsg("系统出现未知异常，请联系管理员!");
+            result.setMsg(e.getMessage());
         }
         }
         // 可以尝试着做异常信息持久化
         // 可以尝试着做异常信息持久化
         return result;
         return result;

 package com.sunyo.wlpt.message.bus.service.service;
 package com.sunyo.wlpt.message.bus.service.service;
 
 
 import com.sunyo.wlpt.message.bus.service.domain.es.ElasticSearchInfo;
 import com.sunyo.wlpt.message.bus.service.domain.es.ElasticSearchInfo;
+import com.sunyo.wlpt.message.bus.service.model.ESPage;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.data.domain.Page;
 
 
 import java.util.List;
 import java.util.List;
 
 
@@ -84,6 +87,8 @@ public interface ElasticSearchInfoService {
      * @return
      * @return
      */
      */
     ResultJson batchRemoveByIds(String ids);
     ResultJson batchRemoveByIds(String ids);
+
+    <T> Page<T> search(SearchSourceBuilder builder, Class<T> c, ESPage page);
 }
 }
 
 
 
 

+package com.sunyo.wlpt.message.bus.service.service;
+
+import com.sunyo.wlpt.message.bus.service.model.ConsumerGroupOffsets;
+
+import java.util.List;
+
+public interface KafkaService {
+
+    boolean addTopic(String TopicName,int partitionNum);
+
+    public void updateAdminclient();
+
+    List<ConsumerGroupOffsets> queueMonitor();
+}

+package com.sunyo.wlpt.message.bus.service.service;
+
+import com.github.pagehelper.PageHelper;
+import com.github.pagehelper.PageInfo;
+import com.sunyo.wlpt.message.bus.service.model.MessageType;
+
+import java.util.List;
+
+public interface MessageTypeService {
+
+    PageInfo<MessageType> pageList(MessageType record);
+
+    List<MessageType> list(MessageType messageType);
+
+}

@@ -2,15 +2,19 @@ package com.sunyo.wlpt.message.bus.service.service.impl;
 
 
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
 import com.github.pagehelper.PageInfo;
+import com.sunyo.wlpt.message.bus.service.exception.CustomException;
+import com.sunyo.wlpt.message.bus.service.exception.CustomExceptionType;
+import com.sunyo.wlpt.message.bus.service.mapper.*;
 import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
 import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
-import com.sunyo.wlpt.message.bus.service.mapper.BusQueueMapper;
-import com.sunyo.wlpt.message.bus.service.mapper.UserInfoMapper;
-import com.sunyo.wlpt.message.bus.service.mapper.UserMessageBindingMapper;
+import com.sunyo.wlpt.message.bus.service.model.ConsumerGroup;
+import com.sunyo.wlpt.message.bus.service.model.UserTopic;
 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.service.BusQueueService;
 import com.sunyo.wlpt.message.bus.service.service.BusQueueService;
+import com.sunyo.wlpt.message.bus.service.service.KafkaService;
 import com.sunyo.wlpt.message.bus.service.utils.IdUtils;
 import com.sunyo.wlpt.message.bus.service.utils.IdUtils;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.StringUtil;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.transaction.annotation.Transactional;
@@ -38,10 +42,19 @@ public class BusQueueServiceImpl implements BusQueueService {
     @Resource
     @Resource
     private UserInfoMapper userInfoMapper;
     private UserInfoMapper userInfoMapper;
 
 
+    @Resource
+    UserTopicMapper userTopicMapper;
+
 
 
     @Resource
     @Resource
     private UserMessageBindingMapper userMessageBindingMapper;
     private UserMessageBindingMapper userMessageBindingMapper;
 
 
+    @Resource
+    private ConsumerGroupMapper consumerGroupMapper;
+
+    @Autowired
+    KafkaService kafkaService;
+
 
 
     @Override
     @Override
     @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
     @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
@@ -119,19 +132,34 @@ public class BusQueueServiceImpl implements BusQueueService {
      * <p>
      * <p>
      * 存储创建的队列信息于数据库
      * 存储创建的队列信息于数据库
      */
      */
+    @Transactional(rollbackFor = Exception.class)
     @Override
     @Override
-    public ResultJson insertSelective(BusQueue record) throws IOException, TimeoutException
-    {
-        if (userInfoMapper.selectUserExist(record.getUsername()).size() == 0) {
-            return new ResultJson<>("400", "该MQ用户信息，不存在");
-        }
-        if (userInfoMapper.selectByBusQueue(record).size() == 0) {
-            return new ResultJson<>("400", "该MQ用户关系，不存在");
-        }
+    public ResultJson insertSelective(BusQueue record)  {
+
+        String id = IdUtils.generateId();
+        ConsumerGroup consumerGroup =new ConsumerGroup();
+        consumerGroup.setId(id);
+        consumerGroup.setName(record.getConsumerGroupName());
+        consumerGroup.setUserId(record.getUserId());
+        consumerGroup.setUsername(record.getUsername());
+        int i =consumerGroupMapper.insertSelective(consumerGroup);
+
+        UserTopic userTopic = new UserTopic(id,record.getUserId(),id,record.getUsername(),record.getQueueName());
+        int iii = userTopicMapper.insertSelective(userTopic);
+//        if (true){
+//            throw new RuntimeException("test");
+//        }
         // 设置id
         // 设置id
-        record.setId(IdUtils.generateId());
-        rabbitUtils.toCreateQueue(record);
-        return busQueueMapper.insertSelective(record) > 0
+        record.setId(id);
+        int ii = busQueueMapper.insertSelective(record);
+
+
+        //插入到kafka服务topic
+        boolean addResult = kafkaService.addTopic(record.getQueueName(),record.getPartitionCount());
+        if (!addResult){
+            throw new CustomException(CustomExceptionType.KAFKA_QUEUE_ADD_ERR);
+        }
+        return ii > 0
                 ? new ResultJson<>("200", "添加消息队列，成功")
                 ? new ResultJson<>("200", "添加消息队列，成功")
                 : new ResultJson<>("500", "添加消息队列，失败");
                 : new ResultJson<>("500", "添加消息队列，失败");
     }
     }

@@ -8,6 +8,7 @@ import com.sunyo.wlpt.message.bus.service.mapper.UserInfoMapper;
 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
 import com.sunyo.wlpt.message.bus.service.service.*;
 import com.sunyo.wlpt.message.bus.service.service.*;
 import com.sunyo.wlpt.message.bus.service.utils.AESUtils;
 import com.sunyo.wlpt.message.bus.service.utils.AESUtils;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.transaction.annotation.Transactional;
@@ -50,6 +51,7 @@ public class BusServerServiceImpl implements BusServerService {
     private BusServerMapper busServerMapper;
     private BusServerMapper busServerMapper;
 
 
 
 
+
     @Override
     @Override
     public PageInfo selectBusServerList(BusServer busServer, Integer pageNum, Integer pageSize)
     public PageInfo selectBusServerList(BusServer busServer, Integer pageNum, Integer pageSize)
     {
     {
@@ -110,6 +112,7 @@ public class BusServerServiceImpl implements BusServerService {
         // 超级用户的密码，使用AES加密
         // 超级用户的密码，使用AES加密
         String superPassword = server.getSuperPassword();
         String superPassword = server.getSuperPassword();
         server.setSuperPassword(AESUtils.encrypt(superPassword));
         server.setSuperPassword(AESUtils.encrypt(superPassword));
+
         return busServerMapper.insertSelective(server);
         return busServerMapper.insertSelective(server);
     }
     }
 
 

 package com.sunyo.wlpt.message.bus.service.service.impl;
 package com.sunyo.wlpt.message.bus.service.service.impl;
 
 
+import com.alibaba.fastjson.JSON;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
 import com.github.pagehelper.PageInfo;
 import com.sunyo.wlpt.message.bus.service.domain.es.ElasticSearchInfo;
 import com.sunyo.wlpt.message.bus.service.domain.es.ElasticSearchInfo;
 import com.sunyo.wlpt.message.bus.service.mapper.ElasticSearchInfoMapper;
 import com.sunyo.wlpt.message.bus.service.mapper.ElasticSearchInfoMapper;
+
+import com.sunyo.wlpt.message.bus.service.model.ESPage;
+
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
 import com.sunyo.wlpt.message.bus.service.service.ElasticSearchInfoService;
 import com.sunyo.wlpt.message.bus.service.service.ElasticSearchInfoService;
 import com.sunyo.wlpt.message.bus.service.utils.IdUtils;
 import com.sunyo.wlpt.message.bus.service.utils.IdUtils;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.StringUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.SearchHits;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.sort.SortOrder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageImpl;
+import org.springframework.data.domain.Sort;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 
 
 import javax.annotation.Resource;
 import javax.annotation.Resource;
+import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.List;
 
 
 import static com.sunyo.wlpt.message.bus.service.common.Constant.RESULT_SUCCESS;
 import static com.sunyo.wlpt.message.bus.service.common.Constant.RESULT_SUCCESS;
@@ -20,12 +40,23 @@ import static com.sunyo.wlpt.message.bus.service.common.Constant.RESULT_SUCCESS;
  * Description：
  * Description：
  * 时间：2020/9/8 15:49
  * 时间：2020/9/8 15:49
  */
  */
+@Slf4j
 @Service
 @Service
 public class ElasticSearchInfoServiceImpl implements ElasticSearchInfoService {
 public class ElasticSearchInfoServiceImpl implements ElasticSearchInfoService {
 
 
     @Resource
     @Resource
     private ElasticSearchInfoMapper elasticSearchInfoMapper;
     private ElasticSearchInfoMapper elasticSearchInfoMapper;
 
 
+
+    @Qualifier("EsHighLevelClient")
+    @Resource
+    private RestHighLevelClient restHighLevelClient;
+
+    /**
+     * 索引名称
+     */
+    private static final String INDEX_NAME = "messagebus";
+
     @Override
     @Override
     public ResultJson deleteByPrimaryKey(String id)
     public ResultJson deleteByPrimaryKey(String id)
     {
     {
@@ -199,6 +230,58 @@ public class ElasticSearchInfoServiceImpl implements ElasticSearchInfoService {
                 : ResultJson.success("新增ES信息,通过检验!");
                 : ResultJson.success("新增ES信息,通过检验!");
     }
     }
 
 
+    /**
+     * @author WCNGS@QQ.COM
+     * @See
+     * @date 2019/10/17 17:14
+     * @param builder   查询参数
+     * @param c 结果类对象
+     * @return java.util.List<T>
+     * @throws
+     * @since
+     */
+    @Override
+    public <T> Page<T> search( SearchSourceBuilder builder, Class<T> c,ESPage page) {
+        SearchRequest request = new SearchRequest(INDEX_NAME);
+        builder = wrapperBuilder(builder,page);
+        request.source(builder);
+        try {
+            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
+            SearchHits searchHits = response.getHits();
+            SearchHit[] hits = searchHits.getHits();
+            long total = searchHits.getTotalHits().value;
+            log.info("搜索总条数:{}",total);
+            List<T> res = new ArrayList<>(hits.length);
+
+            String s = "";
+            for (SearchHit hit : hits) {
+                 s += hit.getSourceAsString()+"\n";
+                res.add(JSON.parseObject(hit.getSourceAsString(), c));
+            }
+
+            //返回分页实体
+            Page<T>  ResponsePage =new PageImpl<T>(res,page,total);
+            log.info("查询到{}条记录,分别是s:{}",hits.length,s);
+            return ResponsePage;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private SearchSourceBuilder wrapperBuilder(SearchSourceBuilder builder, ESPage pageable) {
+        builder.from(pageable.getPageNumber() * pageable.getPageSize());
+        builder.size(pageable.getPageSize());
+        Sort sort = pageable.getSort();
+        Iterator iterator = sort.iterator();
+        while (iterator.hasNext()) {
+            Sort.Order order = (Sort.Order) iterator.next();
+            //用keyword字段来排序，所以在建立索引的时候，就必须同步建立keyword字段
+            builder.sort(order.getProperty(), order.getDirection() == Sort.Direction.ASC ? SortOrder.ASC : SortOrder.DESC);
+        }
+        return builder;
+    }
+
+
 }
 }
 
 
 
 

@@ -51,7 +51,7 @@ public class ElasticsearchService {
     public ResultJson selectMessageNoteList(MessageNote messageNote, Integer pageNum, Integer pageSize) throws IOException
     public ResultJson selectMessageNoteList(MessageNote messageNote, Integer pageNum, Integer pageSize) throws IOException
     {
     {
         // 条件搜索
         // 条件搜索
-        SearchRequest searchRequest = new SearchRequest("message");
+        SearchRequest searchRequest = new SearchRequest("es01");
         SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
         SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
 
 
         // 高亮设置
         // 高亮设置
@@ -110,7 +110,7 @@ public class ElasticsearchService {
             }
             }
             if (!StringUtil.isNullOrEmpty(messageNote.getAlias_sendContent())) {
             if (!StringUtil.isNullOrEmpty(messageNote.getAlias_sendContent())) {
                 String content = messageNote.getAlias_sendContent();
                 String content = messageNote.getAlias_sendContent();
-                WildcardQueryBuilder sendContentQuery = QueryBuilders.wildcardQuery("alias_sendContent", content.matches(".*[a-zA-z].*") ? content.toLowerCase() : content);
+                WildcardQueryBuilder sendContentQuery = QueryBuilders.wildcardQuery("name", content.matches(".*[a-zA-z].*") ? content.toLowerCase() : content);
                 boolQueryBuilder.must(sendContentQuery);
                 boolQueryBuilder.must(sendContentQuery);
             }
             }
             if (messageNote.getSendTimeBegin() != null && messageNote.getSendTimeEnd() != null) {
             if (messageNote.getSendTimeBegin() != null && messageNote.getSendTimeEnd() != null) {

+package com.sunyo.wlpt.message.bus.service.service.impl;
+
+import com.github.pagehelper.PageHelper;
+import com.github.pagehelper.PageInfo;
+import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
+import com.sunyo.wlpt.message.bus.service.mapper.MessageTypeMapper;
+import com.sunyo.wlpt.message.bus.service.model.MessageType;
+import com.sunyo.wlpt.message.bus.service.service.MessageTypeService;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+
+@Service
+public class MessageTypeServiceImp implements MessageTypeService {
+
+    @Resource
+    MessageTypeMapper messageTypeMapper;
+
+    @Override
+    public List<MessageType> list(MessageType messageType) {
+        List<MessageType> list = messageTypeMapper.selectAll(messageType);
+        return list;
+    }
+
+    @Override
+    public PageInfo<MessageType> pageList(MessageType messageType) {
+        PageHelper.startPage(messageType.getPageNum(), messageType.getPageSize());
+        List<MessageType> list = messageTypeMapper.selectAll(messageType);
+        PageInfo<MessageType> pageInfo = new PageInfo<>(list);
+        return pageInfo;
+    }
+}

+package com.sunyo.wlpt.message.bus.service.service.kafka;
+
+
+import com.sunyo.wlpt.message.bus.service.domain.BusServer;
+import com.sunyo.wlpt.message.bus.service.mapper.BusServerMapper;
+import com.sunyo.wlpt.message.bus.service.mapper.ConsumerGroupMapper;
+import com.sunyo.wlpt.message.bus.service.model.ConsumerGroup;
+import com.sunyo.wlpt.message.bus.service.model.ConsumerGroupOffsets;
+import com.sunyo.wlpt.message.bus.service.service.KafkaService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.admin.*;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionReplica;
+import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.stereotype.Service;
+import javax.annotation.Resource;
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+/**
+ * @author mrz
+ */
+@Slf4j
+@Service
+public class KafkaServiceImp implements KafkaService {
+
+    @Resource
+    private BusServerMapper busServerMapper;
+
+    private static AdminClient KAFKA_ADMIN_CLIENT ;
+
+    private static KafkaConsumer KAFKA_CONSUMER;
+
+    @Resource
+    ConsumerGroupMapper consumerGroupMapper;
+
+    /**
+     * 根据用户名列表批量添加topic
+     */
+    @Override
+    public boolean addTopic(String TopicName,int partitionNum){
+
+        intAdminClient();
+        NewTopic newTopic = new NewTopic(TopicName, partitionNum, (short) 1);
+        List<NewTopic> topicList = Arrays.asList(newTopic);
+        KAFKA_ADMIN_CLIENT.createTopics(topicList);
+        return true;
+    }
+
+    public void intAdminClient(){
+        if (KAFKA_ADMIN_CLIENT!=null){
+            return;
+        }else {
+            updateAdminclient();
+        }
+    }
+
+    public void intConsumer(){
+        if (KAFKA_CONSUMER!=null){
+            return;
+        }else {
+            consumer();
+        }
+    }
+
+    @Override
+    public void updateAdminclient(){
+        Map<String, Object> configs = new HashMap<>();
+        String serverMap = ServerListForMap();
+        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+                serverMap);
+        configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        configs.put(AdminClientConfig.CLIENT_ID_CONFIG,"MessageBusManager");
+        KafkaAdmin admin = new KafkaAdmin(configs);
+        KAFKA_ADMIN_CLIENT = AdminClient.create(admin.getConfig());
+    }
+    /**
+     * 获取topicList
+     */
+    public void topicList(){
+
+    }
+
+
+    /**
+     * 删除kafka中的topic
+     */
+    public void delTopic(){
+
+    }
+
+    @Override
+    public List<ConsumerGroupOffsets> queueMonitor()  {
+        List<ConsumerGroupOffsets> result = new ArrayList<ConsumerGroupOffsets>();
+        try{
+            intAdminClient();
+            intConsumer();
+
+            /**
+             * 1.  获取consumerGroup 列表
+             */
+            List<String> consumerGroups = consumerGroupMapper.groups();
+
+            for (String groupname :consumerGroups) {
+
+                /**
+                 * 2. 获取获取consumerGroup消费信息
+                 */
+                ListConsumerGroupOffsetsResult listConsumerGroupOffsetsResult= KAFKA_ADMIN_CLIENT.listConsumerGroupOffsets(groupname);
+                listConsumerGroupOffsetsResult.partitionsToOffsetAndMetadata().get().forEach(
+                        (TopicPartition k, OffsetAndMetadata v)-> {
+                            /**
+                             * 3.获取consumerGroup消费的每个topic的partion的lastoffset的值
+                             */
+                            TopicPartition topicPartition = new TopicPartition(k.topic(),k.partition());
+                            long endoffset = getLogEndOffset(topicPartition);
+
+                            ConsumerGroupOffsets consumerGroupOffsets = new ConsumerGroupOffsets(groupname,k.topic(),k.partition(),v.offset(),endoffset);
+                            consumerGroupOffsets.setLag();
+                            result.add(consumerGroupOffsets);
+                            log.info("cconsumerGroup:{},[topic]：{},[partition]：{},[offset]:{}，[endoffset] = {} \n",
+                                    groupname,k.topic(),k.partition(),v.offset(),endoffset);
+
+
+                        }
+                );
+            }
+            return result;
+
+        }catch (ExecutionException e){
+
+        }catch (InterruptedException e){
+
+        }finally {
+
+        }
+        return null;
+    }
+
+    public long getLogEndOffset(TopicPartition topicPartition){
+        KAFKA_CONSUMER.assign(Arrays.asList(topicPartition));
+        KAFKA_CONSUMER.seekToEnd(Arrays.asList(topicPartition));
+        long endOffset = KAFKA_CONSUMER.position(topicPartition);
+        return endOffset;
+    }
+
+    public void consumer(){
+        Properties props = new Properties();
+        props.put("bootstrap.servers", ServerListForMap());
+        props.put("group.id", "test");
+        props.put("enable.auto.commit", "true");
+        props.put("auto.offset.reset", "earliest");
+        props.put("auto.commit.interval.ms", "1000");
+        props.put("auto.commit.interval.ms", "1000");
+        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+        KAFKA_CONSUMER= consumer;
+    }
+
+    public String ServerListForMap(){
+        List<BusServer> serverList = busServerMapper.selectBusServerList(new BusServer());
+
+        String KAFKA_SERVERS = serverList.stream().map(item -> {
+            return item.getServerIp()+":"+item.getServerPort();
+        })
+                .collect(Collectors.joining(","));
+        return KAFKA_SERVERS;
+    }
+
+}

+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE generatorConfiguration
+        PUBLIC "-//mybatis.org//DTD MyBatis Generator Configuration 1.0//EN"
+        "http://mybatis.org/dtd/mybatis-generator-config_1_0.dtd">
+<generatorConfiguration>
+    <!-- 数据库驱动:选择你的本地硬盘上面的数据库驱动包-->
+    <classPathEntry  location="/Users/mrz/Documents/maven/mysql-connector-java-5.1.7-bin.jar"/>
+    <!--<classPathEntry  location="/Users/mrz/Documents/maven/ojdbc6.jar"/>-->
+    <context id="DB2Tables"  targetRuntime="MyBatis3">
+        <commentGenerator>
+            <property name="suppressDate" value="true"/>
+            <!-- 是否去除自动生成的注释 true：是 ： false:否 -->
+            <property name="suppressAllComments" value="true"/>
+        </commentGenerator>
+        <!--数据库链接URL，用户名、密码 -->
+        <jdbcConnection driverClass="com.mysql.jdbc.Driver"
+                            connectionURL="jdbc:mysql://118.31.66.166:3306/bus_service"
+                            userId="110"
+                            password="QAHqCJf2kFYCLirM">
+        </jdbcConnection>
+        <!--<jdbcConnection driverClass="oracle.jdbc.driver.OracleDriver"-->
+                        <!--connectionURL="jdbc:oracle:thin:@10.50.3.68:1521:CGODW"-->
+                        <!--userId="CGOETL"-->
+                        <!--password="1q2w3e4r">-->
+        <!--</jdbcConnection>-->
+        <!-- 默认false，把JDBC DECIMAL 和 NUMERIC 类型解析为 Integer，为 true时把JDBC DECIMAL 和
+            NUMERIC 类型解析为java.math.BigDecimal -->
+        <!--<javaTypeResolver>-->
+        <!--<property name="forceBigDecimals" value="true" />-->
+        <!--</javaTypeResolver>-->
+        <javaTypeResolver>
+            <property name="forceBigDecimals" value="false"/>
+        </javaTypeResolver>
+        <!-- 生成模型的包名和位置-->
+        <javaModelGenerator targetPackage="com.sunyo.wlpt.message.bus.service.model" targetProject="src/main/java">
+            <property name="enableSubPackages" value="true"/>
+            <property name="trimStrings" value="true"/>
+        </javaModelGenerator>
+        <!-- 生成映射文件的包名和位置-->
+        <sqlMapGenerator targetPackage="mapper" targetProject="src/main/resources">
+            <property name="enableSubPackages" value="true"/>
+        </sqlMapGenerator>
+        <!-- 生成DAO的包名和位置-->
+        <javaClientGenerator type="XMLMAPPER" targetPackage="com.sunyo.wlpt.message.bus.service.mapper" targetProject="src/main/java">
+            <property name="enableSubPackages" value="true"/>
+        </javaClientGenerator>
+        <!-- 要生成的表 tableName是数据库中的表名或视图名 domainObjectName是实体类名-->
+        <table tableName="message_router_reciver_filter" domainObjectName="MessageRouterReciverFilter" enableCountByExample="false" enableUpdateByExample="false" enableDeleteByExample="false" enableSelectByExample="false" selectByExampleQueryId="false"></table>
+    </context>
+</generatorConfiguration>

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.BtypeMapper" >
+  <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.model.Btype" >
+    <id column="id" property="id" jdbcType="VARCHAR" />
+    <result column="name" property="name" jdbcType="VARCHAR" />
+    <result column="parent_id" property="parentId" jdbcType="VARCHAR" />
+    <result column="des" property="des" jdbcType="VARCHAR" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, name, parent_id, des
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select 
+    <include refid="Base_Column_List" />
+    from btype
+    where id = #{id,jdbcType=VARCHAR}
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
+    delete from btype
+    where id = #{id,jdbcType=VARCHAR}
+  </delete>
+  <insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.Btype" >
+    insert into btype (id, name, parent_id, 
+      des)
+    values (#{id,jdbcType=VARCHAR}, #{name,jdbcType=VARCHAR}, #{parentId,jdbcType=VARCHAR}, 
+      #{des,jdbcType=VARCHAR})
+  </insert>
+  <insert id="insertSelective" parameterType="com.sunyo.wlpt.message.bus.service.model.Btype" >
+    insert into btype
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        id,
+      </if>
+      <if test="name != null" >
+        name,
+      </if>
+      <if test="parentId != null" >
+        parent_id,
+      </if>
+      <if test="des != null" >
+        des,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=VARCHAR},
+      </if>
+      <if test="name != null" >
+        #{name,jdbcType=VARCHAR},
+      </if>
+      <if test="parentId != null" >
+        #{parentId,jdbcType=VARCHAR},
+      </if>
+      <if test="des != null" >
+        #{des,jdbcType=VARCHAR},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.model.Btype" >
+    update btype
+    <set >
+      <if test="name != null" >
+        name = #{name,jdbcType=VARCHAR},
+      </if>
+      <if test="parentId != null" >
+        parent_id = #{parentId,jdbcType=VARCHAR},
+      </if>
+      <if test="des != null" >
+        des = #{des,jdbcType=VARCHAR},
+      </if>
+    </set>
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.sunyo.wlpt.message.bus.service.model.Btype" >
+    update btype
+    set name = #{name,jdbcType=VARCHAR},
+      parent_id = #{parentId,jdbcType=VARCHAR},
+      des = #{des,jdbcType=VARCHAR}
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+</mapper>

 <?xml version="1.0" encoding="UTF-8"?>
 <?xml version="1.0" encoding="UTF-8"?>
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.BusQueueMapper">
 <mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.BusQueueMapper">
-    <cache-ref namespace="com.sunyo.wlpt.message.bus.service.mapper.VirtualHostMapper"/>
     <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.domain.BusQueue">
     <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.domain.BusQueue">
         <!--@mbg.generated-->
         <!--@mbg.generated-->
         <!--@Table bus_queue-->
         <!--@Table bus_queue-->
@@ -16,6 +15,9 @@
         <result column="description" jdbcType="VARCHAR" property="description"/>
         <result column="description" jdbcType="VARCHAR" property="description"/>
         <result column="gmt_create" jdbcType="TIMESTAMP" property="gmtCreate"/>
         <result column="gmt_create" jdbcType="TIMESTAMP" property="gmtCreate"/>
         <result column="gmt_modified" jdbcType="TIMESTAMP" property="gmtModified"/>
         <result column="gmt_modified" jdbcType="TIMESTAMP" property="gmtModified"/>
+        <result column="queue_type" jdbcType="INTEGER" property="queueType"/>
+        <result column="server_type" jdbcType="INTEGER" property="serverType"/>
+        <result column="partition_count" jdbcType="INTEGER" property="partitionCount"/>
     </resultMap>
     </resultMap>
 
 
     <!-- 该Mapper映射关系的作用，是队列与虚拟主机的1:1的关系映射 -->
     <!-- 该Mapper映射关系的作用，是队列与虚拟主机的1:1的关系映射 -->
@@ -29,7 +31,7 @@
   <sql id="Base_Column_List">
   <sql id="Base_Column_List">
       <!--@mbg.generated-->
       <!--@mbg.generated-->
       id, queue_name, user_id, username, virtual_host_id, durability, auto_delete, arguments,
       id, queue_name, user_id, username, virtual_host_id, durability, auto_delete, arguments,
-    description, gmt_create, gmt_modified
+    description, gmt_create, gmt_modified, queue_type, partition_count, server_type
   </sql>
   </sql>
 
 
   <select id="selectByPrimaryKey" parameterType="java.lang.String" resultMap="BaseResultMap">
   <select id="selectByPrimaryKey" parameterType="java.lang.String" resultMap="BaseResultMap">
@@ -109,6 +111,12 @@
       <if test="gmtModified != null">
       <if test="gmtModified != null">
           gmt_modified,
           gmt_modified,
       </if>
       </if>
+      <if test="partitionCount >0">
+          partition_count,
+      </if>
+      <if test="queueType >0">
+          queue_type,
+      </if>
       </trim>
       </trim>
     <trim prefix="values (" suffix=")" suffixOverrides=",">
     <trim prefix="values (" suffix=")" suffixOverrides=",">
         <if test="id != null">
         <if test="id != null">
@@ -144,6 +152,12 @@
       <if test="gmtModified != null">
       <if test="gmtModified != null">
           #{gmtModified,jdbcType=TIMESTAMP},
           #{gmtModified,jdbcType=TIMESTAMP},
       </if>
       </if>
+      <if test="partitionCount >0">
+           #{partitionCount,jdbcType=TIMESTAMP},
+      </if>
+        <if test="queueType >0">
+            #{partitionCount,jdbcType=TIMESTAMP},
+        </if>
     </trim>
     </trim>
   </insert>
   </insert>
   <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.domain.BusQueue">
   <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.domain.BusQueue">
@@ -180,6 +194,9 @@
           <if test="gmtModified != null">
           <if test="gmtModified != null">
               gmt_modified = #{gmtModified,jdbcType=TIMESTAMP},
               gmt_modified = #{gmtModified,jdbcType=TIMESTAMP},
           </if>
           </if>
+          <if test="partitionCount >0">
+              partition_count =  #{partitionCount,jdbcType=INTEGER},
+          </if>
       </set>
       </set>
       where id = #{id,jdbcType=VARCHAR}
       where id = #{id,jdbcType=VARCHAR}
   </update>
   </update>
@@ -213,14 +230,9 @@
                q.username,
                q.username,
                q.gmt_create,
                q.gmt_create,
                q.gmt_modified,
                q.gmt_modified,
-               v.virtual_host_name
-        from bus_queue as q,
-             virtual_host v
+               q.partition_count
+        from bus_queue as q
         <where>
         <where>
-            <!-- 所属虚拟主机Id -->
-            <if test="virtualHostId != null and virtualHostId != ''">
-                virtual_host_id = #{virtualHostId,jdbcType=VARCHAR}
-            </if>
             <!-- 用户名称 -->
             <!-- 用户名称 -->
             <if test="username != null and username != ''">
             <if test="username != null and username != ''">
                 and username = #{username,jdbcType=VARCHAR}
                 and username = #{username,jdbcType=VARCHAR}
@@ -229,7 +241,7 @@
             <if test="queueName != null and queueName != ''">
             <if test="queueName != null and queueName != ''">
                 and queue_name = #{queueName,jdbcType=VARCHAR}
                 and queue_name = #{queueName,jdbcType=VARCHAR}
             </if>
             </if>
-            and v.id = q.virtual_host_id
+            and server_type = 1
         </where>
         </where>
     </select>
     </select>
 
 

@@ -16,6 +16,7 @@
         <result column="description" jdbcType="VARCHAR" property="description"/>
         <result column="description" jdbcType="VARCHAR" property="description"/>
         <result column="gmt_create" jdbcType="TIMESTAMP" property="gmtCreate"/>
         <result column="gmt_create" jdbcType="TIMESTAMP" property="gmtCreate"/>
         <result column="gmt_modified" jdbcType="TIMESTAMP" property="gmtModified"/>
         <result column="gmt_modified" jdbcType="TIMESTAMP" property="gmtModified"/>
+        <result column="server_type" jdbcType="INTEGER" property="serverType"/>
     </resultMap>
     </resultMap>
 
 
     <!-- 该Mapper映射关系的作用，是服务器与虚拟主机的1:n的关系映射 -->
     <!-- 该Mapper映射关系的作用，是服务器与虚拟主机的1:n的关系映射 -->
@@ -46,7 +47,7 @@
   <sql id="Base_Column_List">
   <sql id="Base_Column_List">
       <!--@mbg.generated-->
       <!--@mbg.generated-->
       id, `server_name`, server_ip, server_port, client_port, super_username, super_password,
       id, `server_name`, server_ip, server_port, client_port, super_username, super_password,
-    description, gmt_create, gmt_modified
+    description, gmt_create, gmt_modified, server_type
   </sql>
   </sql>
   <select id="selectByPrimaryKey" parameterType="java.lang.String" resultMap="BaseResultMap">
   <select id="selectByPrimaryKey" parameterType="java.lang.String" resultMap="BaseResultMap">
       <!--@mbg.generated-->
       <!--@mbg.generated-->
@@ -106,6 +107,9 @@
       <if test="gmtModified != null">
       <if test="gmtModified != null">
           gmt_modified,
           gmt_modified,
       </if>
       </if>
+      <if test="serverType >0">
+          server_type,
+      </if>
       </trim>
       </trim>
     <trim prefix="values (" suffix=")" suffixOverrides=",">
     <trim prefix="values (" suffix=")" suffixOverrides=",">
         <if test="id != null">
         <if test="id != null">
@@ -138,6 +142,9 @@
       <if test="gmtModified != null">
       <if test="gmtModified != null">
           #{gmtModified,jdbcType=TIMESTAMP},
           #{gmtModified,jdbcType=TIMESTAMP},
       </if>
       </if>
+        <if test="serverType >0 ">
+            #{serverType,jdbcType=INTEGER},
+        </if>
     </trim>
     </trim>
   </insert>
   </insert>
   <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.domain.BusServer">
   <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.domain.BusServer">
@@ -247,6 +254,7 @@
             <if test="serverName != null and serverName != ''">
             <if test="serverName != null and serverName != ''">
                 server_name = #{serverName,jdbcType=VARCHAR}
                 server_name = #{serverName,jdbcType=VARCHAR}
             </if>
             </if>
+            and server_type =1
         </where>
         </where>
     </select>
     </select>
 
 

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.ConsumerGroupMapper" >
+  <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.model.ConsumerGroup" >
+    <id column="id" property="id" jdbcType="VARCHAR" />
+    <result column="name" property="name" jdbcType="VARCHAR" />
+    <result column="des" property="des" jdbcType="VARCHAR" />
+    <result column="user_id" property="userId" jdbcType="VARCHAR" />
+    <result column="username" property="username" jdbcType="VARCHAR" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, name, des, user_id, username
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select 
+    <include refid="Base_Column_List" />
+    from consumer_group
+    where id = #{id,jdbcType=VARCHAR}
+  </select>
+  <select id="groups" resultType="java.lang.String" parameterType="java.lang.String" >
+    select
+    name
+    from consumer_group
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
+    delete from consumer_group
+    where id = #{id,jdbcType=VARCHAR}
+  </delete>
+  <insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.ConsumerGroup" >
+    insert into consumer_group (id, name, des, 
+      user_id, username)
+    values (#{id,jdbcType=VARCHAR}, #{name,jdbcType=VARCHAR}, #{des,jdbcType=VARCHAR}, 
+      #{userId,jdbcType=VARCHAR}, #{username,jdbcType=VARCHAR})
+  </insert>
+  <insert id="insertSelective" parameterType="com.sunyo.wlpt.message.bus.service.model.ConsumerGroup" >
+    insert into consumer_group
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        id,
+      </if>
+      <if test="name != null" >
+        name,
+      </if>
+      <if test="des != null" >
+        des,
+      </if>
+      <if test="userId != null" >
+        user_id,
+      </if>
+      <if test="username != null" >
+        username,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=VARCHAR},
+      </if>
+      <if test="name != null" >
+        #{name,jdbcType=VARCHAR},
+      </if>
+      <if test="des != null" >
+        #{des,jdbcType=VARCHAR},
+      </if>
+      <if test="userId != null" >
+        #{userId,jdbcType=VARCHAR},
+      </if>
+      <if test="username != null" >
+        #{username,jdbcType=VARCHAR},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.model.ConsumerGroup" >
+    update consumer_group
+    <set >
+      <if test="name != null" >
+        name = #{name,jdbcType=VARCHAR},
+      </if>
+      <if test="des != null" >
+        des = #{des,jdbcType=VARCHAR},
+      </if>
+      <if test="userId != null" >
+        user_id = #{userId,jdbcType=VARCHAR},
+      </if>
+      <if test="username != null" >
+        username = #{username,jdbcType=VARCHAR},
+      </if>
+    </set>
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.sunyo.wlpt.message.bus.service.model.ConsumerGroup" >
+    update consumer_group
+    set name = #{name,jdbcType=VARCHAR},
+      des = #{des,jdbcType=VARCHAR},
+      user_id = #{userId,jdbcType=VARCHAR},
+      username = #{username,jdbcType=VARCHAR}
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+</mapper>

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.MessageRouterMapper" >
+  <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.model.MessageRouter" >
+    <id column="id" property="id" jdbcType="VARCHAR" />
+    <result column="sndr" property="sndr" jdbcType="VARCHAR" />
+    <result column="btype" property="btype" jdbcType="VARCHAR" />
+    <result column="stype" property="stype" jdbcType="VARCHAR" />
+    <result column="optype" property="optype" jdbcType="VARCHAR" />
+    <result column="msg_limit" property="msgLimit" jdbcType="TINYINT" />
+    <result column="character" property="character" jdbcType="VARCHAR" />
+    <result column="status" property="status" jdbcType="BIT" />
+    <result column="usage" property="usage" jdbcType="BIT" />
+    <result column="des" property="des" jdbcType="VARCHAR" />
+    <result column="ver" property="ver" jdbcType="VARCHAR" />
+    <result column="creat_time" property="creatTime" jdbcType="TIMESTAMP" />
+    <result column="update_time" property="updateTime" jdbcType="TIMESTAMP" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, sndr, btype, stype, optype, msg_limit, character, status, usage, des, ver, creat_time, 
+    update_time
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select 
+    <include refid="Base_Column_List" />
+    from message_router
+    where id = #{id,jdbcType=VARCHAR}
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
+    delete from message_router
+    where id = #{id,jdbcType=VARCHAR}
+  </delete>
+  <insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouter" >
+    insert into message_router (id, sndr, btype, 
+      stype, optype, msg_limit, 
+      character, status, usage, des, 
+      ver, creat_time, update_time
+      )
+    values (#{id,jdbcType=VARCHAR}, #{sndr,jdbcType=VARCHAR}, #{btype,jdbcType=VARCHAR}, 
+      #{stype,jdbcType=VARCHAR}, #{optype,jdbcType=VARCHAR}, #{msgLimit,jdbcType=TINYINT}, 
+      #{character,jdbcType=VARCHAR}, #{status,jdbcType=BIT}, #{usage,jdbcType=BIT}, #{des,jdbcType=VARCHAR}, 
+      #{ver,jdbcType=VARCHAR}, #{creatTime,jdbcType=TIMESTAMP}, #{updateTime,jdbcType=TIMESTAMP}
+      )
+  </insert>
+  <insert id="insertSelective" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouter" >
+    insert into message_router
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        id,
+      </if>
+      <if test="sndr != null" >
+        sndr,
+      </if>
+      <if test="btype != null" >
+        btype,
+      </if>
+      <if test="stype != null" >
+        stype,
+      </if>
+      <if test="optype != null" >
+        optype,
+      </if>
+      <if test="msgLimit != null" >
+        msg_limit,
+      </if>
+      <if test="character != null" >
+        character,
+      </if>
+      <if test="status != null" >
+        status,
+      </if>
+      <if test="usage != null" >
+        usage,
+      </if>
+      <if test="des != null" >
+        des,
+      </if>
+      <if test="ver != null" >
+        ver,
+      </if>
+      <if test="creatTime != null" >
+        creat_time,
+      </if>
+      <if test="updateTime != null" >
+        update_time,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=VARCHAR},
+      </if>
+      <if test="sndr != null" >
+        #{sndr,jdbcType=VARCHAR},
+      </if>
+      <if test="btype != null" >
+        #{btype,jdbcType=VARCHAR},
+      </if>
+      <if test="stype != null" >
+        #{stype,jdbcType=VARCHAR},
+      </if>
+      <if test="optype != null" >
+        #{optype,jdbcType=VARCHAR},
+      </if>
+      <if test="msgLimit != null" >
+        #{msgLimit,jdbcType=TINYINT},
+      </if>
+      <if test="character != null" >
+        #{character,jdbcType=VARCHAR},
+      </if>
+      <if test="status != null" >
+        #{status,jdbcType=BIT},
+      </if>
+      <if test="usage != null" >
+        #{usage,jdbcType=BIT},
+      </if>
+      <if test="des != null" >
+        #{des,jdbcType=VARCHAR},
+      </if>
+      <if test="ver != null" >
+        #{ver,jdbcType=VARCHAR},
+      </if>
+      <if test="creatTime != null" >
+        #{creatTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="updateTime != null" >
+        #{updateTime,jdbcType=TIMESTAMP},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouter" >
+    update message_router
+    <set >
+      <if test="sndr != null" >
+        sndr = #{sndr,jdbcType=VARCHAR},
+      </if>
+      <if test="btype != null" >
+        btype = #{btype,jdbcType=VARCHAR},
+      </if>
+      <if test="stype != null" >
+        stype = #{stype,jdbcType=VARCHAR},
+      </if>
+      <if test="optype != null" >
+        optype = #{optype,jdbcType=VARCHAR},
+      </if>
+      <if test="msgLimit != null" >
+        msg_limit = #{msgLimit,jdbcType=TINYINT},
+      </if>
+      <if test="character != null" >
+        character = #{character,jdbcType=VARCHAR},
+      </if>
+      <if test="status != null" >
+        status = #{status,jdbcType=BIT},
+      </if>
+      <if test="usage != null" >
+        usage = #{usage,jdbcType=BIT},
+      </if>
+      <if test="des != null" >
+        des = #{des,jdbcType=VARCHAR},
+      </if>
+      <if test="ver != null" >
+        ver = #{ver,jdbcType=VARCHAR},
+      </if>
+      <if test="creatTime != null" >
+        creat_time = #{creatTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="updateTime != null" >
+        update_time = #{updateTime,jdbcType=TIMESTAMP},
+      </if>
+    </set>
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouter" >
+    update message_router
+    set sndr = #{sndr,jdbcType=VARCHAR},
+      btype = #{btype,jdbcType=VARCHAR},
+      stype = #{stype,jdbcType=VARCHAR},
+      optype = #{optype,jdbcType=VARCHAR},
+      msg_limit = #{msgLimit,jdbcType=TINYINT},
+      character = #{character,jdbcType=VARCHAR},
+      status = #{status,jdbcType=BIT},
+      usage = #{usage,jdbcType=BIT},
+      des = #{des,jdbcType=VARCHAR},
+      ver = #{ver,jdbcType=VARCHAR},
+      creat_time = #{creatTime,jdbcType=TIMESTAMP},
+      update_time = #{updateTime,jdbcType=TIMESTAMP}
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+</mapper>

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.MessageRouterReciverFilterMapper" >
+  <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciverFilter" >
+    <id column="id" property="id" jdbcType="VARCHAR" />
+    <result column="filter" property="filter" jdbcType="VARCHAR" />
+    <result column="filter_value" property="filterValue" jdbcType="VARCHAR" />
+    <result column="type" property="type" jdbcType="VARCHAR" />
+    <result column="status" property="status" jdbcType="BIT" />
+    <result column="message_router_reciver_id" property="messageRouterReciverId" jdbcType="VARCHAR" />
+    <result column="creat_time" property="creatTime" jdbcType="TIMESTAMP" />
+    <result column="update_time" property="updateTime" jdbcType="TIMESTAMP" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, filter, filter_value, type, status, message_router_reciver_id, creat_time, update_time
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select 
+    <include refid="Base_Column_List" />
+    from message_router_reciver_filter
+    where id = #{id,jdbcType=VARCHAR}
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
+    delete from message_router_reciver_filter
+    where id = #{id,jdbcType=VARCHAR}
+  </delete>
+  <insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciverFilter" >
+    insert into message_router_reciver_filter (id, filter, filter_value, 
+      type, status, message_router_reciver_id, 
+      creat_time, update_time)
+    values (#{id,jdbcType=VARCHAR}, #{filter,jdbcType=VARCHAR}, #{filterValue,jdbcType=VARCHAR}, 
+      #{type,jdbcType=VARCHAR}, #{status,jdbcType=BIT}, #{messageRouterReciverId,jdbcType=VARCHAR}, 
+      #{creatTime,jdbcType=TIMESTAMP}, #{updateTime,jdbcType=TIMESTAMP})
+  </insert>
+  <insert id="insertSelective" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciverFilter" >
+    insert into message_router_reciver_filter
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        id,
+      </if>
+      <if test="filter != null" >
+        filter,
+      </if>
+      <if test="filterValue != null" >
+        filter_value,
+      </if>
+      <if test="type != null" >
+        type,
+      </if>
+      <if test="status != null" >
+        status,
+      </if>
+      <if test="messageRouterReciverId != null" >
+        message_router_reciver_id,
+      </if>
+      <if test="creatTime != null" >
+        creat_time,
+      </if>
+      <if test="updateTime != null" >
+        update_time,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=VARCHAR},
+      </if>
+      <if test="filter != null" >
+        #{filter,jdbcType=VARCHAR},
+      </if>
+      <if test="filterValue != null" >
+        #{filterValue,jdbcType=VARCHAR},
+      </if>
+      <if test="type != null" >
+        #{type,jdbcType=VARCHAR},
+      </if>
+      <if test="status != null" >
+        #{status,jdbcType=BIT},
+      </if>
+      <if test="messageRouterReciverId != null" >
+        #{messageRouterReciverId,jdbcType=VARCHAR},
+      </if>
+      <if test="creatTime != null" >
+        #{creatTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="updateTime != null" >
+        #{updateTime,jdbcType=TIMESTAMP},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciverFilter" >
+    update message_router_reciver_filter
+    <set >
+      <if test="filter != null" >
+        filter = #{filter,jdbcType=VARCHAR},
+      </if>
+      <if test="filterValue != null" >
+        filter_value = #{filterValue,jdbcType=VARCHAR},
+      </if>
+      <if test="type != null" >
+        type = #{type,jdbcType=VARCHAR},
+      </if>
+      <if test="status != null" >
+        status = #{status,jdbcType=BIT},
+      </if>
+      <if test="messageRouterReciverId != null" >
+        message_router_reciver_id = #{messageRouterReciverId,jdbcType=VARCHAR},
+      </if>
+      <if test="creatTime != null" >
+        creat_time = #{creatTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="updateTime != null" >
+        update_time = #{updateTime,jdbcType=TIMESTAMP},
+      </if>
+    </set>
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciverFilter" >
+    update message_router_reciver_filter
+    set filter = #{filter,jdbcType=VARCHAR},
+      filter_value = #{filterValue,jdbcType=VARCHAR},
+      type = #{type,jdbcType=VARCHAR},
+      status = #{status,jdbcType=BIT},
+      message_router_reciver_id = #{messageRouterReciverId,jdbcType=VARCHAR},
+      creat_time = #{creatTime,jdbcType=TIMESTAMP},
+      update_time = #{updateTime,jdbcType=TIMESTAMP}
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+</mapper>

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.MessageRouterReciverMapper" >
+  <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciver" >
+    <id column="id" property="id" jdbcType="VARCHAR" />
+    <result column="rcvr_topic" property="rcvrTopic" jdbcType="VARCHAR" />
+    <result column="message_router_id" property="messageRouterId" jdbcType="VARCHAR" />
+    <result column="creat_time" property="creatTime" jdbcType="TIMESTAMP" />
+    <result column="update_time" property="updateTime" jdbcType="TIMESTAMP" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, rcvr_topic, message_router_id, creat_time, update_time
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select 
+    <include refid="Base_Column_List" />
+    from message_router_reciver
+    where id = #{id,jdbcType=VARCHAR}
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
+    delete from message_router_reciver
+    where id = #{id,jdbcType=VARCHAR}
+  </delete>
+  <insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciver" >
+    insert into message_router_reciver (id, rcvr_topic, message_router_id, 
+      creat_time, update_time)
+    values (#{id,jdbcType=VARCHAR}, #{rcvrTopic,jdbcType=VARCHAR}, #{messageRouterId,jdbcType=VARCHAR}, 
+      #{creatTime,jdbcType=TIMESTAMP}, #{updateTime,jdbcType=TIMESTAMP})
+  </insert>
+  <insert id="insertSelective" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciver" >
+    insert into message_router_reciver
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        id,
+      </if>
+      <if test="rcvrTopic != null" >
+        rcvr_topic,
+      </if>
+      <if test="messageRouterId != null" >
+        message_router_id,
+      </if>
+      <if test="creatTime != null" >
+        creat_time,
+      </if>
+      <if test="updateTime != null" >
+        update_time,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=VARCHAR},
+      </if>
+      <if test="rcvrTopic != null" >
+        #{rcvrTopic,jdbcType=VARCHAR},
+      </if>
+      <if test="messageRouterId != null" >
+        #{messageRouterId,jdbcType=VARCHAR},
+      </if>
+      <if test="creatTime != null" >
+        #{creatTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="updateTime != null" >
+        #{updateTime,jdbcType=TIMESTAMP},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciver" >
+    update message_router_reciver
+    <set >
+      <if test="rcvrTopic != null" >
+        rcvr_topic = #{rcvrTopic,jdbcType=VARCHAR},
+      </if>
+      <if test="messageRouterId != null" >
+        message_router_id = #{messageRouterId,jdbcType=VARCHAR},
+      </if>
+      <if test="creatTime != null" >
+        creat_time = #{creatTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="updateTime != null" >
+        update_time = #{updateTime,jdbcType=TIMESTAMP},
+      </if>
+    </set>
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciver" >
+    update message_router_reciver
+    set rcvr_topic = #{rcvrTopic,jdbcType=VARCHAR},
+      message_router_id = #{messageRouterId,jdbcType=VARCHAR},
+      creat_time = #{creatTime,jdbcType=TIMESTAMP},
+      update_time = #{updateTime,jdbcType=TIMESTAMP}
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+</mapper>

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.MessageTypeMapper" >
+  <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.model.MessageType" >
+    <id column="ID" property="id" jdbcType="VARCHAR" />
+    <result column="name" property="name" jdbcType="VARCHAR" />
+    <result column="parent_id" property="parentId" jdbcType="VARCHAR" />
+    <result column="des" property="des" jdbcType="VARCHAR" />
+    <result column="type" property="type" jdbcType="INTEGER" />
+    <collection column="ID" property="children" select="selectByParentId" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    ID, name, parent_id, des, type
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select 
+    <include refid="Base_Column_List" />
+    from message_type
+    where ID = #{id,jdbcType=VARCHAR}
+  </select>
+  <select id="selectByParentId" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select
+    <include refid="Base_Column_List" />
+    from message_type
+    where parent_id = #{parentId,jdbcType=VARCHAR}
+  </select>
+  <select id="selectAll" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageType" resultMap="BaseResultMap">
+    select
+    <include refid="Base_Column_List" />
+    from message_type
+    <where>
+        parent_id = 0
+      <if test="id != null and id != ''">
+        and ID = #{id,jdbcType=VARCHAR}
+      </if>
+      <if test="name != null and name != ''">
+       and  name = #{name,jdbcType=VARCHAR}
+      </if>
+      <if test="type !=null  and type > 0">
+        and type = #{type,jdbcType=INTEGER}
+      </if>
+    </where>
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
+    delete from message_type
+    where ID = #{id,jdbcType=VARCHAR}
+  </delete>
+  <insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageType" >
+    insert into message_type (ID, name, parent_id, 
+      des, type)
+    values (#{id,jdbcType=VARCHAR}, #{name,jdbcType=VARCHAR}, #{parentId,jdbcType=VARCHAR}, 
+      #{des,jdbcType=VARCHAR}, #{type,jdbcType=INTEGER})
+  </insert>
+  <insert id="insertSelective" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageType" >
+    insert into message_type
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        ID,
+      </if>
+      <if test="name != null" >
+        name,
+      </if>
+      <if test="parentId != null" >
+        parent_id,
+      </if>
+      <if test="des != null" >
+        des,
+      </if>
+      <if test="type != null" >
+        type,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=VARCHAR},
+      </if>
+      <if test="name != null" >
+        #{name,jdbcType=VARCHAR},
+      </if>
+      <if test="parentId != null" >
+        #{parentId,jdbcType=VARCHAR},
+      </if>
+      <if test="des != null" >
+        #{des,jdbcType=VARCHAR},
+      </if>
+      <if test="type != null" >
+        #{type,jdbcType=INTEGER},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageType" >
+    update message_type
+    <set >
+      <if test="name != null" >
+        name = #{name,jdbcType=VARCHAR},
+      </if>
+      <if test="parentId != null" >
+        parent_id = #{parentId,jdbcType=VARCHAR},
+      </if>
+      <if test="des != null" >
+        des = #{des,jdbcType=VARCHAR},
+      </if>
+      <if test="type != null" >
+        type = #{type,jdbcType=INTEGER},
+      </if>
+    </set>
+    where ID = #{id,jdbcType=VARCHAR}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageType" >
+    update message_type
+    set name = #{name,jdbcType=VARCHAR},
+      parent_id = #{parentId,jdbcType=VARCHAR},
+      des = #{des,jdbcType=VARCHAR},
+      type = #{type,jdbcType=INTEGER}
+    where ID = #{id,jdbcType=VARCHAR}
+  </update>
+</mapper>

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.StypeMapper" >
+  <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.model.Stype" >
+    <id column="id" property="id" jdbcType="VARCHAR" />
+    <result column="name" property="name" jdbcType="VARCHAR" />
+    <result column="des" property="des" jdbcType="VARCHAR" />
+    <result column="parent_id" property="parentId" jdbcType="VARCHAR" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, name, des, parent_id
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select 
+    <include refid="Base_Column_List" />
+    from stype
+    where id = #{id,jdbcType=VARCHAR}
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
+    delete from stype
+    where id = #{id,jdbcType=VARCHAR}
+  </delete>
+  <insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.Stype" >
+    insert into stype (id, name, des, 
+      parent_id)
+    values (#{id,jdbcType=VARCHAR}, #{name,jdbcType=VARCHAR}, #{des,jdbcType=VARCHAR}, 
+      #{parentId,jdbcType=VARCHAR})
+  </insert>
+  <insert id="insertSelective" parameterType="com.sunyo.wlpt.message.bus.service.model.Stype" >
+    insert into stype
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        id,
+      </if>
+      <if test="name != null" >
+        name,
+      </if>
+      <if test="des != null" >
+        des,
+      </if>
+      <if test="parentId != null" >
+        parent_id,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=VARCHAR},
+      </if>
+      <if test="name != null" >
+        #{name,jdbcType=VARCHAR},
+      </if>
+      <if test="des != null" >
+        #{des,jdbcType=VARCHAR},
+      </if>
+      <if test="parentId != null" >
+        #{parentId,jdbcType=VARCHAR},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.model.Stype" >
+    update stype
+    <set >
+      <if test="name != null" >
+        name = #{name,jdbcType=VARCHAR},
+      </if>
+      <if test="des != null" >
+        des = #{des,jdbcType=VARCHAR},
+      </if>
+      <if test="parentId != null" >
+        parent_id = #{parentId,jdbcType=VARCHAR},
+      </if>
+    </set>
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.sunyo.wlpt.message.bus.service.model.Stype" >
+    update stype
+    set name = #{name,jdbcType=VARCHAR},
+      des = #{des,jdbcType=VARCHAR},
+      parent_id = #{parentId,jdbcType=VARCHAR}
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+</mapper>

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.sunyo.wlpt.message.bus.service.mapper.UserTopicMapper" >
+  <resultMap id="BaseResultMap" type="com.sunyo.wlpt.message.bus.service.model.UserTopic" >
+    <id column="id" property="id" jdbcType="VARCHAR" />
+    <result column="user_id" property="userId" jdbcType="VARCHAR" />
+    <result column="bus_queue_id" property="busQueueId" jdbcType="VARCHAR" />
+    <result column="username" property="username" jdbcType="VARCHAR" />
+    <result column="topic" property="topic" jdbcType="VARCHAR" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, user_id, bus_queue_id, username, topic
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >
+    select 
+    <include refid="Base_Column_List" />
+    from user_topic
+    where id = #{id,jdbcType=VARCHAR}
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
+    delete from user_topic
+    where id = #{id,jdbcType=VARCHAR}
+  </delete>
+  <insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.UserTopic" >
+    insert into user_topic (id, user_id, bus_queue_id, 
+      username, topic)
+    values (#{id,jdbcType=VARCHAR}, #{userId,jdbcType=VARCHAR}, #{busQueueId,jdbcType=VARCHAR}, 
+      #{username,jdbcType=VARCHAR}, #{topic,jdbcType=VARCHAR})
+  </insert>
+  <insert id="insertSelective" parameterType="com.sunyo.wlpt.message.bus.service.model.UserTopic" >
+    insert into user_topic
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        id,
+      </if>
+      <if test="userId != null" >
+        user_id,
+      </if>
+      <if test="busQueueId != null" >
+        bus_queue_id,
+      </if>
+      <if test="username != null" >
+        username,
+      </if>
+      <if test="topic != null" >
+        topic,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=VARCHAR},
+      </if>
+      <if test="userId != null" >
+        #{userId,jdbcType=VARCHAR},
+      </if>
+      <if test="busQueueId != null" >
+        #{busQueueId,jdbcType=VARCHAR},
+      </if>
+      <if test="username != null" >
+        #{username,jdbcType=VARCHAR},
+      </if>
+      <if test="topic != null" >
+        #{topic,jdbcType=VARCHAR},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.sunyo.wlpt.message.bus.service.model.UserTopic" >
+    update user_topic
+    <set >
+      <if test="userId != null" >
+        user_id = #{userId,jdbcType=VARCHAR},
+      </if>
+      <if test="busQueueId != null" >
+        bus_queue_id = #{busQueueId,jdbcType=VARCHAR},
+      </if>
+      <if test="username != null" >
+        username = #{username,jdbcType=VARCHAR},
+      </if>
+      <if test="topic != null" >
+        topic = #{topic,jdbcType=VARCHAR},
+      </if>
+    </set>
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.sunyo.wlpt.message.bus.service.model.UserTopic" >
+    update user_topic
+    set user_id = #{userId,jdbcType=VARCHAR},
+      bus_queue_id = #{busQueueId,jdbcType=VARCHAR},
+      username = #{username,jdbcType=VARCHAR},
+      topic = #{topic,jdbcType=VARCHAR}
+    where id = #{id,jdbcType=VARCHAR}
+  </update>
+</mapper>