2.0.0项目初始化

datax-common/datax-common-redis/pom.xml

@@ -24,5 +24,10 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-pool2</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.redisson</groupId>
+            <artifactId>redisson</artifactId>
+            <version>${redisson.version}</version>
+        </dependency>
     </dependencies>
 </project>
\ No newline at end of file

datax-common/datax-common-redis/src/main/java/cn/datax/common/redis/config/RedissonConfig.java

+package cn.datax.common.redis.config;
+
+import cn.datax.common.redis.service.impl.RedissonLocker;
+import cn.datax.common.redis.utils.LockUtil;
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
+import org.redisson.config.SingleServerConfig;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+
+public class RedissonConfig {
+
+    @Value("${spring.redis.database}")
+    private int database;
+    @Value("${spring.redis.host}")
+    private String host;
+    @Value("${spring.redis.port}")
+    private String port;
+    @Value("${spring.redis.password}")
+    private String password;
+    @Value("${spring.redis.timeout}")
+    private int timeout;
+
+    /**
+     * RedissonClient,单机模式
+     *
+     * @return
+     */
+    @Bean(destroyMethod = "shutdown")
+    public RedissonClient redissonClient() {
+        Config config = new Config();
+        SingleServerConfig singleServerConfig = config.useSingleServer();
+        singleServerConfig.setAddress("redis://" + host + ":" + port);
+        singleServerConfig.setTimeout(timeout);
+        singleServerConfig.setDatabase(database);
+        if (password != null && !"".equals(password)) {
+            singleServerConfig.setPassword(password);
+        }
+        return Redisson.create(config);
+    }
+
+    @Bean
+    public RedissonLocker redissonLocker(RedissonClient redissonClient) {
+        RedissonLocker locker = new RedissonLocker(redissonClient);
+        // 设置LockUtil的锁处理对象
+        LockUtil.setLocker(locker);
+        return locker;
+    }
+}

datax-common/datax-common-redis/src/main/java/cn/datax/common/redis/service/Locker.java

