项目初始化

datax-common/datax-common-core/src/main/java/cn/datax/common/core/DataConstant.java

@@ -206,8 +206,8 @@ public class DataConstant {
      */
     public enum ApiState{
         WAIT("1","待发布"),
-        PUBLISHED("2","已发布"),
-        OFFLINE("3","已下线");
+        RELEASE("2","已发布"),
+        CANCEL("3","已下线");
         ApiState(String key, String val){
             this.key = key;
             this.val = val;

datax-common/datax-common-rabbitmq/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-common</artifactId>
+        <groupId>cn.datax</groupId>
+        <version>2.0.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <version>2.0.0</version>
+    <artifactId>datax-common-rabbitmq</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
+    </dependencies>
+</project>
\ No newline at end of file

datax-common/datax-common-rabbitmq/src/main/java/cn/datax/common/rabbitmq/config/RabbitMqConfig.java

+package cn.datax.common.rabbitmq.config;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.context.annotation.Bean;
+
+public class RabbitMqConfig implements BeanPostProcessor {
+
+    @Autowired
+    private RabbitAdmin rabbitAdmin;
+
+    @Bean
+    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
+        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
+        // 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
+        rabbitAdmin.setAutoStartup(true);
+        return rabbitAdmin;
+    }
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        rabbitAdmin.declareExchange(rabbitmqApiReleaseFanoutExchange());
+        rabbitAdmin.declareQueue(fanoutExchangeQueueApiRelease());
+        rabbitAdmin.declareExchange(rabbitmqApiCancelFanoutExchange());
+        rabbitAdmin.declareQueue(fanoutExchangeQueueApiCancel());
+        return null;
+    }
+
+    /**
+     * 队列API发布
+     * @return
+     */
+    @Bean
+    public Queue fanoutExchangeQueueApiRelease() {
+        /**
+         * 1、name:    队列名称
+         * 2、durable: 是否持久化
+         * 3、exclusive: 是否独享、排外的。如果设置为true，定义为排他队列。则只有创建者可以使用此队列。也就是private私有的。
+         * 4、autoDelete: 是否自动删除。也就是临时队列。当最后一个消费者断开连接后，会自动删除。
+         * */
+        return new Queue(RabbitMqConstant.FANOUT_EXCHANGE_QUEUE_TOPIC_API_RELEASE, true, false, false);
+    }
+
+    /**
+     * 创建队列API发布FanoutExchange类型交换机
+     * @return
+     */
+    @Bean
+    public FanoutExchange rabbitmqApiReleaseFanoutExchange() {
+        return new FanoutExchange(RabbitMqConstant.FANOUT_EXCHANGE_API_RELEASE_NAME, true, false);
+    }
+
+    /**
+     * 队列API发布绑定到FanoutExchange交换机
+     * @return
+     */
+    @Bean
+    public Binding bindFanoutApiRelease() {
+        return BindingBuilder.bind(fanoutExchangeQueueApiRelease()).to(rabbitmqApiReleaseFanoutExchange());
+    }
+
+    /**
+     * 队列API注销
+     * @return
+     */
+    @Bean
+    public Queue fanoutExchangeQueueApiCancel() {
+        /**
+         * 1、name:    队列名称
+         * 2、durable: 是否持久化
+         * 3、exclusive: 是否独享、排外的。如果设置为true，定义为排他队列。则只有创建者可以使用此队列。也就是private私有的。
+         * 4、autoDelete: 是否自动删除。也就是临时队列。当最后一个消费者断开连接后，会自动删除。
+         * */
+        return new Queue(RabbitMqConstant.FANOUT_EXCHANGE_QUEUE_TOPIC_API_CANCEL, true, false, false);
+    }
+
+    /**
+     * 创建队列API注销FanoutExchange类型交换机
+     * @return
+     */
+    @Bean
+    public FanoutExchange rabbitmqApiCancelFanoutExchange() {
+        return new FanoutExchange(RabbitMqConstant.FANOUT_EXCHANGE_API_CANCEL_NAME, true, false);
+    }
+
+    /**
+     * 队列API注销绑定到FanoutExchange交换机
+     * @return
+     */
+    @Bean
+    public Binding bindFanoutApiCancel() {
+        return BindingBuilder.bind(fanoutExchangeQueueApiCancel()).to(rabbitmqApiCancelFanoutExchange());
+    }
+}

datax-common/datax-common-rabbitmq/src/main/java/cn/datax/common/rabbitmq/config/RabbitMqConstant.java

+package cn.datax.common.rabbitmq.config;
+
+public class RabbitMqConstant {
+
+    /**
+     * RabbitMQ的FANOUT_EXCHANG交换机类型的队列API发布的名称
+     */
+    public static final String FANOUT_EXCHANGE_QUEUE_TOPIC_API_RELEASE = "fanout.api.release";
+
+    /**
+     * RabbitMQ的FANOUT_EXCHANG交换机类型的名称
+     */
+    public static final String FANOUT_EXCHANGE_API_RELEASE_NAME = "fanout.exchange.api.release.name";
+
+    /**
+     * RabbitMQ的FANOUT_EXCHANG交换机类型的队列API注销的名称
+     */
+    public static final String FANOUT_EXCHANGE_QUEUE_TOPIC_API_CANCEL = "fanout.api.cancel";
+
+    /**
+     * RabbitMQ的FANOUT_EXCHANG交换机类型的名称
+     */
+    public static final String FANOUT_EXCHANGE_API_CANCEL_NAME = "fanout.exchange.api.cancel.name";
+}

datax-common/datax-common-rabbitmq/src/main/resources/META-INF/spring.factories

+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+  cn.datax.common.rabbitmq.config.RabbitMqConfig

datax-common/pom.xml

@@ -23,5 +23,6 @@
         <module>datax-common-dictionary</module>
         <module>datax-common-qrcode</module>
         <module>datax-common-jasperreport</module>
+        <module>datax-common-rabbitmq</module>
     </modules>
 </project>
\ No newline at end of file

datax-config/src/main/resources/config/datax-service-data-market-dev.yml

@@ -12,6 +12,11 @@ spring:
         max-wait: -1ms    # 连接池最大阻塞等待时间（使用负值表示没有限制）
         max-idle: 10      # 连接池中的最大空闲连接
         min-idle: 5       # 连接池中的最小空闲连接
+  rabbitmq:
+    host: 192.168.234.106
+    port: 5672
+    username: admin
+    password: 1234@abcd
   datasource:
     dynamic:
       type: com.zaxxer.hikari.HikariDataSource

datax-modules/data-market-service-parent/data-market-service/pom.xml

@@ -79,6 +79,11 @@
             <artifactId>data-factory-service-api</artifactId>
             <version>2.0.0</version>
         </dependency>
+        <dependency>
+            <groupId>cn.datax</groupId>
+            <artifactId>datax-common-rabbitmq</artifactId>
+            <version>2.0.0</version>
+        </dependency>
     </dependencies>
 
     <build>

datax-modules/data-market-service-parent/data-market-service/src/main/java/cn/datax/service/data/market/config/StartedUpRunner.java

@@ -40,12 +40,12 @@ public class StartedUpRunner implements ApplicationRunner {
                     "-----------------------------------------";
             System.out.println(banner);
             // 项目启动时，初始化已发布的接口
-            QueryWrapper<DataApiEntity> queryWrapper = new QueryWrapper<>();
-            queryWrapper.eq("status", DataConstant.ApiState.PUBLISHED.getKey());
-            List<DataApiEntity> dataApiEntityList = dataApiDao.selectList(queryWrapper);
-            if (CollUtil.isNotEmpty(dataApiEntityList)) {
-                dataApiEntityList.stream().forEach(api -> mappingHandlerMapping.registerMapping(api));
-            }
+//            QueryWrapper<DataApiEntity> queryWrapper = new QueryWrapper<>();
+//            queryWrapper.eq("status", DataConstant.ApiState.RELEASE.getKey());
+//            List<DataApiEntity> dataApiEntityList = dataApiDao.selectList(queryWrapper);
+//            if (CollUtil.isNotEmpty(dataApiEntityList)) {
+//                dataApiEntityList.stream().forEach(api -> mappingHandlerMapping.registerMapping(api));
+//            }
         }
     }
 }

