Merge #57 into master from pingkang

Merge branch 'master' of codeup.aliyun.com:60c06368dbbed34fe0f0c263/job/hr_job-...

* pingkang: (4 commits)
  xxl非指定前缀ip跳过初始化
  xxl非指定前缀ip跳过初始化
  adq creative list添加字段
  Merge branch 'master' of codeup.aliyun.com:60c06368dbbed34fe0f0c263/job/hr_job-...

Signed-off-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>
Merged-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>

CR-link: https://codeup.aliyun.com/60c06368dbbed34fe0f0c263/job/hr_job-bytedance/change/57

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/config/XxlJobConfig.java

@@ -21,6 +21,9 @@ public class XxlJobConfig {
   @Value("${xxl.job.executor.ip}")
   private String ip;
 
+  @Value("${xxl.job.executor.includeIpPrefix}")
+  private String includeIpPrefix;
+
   @Value("${xxl.job.executor.port}")
   private int port;
 
@@ -39,6 +42,7 @@ public class XxlJobConfig {
     xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
     xxlJobSpringExecutor.setAppname(appName);
     xxlJobSpringExecutor.setIp(ip);
+    xxlJobSpringExecutor.setIncludeIpPrefix(includeIpPrefix);
     xxlJobSpringExecutor.setPort(port);
     xxlJobSpringExecutor.setAccessToken(accessToken);
     xxlJobSpringExecutor.setLogPath(logPath);

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/controller/TestController.java

@@ -0,0 +1,33 @@
+package cn.com.ctop.job.bytedance.controller;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Objects;
+
+@RestController
+@RequestMapping("/test")
+@Slf4j
+public class TestController {
+    @Autowired
+    private RedisTemplate redisTemplateDefault;
+
+    @Autowired
+    private RedisTemplate redis2Template_db8;
+
+    @GetMapping("/testRedis")
+    public void testRedis(String key) {
+        Object o = redisTemplateDefault.opsForValue().get(key);
+        System.out.println(o);
+    }
+
+    @GetMapping("/testRedis2")
+    public void testRedis2(String key) {
+        System.out.println(Objects.requireNonNull(redis2Template_db8.opsForValue().get(key)));
+    }
+
+}

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/BytedanceCdpProjectServiceImpl.java

@@ -497,6 +497,7 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
         fields.add("creative_components");
         fields.add("created_time");
         fields.add("last_modified_time");
+        fields.add("creative_set_approval_status");
         param.put("fields", fields);
         String result = HttpUtils.httpGet("https://api.e.qq.com/v3.0/dynamic_creatives/get", param, new HashMap<>());
         JSONObject obj = JSONObject.parseObject(result);

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/RedisConfig.java

@@ -0,0 +1,55 @@
+package cn.com.ctop.job.bytedance.utils;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+@Configuration
+public class RedisConfig {
+    @Value("${spring.redis.host}")
+    private String host;
+
+    @Value("${spring.redis.port}")
+    private int port;
+
+    @Value("${spring.redis.database}")
+    private int database;
+
+    @Value("${spring.redis.password}")
+    private String password;
+
+    @Bean(name = "redisTemplateDefault")
+    public RedisTemplate<String, Object> redisTemplateDefault() {
+        return createTemplate(database);
+    }
+
+    @Bean(name = "redisTemplate_db10")
+    public RedisTemplate<String, Object> redisTemplate1() {
+        return createTemplate(10);
+    }
+
+    private RedisTemplate<String, Object> createTemplate(int dbIndex) {
+        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration();
+        config.setHostName(host);
+        config.setPort(port);
+        config.setDatabase(dbIndex);
+        config.setPassword(password);
+
+        LettuceConnectionFactory factory = new LettuceConnectionFactory(config);
+        factory.afterPropertiesSet();
+
+        RedisTemplate<String, Object> template = new RedisTemplate<>();
+        template.setKeySerializer(new StringRedisSerializer());
+        template.setValueSerializer(new StringRedisSerializer());
+        template.setHashKeySerializer(new StringRedisSerializer());
+        template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
+        template.setConnectionFactory(factory);
+        template.afterPropertiesSet();
+        return template;
+    }
+}

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/RedisUtil.java

@@ -1,243 +0,0 @@
-package cn.com.ctop.job.bytedance.utils;
-
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.ibatis.cursor.Cursor;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cache.annotation.CachingConfigurerSupport;
-import org.springframework.context.annotation.Bean;
-import org.springframework.data.redis.connection.RedisConnectionFactory;
-import org.springframework.data.redis.core.RedisCallback;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.data.redis.core.ScanOptions;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.data.redis.serializer.RedisSerializer;
-import org.springframework.data.redis.serializer.StringRedisSerializer;
-import org.springframework.stereotype.Component;
-import org.springframework.util.CollectionUtils;
-import org.springframework.util.StringUtils;
-
-import javax.annotation.Resource;
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-
-/**
- * redis 工具类
- * @Author Scott
- *
- */
-@Component
-@Slf4j
-public class RedisUtil {
-
-
-	@Autowired
-	private StringRedisTemplate redisTemplate;
-
-
-	private final String DEFAULT_KEY_PREFIX = "";
-	private final int EXPIRE_TIME = 1;
-	private final TimeUnit EXPIRE_TIME_TYPE = TimeUnit.DAYS;
-
-
-	/**
-	 * 数据缓存至redis
-	 *
-	 * @param key
-	 * @param value
-	 * @return
-	 */
-	public <K, V> void add(K key, V value) {
-		try {
-			if (value != null) {
-				redisTemplate
-						.opsForValue()
-						.set(DEFAULT_KEY_PREFIX + key, JSON.toJSONString(value));
-			}
-		} catch (Exception e) {
-			log.error(e.getMessage(), e);
-			throw new RuntimeException("数据缓存至redis失败");
-		}
-	}
-
-	/**
-	 * 数据缓存至redis并设置过期时间
-	 *
-	 * @param key
-	 * @param value
-	 * @return
-	 */
-	public <K, V> void add(K key, V value, long timeout, TimeUnit unit) {
-		try {
-			if (value != null) {
-				redisTemplate
-						.opsForValue()
-						.set(DEFAULT_KEY_PREFIX + key, JSON.toJSONString(value), timeout, unit);
-			}
-		} catch (Exception e) {
-			log.error(e.getMessage(), e);
-			throw new RuntimeException("数据缓存至redis失败");
-		}
-	}
-
-	/**
-	 * 写入 hash-set,已经是key-value的键值，不能再写入为hash-set
-	 *
-	 * @param key    must not be {@literal null}.
-	 * @param subKey must not be {@literal null}.
-	 * @param value  写入的值
-	 */
-	public <K, SK, V> void addHashCache(K key, SK subKey, V value) {
-		redisTemplate.opsForHash().put(DEFAULT_KEY_PREFIX + key, subKey, value);
-	}
-
-	/**
-	 * 写入 hash-set,并设置过期时间
-	 *
-	 * @param key    must not be {@literal null}.
-	 * @param subKey must not be {@literal null}.
-	 * @param value  写入的值
-	 */
-	public <K, SK, V> void addHashCache(K key, SK subKey, V value, long timeout, TimeUnit unit) {
-		redisTemplate.opsForHash().put(DEFAULT_KEY_PREFIX + key, subKey, value);
-		redisTemplate.expire(DEFAULT_KEY_PREFIX + key, timeout, unit);
-	}
-
-	/**
-	 * 获取 hash-setvalue
-	 *
-	 * @param key    must not be {@literal null}.
-	 * @param subKey must not be {@literal null}.
-	 */
-	public <K, SK> Object getHashCache(K key, SK subKey) {
-		return  redisTemplate.opsForHash().get(DEFAULT_KEY_PREFIX + key, subKey);
-	}
-
-
-	/**
-	 * 从redis中获取缓存数据，转成对象
-	 *
-	 * @param key   must not be {@literal null}.
-	 * @param clazz 对象类型
-	 * @return
-	 */
-	public <K, V> V getObject(K key, Class<V> clazz) {
-		String value = this.get(key);
-		V result = null;
-		if (!StringUtils.isEmpty(value)) {
-			result = JSONObject.parseObject(value, clazz);
-		}
-		return result;
-	}
-
-	/**
-	 * 从redis中获取缓存数据，转成list
-	 *
-	 * @param key   must not be {@literal null}.
-	 * @param clazz 对象类型
-	 * @return
-	 */
-	public <K, V> List<V> getList(K key, Class<V> clazz) {
-		String value = this.get(key);
-		List<V> result = Collections.emptyList();
-		if (!StringUtils.isEmpty(value)) {
-			result = JSONArray.parseArray(value, clazz);
-		}
-		return result;
-	}
-
-	/**
-	 * 功能描述：Get the value of {@code key}.
-	 *
-	 * @param key must not be {@literal null}.
-	 * @return java.lang.String
-	 * @date 2021/9/19
-	 **/
-	public <K> String get(K key) {
-		String value;
-		try {
-			value = redisTemplate.opsForValue().get(DEFAULT_KEY_PREFIX + key);
-		} catch (Exception e) {
-			log.error(e.getMessage(), e);
-			throw new RuntimeException("从redis缓存中获取缓存数据失败");
-		}
-		return value;
-	}
-
-	/**
-	 * 删除key
-	 */
-	public void delete(String key) {
-		redisTemplate.delete(key);
-	}
-
-	/**
-	 * 批量删除key
-	 */
-	public void delete(Collection<String> keys) {
-		redisTemplate.delete(keys);
-	}
-
-	/**
-	 * 序列化key
-	 */
-	public byte[] dump(String key) {
-		return redisTemplate.dump(key);
-	}
-
-	/**
-	 * 是否存在key
-	 */
-	public Boolean hasKey(String key) {
-		return redisTemplate.hasKey(key);
-	}
-
-	/**
-	 * 设置过期时间
-	 */
-	public Boolean expire(String key, long timeout, TimeUnit unit) {
-		return redisTemplate.expire(key, timeout, unit);
-	}
-
-	/**
-	 * 设置过期时间
-	 */
-	public Boolean expireAt(String key, Date date) {
-		return redisTemplate.expireAt(key, date);
-	}
-
-
-	/**
-	 * 移除 key 的过期时间，key 将持久保持
-	 */
-	public Boolean persist(String key) {
-		return redisTemplate.persist(key);
-	}
-
-	/**
-	 * 返回 key 的剩余的过期时间
-	 */
-	public Long getExpire(String key, TimeUnit unit) {
-		return redisTemplate.getExpire(key, unit);
-	}
-
-	/**
-	 * 返回 key 的剩余的过期时间
-	 */
-	public Long getExpire(String key) {
-		return redisTemplate.getExpire(key);
-	}
-
-
-	/**
-	 *  返回 满足前缀 的key
-	 * @param pattern
-	 * @return
-	 */
-	public Set<String> kesy(String pattern){
-		return redisTemplate.keys(pattern);
-	}
-
-}

job-bytedance/src/main/resources/application.yml

@@ -172,6 +172,7 @@ xxl:
       #      appname: module-job-test
       port: 9004
       ip:
+      includeIpPrefix: 172
       logpath: /data/webapp/hr_job-bytedance/log
       logretentiondays: 3
     admin:

xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java

@@ -25,7 +25,7 @@ import java.util.concurrent.ConcurrentMap;
 /**
  * Created by xuxueli on 2016/3/2 21:14.
  */
-public class XxlJobExecutor  {
+public class XxlJobExecutor {
     private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);
 
     // ---------------------- param ----------------------
@@ -34,6 +34,7 @@ public class XxlJobExecutor  {
     private String appname;
     private String address;
     private String ip;
+    private String includeIpPrefix;
     private int port;
     private String logPath;
     private int logRetentionDays;
@@ -41,24 +42,35 @@ public class XxlJobExecutor  {
     public void setAdminAddresses(String adminAddresses) {
         this.adminAddresses = adminAddresses;
     }
+
     public void setAccessToken(String accessToken) {
         this.accessToken = accessToken;
     }
+
     public void setAppname(String appname) {
         this.appname = appname;
     }
+
     public void setAddress(String address) {
         this.address = address;
     }
+
     public void setIp(String ip) {
         this.ip = ip;
     }
+
+    public void setIncludeIpPrefix(String includeIpPrefix) {
+        this.includeIpPrefix = includeIpPrefix;
+    }
+
     public void setPort(int port) {
         this.port = port;
     }
+
     public void setLogPath(String logPath) {
         this.logPath = logPath;
     }
+
     public void setLogRetentionDays(int logRetentionDays) {
         this.logRetentionDays = logRetentionDays;
     }
@@ -66,7 +78,6 @@ public class XxlJobExecutor  {
 
     // ---------------------- start + stop ----------------------
     public void start() throws Exception {
-
         // init logpath
         XxlJobFileAppender.initLogPath(logPath);
 
@@ -83,13 +94,19 @@ public class XxlJobExecutor  {
         // init executor-server
         initEmbedServer(address, ip, port, appname, accessToken);
     }
-    public void destroy(){
+
+    public boolean checkIncludeIpPrefix() {
+        String realIp = (ip != null && !ip.trim().isEmpty()) ? ip : IpUtil.getIp();
+        return includeIpPrefix == null || includeIpPrefix.trim().isEmpty() || includeIpPrefix.contains(realIp.split("\\.")[0]);
+    }
+
+    public void destroy() {
         // destroy executor-server
         stopEmbedServer();
 
         // destroy jobThreadRepository
         if (jobThreadRepository.size() > 0) {
-            for (Map.Entry<Integer, JobThread> item: jobThreadRepository.entrySet()) {
+            for (Map.Entry<Integer, JobThread> item : jobThreadRepository.entrySet()) {
                 JobThread oldJobThread = removeJobThread(item.getKey(), "web container destroy and kill the job.");
                 // wait for job thread push result to callback queue
                 if (oldJobThread != null) {
@@ -116,10 +133,11 @@ public class XxlJobExecutor  {
 
     // ---------------------- admin-client (rpc invoker) ----------------------
     private static List<AdminBiz> adminBizList;
+
     private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
-        if (adminAddresses!=null && adminAddresses.trim().length()>0) {
-            for (String address: adminAddresses.trim().split(",")) {
-                if (address!=null && address.trim().length()>0) {
+        if (adminAddresses != null && adminAddresses.trim().length() > 0) {
+            for (String address : adminAddresses.trim().split(",")) {
+                if (address != null && address.trim().length() > 0) {
 
                     AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);
 
@@ -131,7 +149,8 @@ public class XxlJobExecutor  {
             }
         }
     }
-    public static List<AdminBiz> getAdminBizList(){
+
+    public static List<AdminBiz> getAdminBizList() {
         return adminBizList;
     }
 
@@ -141,17 +160,17 @@ public class XxlJobExecutor  {
     private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {
 
         // fill ip port
-        port = port>0?port: NetUtil.findAvailablePort(9999);
-        ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
+        port = port > 0 ? port : NetUtil.findAvailablePort(9999);
+        ip = (ip != null && ip.trim().length() > 0) ? ip : IpUtil.getIp();
 
         // generate address
-        if (address==null || address.trim().length()==0) {
+        if (address == null || address.trim().length() == 0) {
             String ip_port_address = IpUtil.getIpPort(ip, port);   // registry-address：default use address to registry , otherwise use ip:port if address is null
             address = "http://{ip_port}/".replace("{ip_port}", ip_port_address);
         }
 
         // accessToken
-        if (accessToken==null || accessToken.trim().length()==0) {
+        if (accessToken == null || accessToken.trim().length() == 0) {
             logger.warn(">>>>>>>>>>> xxl-job accessToken is empty. To ensure system security, please set the accessToken.");
         }
 
@@ -174,14 +193,17 @@ public class XxlJobExecutor  {
 
     // ---------------------- job handler repository ----------------------
     private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
-    public static IJobHandler loadJobHandler(String name){
+
+    public static IJobHandler loadJobHandler(String name) {
         return jobHandlerRepository.get(name);
     }
-    public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
+
+    public static IJobHandler registJobHandler(String name, IJobHandler jobHandler) {
         logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
         return jobHandlerRepository.put(name, jobHandler);
     }
-    protected void registJobHandler(XxlJob xxlJob, Object bean, Method executeMethod){
+
+    protected void registJobHandler(XxlJob xxlJob, Object bean, Method executeMethod) {
         if (xxlJob == null) {
             return;
         }
@@ -238,12 +260,13 @@ public class XxlJobExecutor  {
 
     // ---------------------- job thread repository ----------------------
     private static ConcurrentMap<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
-    public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){
+
+    public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason) {
         JobThread newJobThread = new JobThread(jobId, handler);
         newJobThread.start();
         logger.info(">>>>>>>>>>> xxl-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler});
 
-        JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread);	// putIfAbsent | oh my god, map's put method return the old value!!!
+        JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread);    // putIfAbsent | oh my god, map's put method return the old value!!!
         if (oldJobThread != null) {
             oldJobThread.toStop(removeOldReason);
             oldJobThread.interrupt();
@@ -251,7 +274,8 @@ public class XxlJobExecutor  {
 
         return newJobThread;
     }
-    public static JobThread removeJobThread(int jobId, String removeOldReason){
+
+    public static JobThread removeJobThread(int jobId, String removeOldReason) {
         JobThread oldJobThread = jobThreadRepository.remove(jobId);
         if (oldJobThread != null) {
             oldJobThread.toStop(removeOldReason);
@@ -261,7 +285,8 @@ public class XxlJobExecutor  {
         }
         return null;
     }
-    public static JobThread loadJobThread(int jobId){
+
+    public static JobThread loadJobThread(int jobId) {
         JobThread jobThread = jobThreadRepository.get(jobId);
         return jobThread;
     }

xxl-job-core/src/main/java/com/xxl/job/core/executor/impl/XxlJobSpringExecutor.java

@@ -30,6 +30,10 @@ public class XxlJobSpringExecutor extends XxlJobExecutor implements ApplicationC
     // start
     @Override
     public void afterSingletonsInstantiated() {
+        if (!super.checkIncludeIpPrefix()){
+            logger.info("本机ip不在指定ip范围内, 跳过初始化xxl！！");
+            return;
+        }
 
         // init JobHandler Repository
         /*initJobHandlerRepository(applicationContext);*/