版本2.0更新

ddp-runner-api/src/main/java/com/fibo/ddp/runner/api/JarDdpRunnerApiApplication.java

@@ -0,0 +1,21 @@
+package com.fibo.ddp.runner.api;
+
+import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
+
+@SpringBootApplication
+@EnableTransactionManagement
+@MapperScan({"com.fibo.ddp.common.dao.**"})
+@ComponentScan(basePackages = "com.fibo.ddp.**")
+@EnableAsync
+public class JarDdpRunnerApiApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(JarDdpRunnerApiApplication.class, args);
+    }
+
+}

ddp-runner-api/src/main/java/com/fibo/ddp/runner/api/config/ConfigHolder.java

@@ -0,0 +1,72 @@
+package com.fibo.ddp.runner.api.config;
+
+import lombok.Data;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@Data
+public class ConfigHolder {
+
+    //redisConfig
+    @Value("${redis.host}")
+    private String redisHost;
+    @Value("${redis.port}")
+    private int redisPort;
+    @Value("${redis.db}")
+    private int redisDb;
+    @Value("${redis.password}")
+    private String redisPwd;
+    @Value("${redis.pool.maxTotal}")
+    private int redisMaxTotal;
+    @Value("${redis.pool.maxIdle}")
+    private int redisMaxIdle;
+    @Value("${redis.pool.maxWait}")
+    private int redisMaxWait;
+    @Value("${redis.pool.timeout}")
+    private int redisTimeout;
+
+    // 业务逻辑是否使用缓存
+    @Value("${switch.use.cache}")
+    private String cacheSwitch;
+    // canal缓存同步是否开启
+    @Value("${switch.canal.cache}")
+    private String canalCacheSwitch;
+    // canal主机地址
+    @Value("${canal.hostname}")
+    private String canalHostName;
+    // canal端口
+    @Value("${canal.port}")
+    private int canalPort;
+
+    //jdbcConfig
+    /*@Value("${jdbc.url}")
+    private String jdbcUrl;
+    @Value("${jdbc.driver}")
+    private String DriverName;
+    @Value("${pool.maxPoolSize}")
+    private int maxPoolSize;
+    @Value("${jdbc.username}")
+    private String jdbcUserName;
+    @Value("${jdbc.password}")
+    private String jdbcPwd;
+    @Value("${pool.maxWait}")
+    private int jdbcMaxWait;
+    @Value("${pool.timeBetweenEvictionRunsMillis}")
+    private int timeBetweenEvictionRunsMillis;
+    @Value("${pool.minEvictableIdleTimeMillis}")
+    private int minEvictableIdleTimeMillis;
+    @Value("${pool.validationQuery}")
+    private String validationQuery;
+
+    //rabbitconfig
+    @Value("${rabbitMQ.host}")
+    private String rabbitHost;
+    @Value("${rabbitMQ.port}")
+    private int rabbitPort;
+    @Value("${rabbitMQ.username}")
+    private String rabbitUsername;
+    @Value("${rabbitMQ.password}")
+    private String rabbitPassword;*/
+
+}

ddp-runner-api/src/main/java/com/fibo/ddp/runner/api/config/ConfigurationContainor.java

@@ -0,0 +1,46 @@
+package com.fibo.ddp.runner.api.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisPoolConfig;
+
+import javax.annotation.Resource;
+import java.util.concurrent.ThreadPoolExecutor;
+
+@Configuration
+public class ConfigurationContainor {
+
+    @Resource
+    private ConfigHolder configHolder;
+
+    @Bean(name = "threadPoolTaskExecutor")
+    ThreadPoolTaskExecutor threadPoolTaskExecutor(){
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(2000);
+        executor.setMaxPoolSize(10000);
+        executor.setQueueCapacity(100000);
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
+        return executor;
+    }
+
+    @Bean(name = "jedisPool")
+    public JedisPool jedisPool(){
+        JedisPoolConfig config = new JedisPoolConfig();
+        config.setMaxTotal(configHolder.getRedisMaxTotal());
+        config.setMaxIdle(configHolder.getRedisMaxIdle());
+        config.setMaxWaitMillis(configHolder.getRedisMaxWait());
+        config.setTestOnBorrow(true);
+//        config.setTestOnReturn(true);
+
+        JedisPool pool = new JedisPool(config,
+                configHolder.getRedisHost(),
+                configHolder.getRedisPort(),
+                configHolder.getRedisTimeout(),
+                configHolder.getRedisPwd(),
+                configHolder.getRedisDb());
+        return pool;
+    }
+
+}