datax-modules/data-market-service-parent/data-market-service/src/main/java/cn/datax/service/data/market/controller/DataApiController.java

@@ -154,9 +154,36 @@ public class DataApiController extends BaseController {
         return R.ok().setData(sqlParseVo);
     }
 
+    /**
+     * 拷贝接口
+     * @param id
+     * @return
+     */
     @PostMapping("/{id}/copy")
     public R copyDataApi(@PathVariable String id) {
         dataApiService.copyDataApi(id);
         return R.ok();
     }
+
+    /**
+     * 发布接口
+     * @param id
+     * @return
+     */
+    @PostMapping(value = "/{id}/release")
+    public R releaseDataApi(@PathVariable String id){
+        dataApiService.releaseDataApi(id);
+        return R.ok();
+    }
+
+    /**
+     * 注销接口
+     * @param id
+     * @return
+     */
+    @PostMapping(value = "/{id}/cancel")
+    public R cancelDataApi(@PathVariable String id){
+        dataApiService.cancelDataApi(id);
+        return R.ok();
+    }
 }

datax-modules/data-market-service-parent/data-market-service/src/main/java/cn/datax/service/data/market/service/DataApiService.java

@@ -31,4 +31,8 @@ public interface DataApiService extends BaseService<DataApiEntity> {
     SqlParseVo sqlParse(SqlParseDto sqlParseDto);
 
     void copyDataApi(String id);
+
+    void releaseDataApi(String id);
+
+    void cancelDataApi(String id);
 }

