新增redis 分布式锁实现

hsweb-concurrent/hsweb-concurrent-lock/hsweb-concurrent-lock-redis/pom.xml

@@ -0,0 +1,33 @@
+<?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>hsweb-concurrent-lock</artifactId>
+        <groupId>org.hswebframework.web</groupId>
+        <version>3.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>hsweb-concurrent-lock-redis</artifactId>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>org.redisson</groupId>
+            <artifactId>redisson</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.hswebframework.web</groupId>
+            <artifactId>hsweb-concurrent-lock-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport-native-epoll</artifactId>
+            <version>4.1.8.Final</version>
+            <scope>provided</scope>
+            <classifier>linux-x86_64</classifier>
+        </dependency>
+    </dependencies>
+</project>

hsweb-concurrent/hsweb-concurrent-lock/hsweb-concurrent-lock-redis/src/main/java/org/hswebframework/web/concurrent/lock/redis/RedissonLockManager.java

@@ -0,0 +1,30 @@
+package org.hswebframework.web.concurrent.lock.redis;
+
+import org.hswebframework.web.concurrent.lock.AbstactLocakManager;
+import org.redisson.Redisson;
+
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+
+/**
+ *
+ * @author zhouhao
+ */
+public class RedissonLockManager extends AbstactLocakManager {
+    private Redisson redisson;
+
+    public RedissonLockManager(Redisson redisson) {
+        if (null == redisson) throw new NullPointerException();
+        this.redisson = redisson;
+    }
+
+    @Override
+    protected Lock createLock(String lockKey) {
+        return redisson.getLock(lockKey);
+    }
+
+    @Override
+    protected ReadWriteLock createReadWriteLock(String lockKey) {
+        return redisson.getReadWriteLock(lockKey);
+    }
+}

hsweb-concurrent/hsweb-concurrent-lock/hsweb-concurrent-lock-redis/src/test/java/org/hswebframework/web/concurrent/lock/redis/RedissonLockTest.java

@@ -0,0 +1,104 @@
+package org.hswebframework.web.concurrent.lock.redis;
+
+
+import org.hswebframework.web.concurrent.lock.LockManager;
+import org.junit.Assert;
+import org.redisson.Redisson;
+import org.redisson.config.Config;
+
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.function.Consumer;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public class RedissonLockTest {
+    static long counter = 0;
+
+    static LockManager lockManager = null;
+
+    static Redisson redisson;
+
+    public static LockManager createLockFactory() {
+        if (lockManager != null) {
+            return lockManager;
+        }
+        Config config = new Config();
+//        config.setUseLinuxNativeEpoll(true);
+        config.useSingleServer().setAddress("127.0.0.1:6379");
+        redisson = (Redisson) Redisson.create(config);
+        return lockManager = new RedissonLockManager(redisson);
+    }
+
+    public static void main(String[] args) throws InterruptedException {
+        testLock();
+
+        testReadWriteLock();
+        redisson.shutdown();
+    }
+
+    public static void testReadWriteLock() throws InterruptedException {
+        counter = 0;
+        LockManager lockManager = createLockFactory();
+
+        ReadWriteLock readWriteLock = lockManager.getReadWriteLock("foo");
+
+        Lock readLock = readWriteLock.readLock();
+
+        Lock writeLock = readWriteLock.writeLock();
+        Consumer<Long>[] consumer = new Consumer[1];
+        consumer[0] = System.out::println;
+        for (int i = 0; i < 10; i++) {
+            new Thread(() -> {
+                for (int i1 = 0; i1 < 10; i1++) {
+                    try {
+                        Thread.sleep(100);
+                    } catch (InterruptedException ignored) {
+                    }
+                    writeLock.lock();
+                    long tmp = ++counter;
+
+                    //判断增加的值与 读取的值一致
+                    consumer[0] = l -> Assert.assertEquals(Long.valueOf(tmp), l);
+                    System.out.println("write:" + counter);
+                    writeLock.unlock();
+                }
+            }).start();
+            new Thread(() -> {
+                for (int i1 = 0; i1 < 10; i1++) {
+                    try {
+                        Thread.sleep(100);
+                    } catch (InterruptedException ignored) {
+                    }
+                    readLock.lock();
+                    consumer[0].accept(counter);
+                    System.out.println("read:" + counter);
+                    readLock.unlock();
+                }
+            }).start();
+        }
+        Thread.sleep(5000);
+        System.out.println("wait 5s");
+    }
+
+    public static void testLock() throws InterruptedException {
+        counter = 0;
+        LockManager lockManager = createLockFactory();
+        Lock lock = lockManager.getLock("foo");
+        for (int i = 0; i < 100; i++) {
+            new Thread(() -> {
+                lock.lock();
+                for (int i1 = 0; i1 < 100; i1++) {
+                    counter++;
+                }
+                lock.unlock();
+            }).start();
+        }
+        Thread.sleep(1000);
+        System.out.println(counter);
+        Assert.assertEquals(counter, 100 * 100);
+    }
+}