ddp-runner-api/src/main/java/com/fibo/ddp/runner/api/config/DataSourceConfig.java

@@ -0,0 +1,69 @@
+package com.fibo.ddp.runner.api.config;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
+import com.fibo.ddp.common.service.datax.runner.mysql.DynamicDataSource;
+import org.apache.ibatis.session.SqlSessionFactory;
+import org.mybatis.spring.SqlSessionTemplate;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.core.io.support.ResourcePatternResolver;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+
+import javax.sql.DataSource;
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class DataSourceConfig {
+
+    // 将所有前缀为spring.datasource.druid下的配置项都加载DataSource中
+    @ConfigurationProperties(prefix = "spring.datasource.druid")
+    @Bean
+    public DruidDataSource druidDataSource() {
+        return new DruidDataSource();
+    }
+
+    @Bean
+    public DynamicDataSource dynamicDataSource(@Qualifier("druidDataSource") DruidDataSource defaultDataSource) {
+        DynamicDataSource dynamicDataSource = DynamicDataSource.getInstance();
+        Map<Object,Object> map = new HashMap<>();
+        map.put("default", defaultDataSource);
+        dynamicDataSource.setTargetDataSources(map);
+        dynamicDataSource.setDefaultTargetDataSource(defaultDataSource);
+        return dynamicDataSource;
+    }
+
+    @Bean(name = "dataSourceTransactionManager")
+    public DataSourceTransactionManager dataSourceTransactionManager(@Qualifier("druidDataSource") DruidDataSource defaultDataSource){
+        DataSourceTransactionManager dm = new DataSourceTransactionManager();
+        dm.setDataSource(defaultDataSource);
+        return dm;
+    }
+
+    @Bean
+    public SqlSessionFactory sqlSessionFactory(
+            @Qualifier("dynamicDataSource") DataSource dynamicDataSource)
+            throws Exception {
+        MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();
+        ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
+        Resource[] mapperXmlResource = resolver.getResources("classpath*:**/*Mapper.xml");
+        bean.setDataSource(dynamicDataSource);
+        bean.setMapperLocations(mapperXmlResource);
+        bean.setTypeAliasesPackage("com.fibo.ddp.common.model");
+        bean.getObject().getConfiguration().setMapUnderscoreToCamelCase(true);
+        bean.getObject().getConfiguration().setCacheEnabled(false);
+        return bean.getObject();
+    }
+
+    @Bean(name = "sqlSessionTemplate")
+    public SqlSessionTemplate sqlSessionTemplate(
+            @Qualifier("sqlSessionFactory") SqlSessionFactory sqlSessionFactory)
+            throws Exception {
+        return new SqlSessionTemplate(sqlSessionFactory);
+    }
+}

ddp-runner-api/src/main/java/com/fibo/ddp/runner/api/config/RestTemplateConfig.java

@@ -0,0 +1,52 @@
+package com.fibo.ddp.runner.api.config;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.http.client.AsyncClientHttpRequestFactory;
+import org.springframework.http.client.ClientHttpRequestFactory;
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
+import org.springframework.web.client.AsyncRestTemplate;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * RestTemplate配置
+ */
+@Configuration
+public class RestTemplateConfig {
+
+    @Bean
+    public RestTemplate restTemplate(@Qualifier("clientHttpRequestFactory") ClientHttpRequestFactory factory){
+        return new RestTemplate(factory);
+    }
+
+    @Bean
+    public AsyncRestTemplate asyncRestTemplate(@Qualifier("asyncClientHttpRequestFactory") AsyncClientHttpRequestFactory factory){
+        return new AsyncRestTemplate(factory);
+    }
+
+    @Bean
+    public ClientHttpRequestFactory clientHttpRequestFactory(){
+        // 创建一个 httpCilent 简单工厂
+        SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
+        // 设置连接超时
+        factory.setConnectTimeout(15000);
+        // 设置读取超时
+        factory.setReadTimeout(5000);
+        return factory;
+    }
+
+    @Bean
+    public AsyncClientHttpRequestFactory asyncClientHttpRequestFactory(){
+        // 创建一个 httpCilent 简单工厂
+        SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
+        // 设置连接超时
+//        factory.setConnectTimeout(15000);
+        // 设置读取超时
+//        factory.setReadTimeout(5000);
+        //设置异步任务（线程不会重用，每次调用时都会重新启动一个新的线程）
+        factory.setTaskExecutor(new SimpleAsyncTaskExecutor());
+        return factory;
+    }
+}