datax-modules/data-market-service-parent/data-market-service/src/main/java/cn/datax/service/data/market/service/impl/ApiExecuteServiceImpl.java

@@ -40,7 +40,7 @@ public class ApiExecuteServiceImpl implements ApiExecuteService {
         DataApiEntity dataApiEntity = dataApiDao.selectById(id);
         if (dataApiEntity != null) {
             mappingHandlerMapping.registerMapping(dataApiEntity);
-            dataApiEntity.setStatus(DataConstant.ApiState.PUBLISHED.getKey());
+            dataApiEntity.setStatus(DataConstant.ApiState.RELEASE.getKey());
             dataApiDao.updateById(dataApiEntity);
         }
     }
@@ -50,7 +50,7 @@ public class ApiExecuteServiceImpl implements ApiExecuteService {
         DataApiEntity dataApiEntity = dataApiDao.selectById(id);
         if (dataApiEntity != null) {
             mappingHandlerMapping.unregisterMapping(dataApiEntity);
-            dataApiEntity.setStatus(DataConstant.ApiState.OFFLINE.getKey());
+            dataApiEntity.setStatus(DataConstant.ApiState.CANCEL.getKey());
             dataApiDao.updateById(dataApiEntity);
         }
     }

datax-modules/data-market-service-parent/data-market-service/src/main/java/cn/datax/service/data/market/service/impl/DataApiServiceImpl.java

@@ -2,6 +2,7 @@ package cn.datax.service.data.market.service.impl;
 
 import cn.datax.common.core.DataConstant;
 import cn.datax.common.exception.DataException;
+import cn.datax.common.rabbitmq.config.RabbitMqConstant;
 import cn.datax.common.utils.ThrowableUtil;
 import cn.datax.service.data.market.api.dto.*;
 import cn.datax.service.data.market.api.entity.DataApiEntity;
@@ -12,6 +13,8 @@ import cn.datax.service.data.market.mapstruct.DataApiMapper;
 import cn.datax.service.data.market.dao.DataApiDao;
 import cn.datax.common.base.BaseServiceImpl;
 import cn.datax.service.data.market.utils.SqlBuilderUtil;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
 import net.sf.jsqlparser.JSQLParserException;
 import net.sf.jsqlparser.expression.ExpressionVisitorAdapter;