+package cn.datax.common.redis.service;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 锁接口
+ */
+public interface Locker {
+
+    /**
+     * 获取锁，如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。
+     *
+     * @param lockKey
+     */
+    void lock(String lockKey);
+
+    /**
+     * 释放锁
+     *
+     * @param lockKey
+     */
+    void unlock(String lockKey);
+
+    /**
+     * 获取锁,如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。如果获取到锁后，执行结束后解锁或达到超时时间后会自动释放锁
+     *
+     * @param lockKey
+     * @param timeout
+     */
+    void lock(String lockKey, int timeout);
+
+    /**
+     * 获取锁,如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。如果获取到锁后，执行结束后解锁或达到超时时间后会自动释放锁
+     *
+     * @param lockKey
+     * @param unit
+     * @param timeout
+     */
+    void lock(String lockKey, TimeUnit unit, int timeout);
+
+    /**
+     * 尝试获取锁，获取到立即返回true,未获取到立即返回false
+     *
+     * @param lockKey
+     * @return
+     */
+    boolean tryLock(String lockKey);
+
+    /**
+     * 尝试获取锁，在等待时间内获取到锁则返回true,否则返回false,如果获取到锁，则要么执行完后程序释放锁，
+     * 要么在给定的超时时间leaseTime后释放锁
+     *
+     * @param lockKey
+     * @param waitTime
+     * @param leaseTime
+     * @param unit
+     * @return
+     */
+    boolean tryLock(String lockKey, long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
+
+    /**
+     * 锁是否被任意一个线程锁持有
+     *
+     * @param lockKey
+     * @return
+     */
+    boolean isLocked(String lockKey);
+}

datax-common/datax-common-redis/src/main/java/cn/datax/common/redis/service/impl/RedissonLocker.java

+package cn.datax.common.redis.service.impl;
+
+import cn.datax.common.redis.service.Locker;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 基于Redisson的分布式锁
+ */
+public class RedissonLocker implements Locker {
+
+    private RedissonClient redissonClient;
+
+    public RedissonLocker(RedissonClient redissonClient) {
+        super();
+        this.redissonClient = redissonClient;
+    }
+
+    @Override
+    public void lock(String lockKey) {
+        RLock lock = redissonClient.getLock(lockKey);
+        lock.lock();
+    }
+
+    @Override
+    public void unlock(String lockKey) {
+        RLock lock = redissonClient.getLock(lockKey);
+        lock.unlock();
+    }
+
+    @Override
+    public void lock(String lockKey, int leaseTime) {
+        RLock lock = redissonClient.getLock(lockKey);
+        lock.lock(leaseTime, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void lock(String lockKey, TimeUnit unit, int timeout) {
+        RLock lock = redissonClient.getLock(lockKey);
+        lock.lock(timeout, unit);
+    }
+
+    public void setRedissonClient(RedissonClient redissonClient) {
+        this.redissonClient = redissonClient;
+    }
+
+    @Override
+    public boolean tryLock(String lockKey) {
+        RLock lock = redissonClient.getLock(lockKey);
+        return lock.tryLock();
+    }
+
+    @Override
+    public boolean tryLock(String lockKey, long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
+        RLock lock = redissonClient.getLock(lockKey);
+        return lock.tryLock(waitTime, leaseTime, unit);
+    }
+
+    @Override
+    public boolean isLocked(String lockKey) {
+        RLock lock = redissonClient.getLock(lockKey);
+        return lock.isLocked();
+    }
+}

datax-common/datax-common-redis/src/main/java/cn/datax/common/redis/utils/LockUtil.java

+package cn.datax.common.redis.utils;
+
+import cn.datax.common.redis.service.Locker;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * redis分布式锁工具类
+ */
+public final class LockUtil {
+
+    private static Locker locker;
+
+    /**
+     * 设置工具类使用的locker
+     *
+     * @param locker
+     */
+    public static void setLocker(Locker locker) {
+        LockUtil.locker = locker;
+    }
+
+    /**
+     * 获取锁
+     *
+     * @param lockKey
+     */
+    public static void lock(String lockKey) {
+        locker.lock(lockKey);
+    }
+
+    /**
+     * 释放锁
+     *
+     * @param lockKey
+     */
+    public static void unlock(String lockKey) {
+        locker.unlock(lockKey);
+    }
+
+    /**
+     * 获取锁，超时释放
+     *
+     * @param lockKey
+     * @param timeout
+     */
+    public static void lock(String lockKey, int timeout) {
+        locker.lock(lockKey, timeout);
+    }
+
+    /**
+     * 获取锁，超时释放，指定时间单位
+     *
+     * @param lockKey
+     * @param unit
+     * @param timeout
+     */
+    public static void lock(String lockKey, TimeUnit unit, int timeout) {
+        locker.lock(lockKey, unit, timeout);
+    }
+
+    /**
+     * 尝试获取锁，获取到立即返回true,获取失败立即返回false
+     *
+     * @param lockKey
+     * @return
+     */
+    public static boolean tryLock(String lockKey) {
+        return locker.tryLock(lockKey);
+    }
+
+    /**
+     * 尝试获取锁，在给定的waitTime时间内尝试，获取到返回true,获取失败返回false,获取到后再给定的leaseTime时间超时释放
+     *
+     * @param lockKey
+     * @param waitTime
+     * @param leaseTime
+     * @param unit
+     * @return
+     * @throws InterruptedException
+     */
+    public static boolean tryLock(String lockKey, long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
+        return locker.tryLock(lockKey, waitTime, leaseTime, unit);
+    }
+
+    /**
+     * 锁释放被任意一个线程持有
+     *
+     * @param lockKey
+     * @return
+     */
+    public static boolean isLocked(String lockKey) {
+        return locker.isLocked(lockKey);
+    }
+}

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

 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-  cn.datax.common.redis.config.RedisConfig
+  cn.datax.common.redis.config.RedisConfig,\
+  cn.datax.common.redis.config.RedissonConfig

pom.xml

@@ -48,6 +48,7 @@
     <sqlserver.version>8.2.1.jre8</sqlserver.version>
     <zxing.version>3.4.0</zxing.version>
     <aspose.version>20.3</aspose.version>
+    <redisson.version>3.12.5</redisson.version>
   </properties>
 
   <modules>