@@ -27,6 +30,7 @@ import net.sf.jsqlparser.statement.select.*;
 import net.sf.jsqlparser.util.SelectUtils;
 import net.sf.jsqlparser.util.deparser.ExpressionDeParser;
 import net.sf.jsqlparser.util.deparser.SelectDeParser;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
@@ -56,6 +60,12 @@ public class DataApiServiceImpl extends BaseServiceImpl<DataApiDao, DataApiEntit
     @Autowired
     private DataApiMapper dataApiMapper;
 
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    @Autowired
+    private ObjectMapper objectMapper;
+
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void saveDataApi(DataApiDto dataApiDto) {
@@ -216,4 +226,20 @@ public class DataApiServiceImpl extends BaseServiceImpl<DataApiDao, DataApiEntit
         dataApiEntity.setStatus(DataConstant.ApiState.WAIT.getKey());
         dataApiDao.insert(dataApiEntity);
     }
+
+    @Override
+    public void releaseDataApi(String id) {
+        DataApiEntity dataApiEntity = super.getById(id);
+        try {
+            rabbitTemplate.convertAndSend(RabbitMqConstant.FANOUT_EXCHANGE_API_RELEASE_NAME, "", objectMapper.writeValueAsString(dataApiEntity));
+        } catch (JsonProcessingException e) {}
+    }
+
+    @Override
+    public void cancelDataApi(String id) {
+        DataApiEntity dataApiEntity = super.getById(id);
+        try {
+            rabbitTemplate.convertAndSend(RabbitMqConstant.FANOUT_EXCHANGE_API_CANCEL_NAME, "", objectMapper.writeValueAsString(dataApiEntity));
+        } catch (JsonProcessingException e) {}
+    }
 }

datax-ui/src/api/market/dataapi.js

@@ -68,6 +68,20 @@ export function copyDataApi (id) {
   })
 }
 
+export function releaseDataApi (id) {
+  return request({
+    url: '/data/market/dataApis/' + id + '/release',
+    method: 'post'
+  })
+}
+
+export function cancelDataApi (id) {
+  return request({
+    url: '/data/market/dataApis/' + id + '/cancel',
+    method: 'post'
+  })
+}
+
 export function getApiHeader (id) {
   return request({
     url: '/data/market/apis/' + id + '/header',

datax-ui/src/views/market/dataapi/DataApiList.vue

@@ -165,7 +165,7 @@
                 icon="el-icon-download"
                 @click="handleUnRegister(scope.row)"
                 v-hasPerm="['market:dataapi:unregister']"
-              >下线</el-button>
+              >注销</el-button>
               <el-button slot="reference">操作</el-button>
             </el-popover>
           </template>
@@ -186,7 +186,7 @@
 </template>
 
 <script>
-import { pageDataApi, delDataApi, delDataApis, copyDataApi, apiRegister, apiUnRegister } from '@/api/market/dataapi'
+import { pageDataApi, delDataApi, delDataApis, copyDataApi, releaseDataApi, cancelDataApi, apiRegister, apiUnRegister } from '@/api/market/dataapi'
 
 export default {
   name: 'DataApiList',
@@ -356,18 +356,18 @@ export default {
     },
     /** 接口发布 */
     handleRegister (row) {
-      apiRegister(row.id).then(response => {
+      releaseDataApi(row.id).then(response => {
         if (response.success) {
           this.$message.success('接口发布成功')
           this.getList()
         }
       })
     },
-    /** 接口下线 */
+    /** 接口注销 */
     handleUnRegister (row) {
-      apiUnRegister(row.id).then(response => {
+      cancelDataApi(row.id).then(response => {
         if (response.success) {
-          this.$message.success('接口下线成功')
+          this.$message.success('接口注销成功')
           this.getList()
         }
       })