分布式事务、分布式锁、分布式session

<Valve className="com.orangefunction.tomcat.redissessions.RedisSessionHandlerValve" />

<Manager className="com.orangefunction.tomcat.redissessions.RedisSessionManager"
         host="{redis.host}"
         port="{redis.port}"
         database="{redis.dbnum}"
         maxInactiveInterval="60"/>

<Valve className="com.orangefunction.tomcat.redissessions.RedisSessionHandlerValve" />

<Manager className="com.orangefunction.tomcat.redissessions.RedisSessionManager"
         sentinelMaster="mymaster"
         sentinels=":26379,:26379,:26379"
         maxInactiveInterval="60"/>

<dependency>
    <groupId>org.springframework.sessiongroupId>
    <artifactId>spring-session-data-redisartifactId>
    <version>1.2.1.RELEASEversion>
dependency>

<dependency>
      <groupId>redis.clientsgroupId>
      <artifactId>jedisartifactId>
      <version>2.8.1version>
dependency>

<bean id="redisHttpSessionConfiguration"
class="org.springframework.session.data.redis.config.annotation.web.http.RedisHttpSessionConfiguration">
    <property name="maxInactiveIntervalInSeconds" value="600"/>
bean>
<bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig">
    <property name="maxTotal" value="100" />
    <property name="maxIdle" value="10" />
bean>
<bean id="jedisConnectionFactory"
class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory" destroy-method="destroy">
    <property name="hostName" value="${redis_hostname}"/>
    <property name="port" value="${redis_port}"/>
    <property name="password" value="${redis_pwd}" />
    <property name="timeout" value="3000"/>
    <property name="usePool" value="true"/>
    <property name="poolConfig" ref="jedisPoolConfig"/>
bean>

<filter>
  <filter-name>springSessionRepositoryFilterfilter-name>
  <filter-class>org.springframework.web.filter.DelegatingFilterProxyfilter-class>
filter>
<filter-mapping>
  <filter-name>springSessionRepositoryFilterfilter-name>
  <url-pattern>/*url-pattern>
filter-mapping>

@RestController
@RequestMapping("/test")
public class TestController {

  @RequestMapping("/putIntoSession")
  public String putIntoSession(HttpServletRequest request, String username) {
    request.getSession().setAttribute("name",  "leo");
    return "ok";
  }
    
@RequestMapping("/getFromSession")
  public String getFromSession(HttpServletRequest request, Model model){
    String name = request.getSession().getAttribute("name");
    return name;
  }
  
}

public class ZooKeeperSession {

private static CountDownLatch connectedSemaphore = new CountDownLatch(1);

private ZooKeeper zookeeper;
private CountDownLatch latch;

public ZooKeeperSession() {
try {
this.zookeeper = new ZooKeeper("192.168.31.187:2181,192.168.31.19:2181,192.168.31.227:2181", 50000, new ZooKeeperWatcher());
try {
                connectedSemaphore.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("ZooKeeper session established......");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

/**
     * 获取分布式锁
     *
     * @param productId
     */
public Boolean acquireDistributedLock(Long productId) {
        String path = "/product-lock-" + productId;

try {
            zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
return true;
        } catch (Exception e) {
while (true) {
try {
// 相当于是给node注册一个监听器，去看看这个监听器是否存在
                    Stat stat = zk.exists(path, true);

if (stat != null) {
this.latch = new CountDownLatch(1);
this.latch.await(waitTime, TimeUnit.MILLISECONDS);
this.latch = null;
                    }
                    zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
return true;
                } catch (Exception ee) {
continue;
                }
            }

        }
return true;
    }

/**
     * 释放掉一个分布式锁
     *
     * @param productId
     */
public void releaseDistributedLock(Long productId) {
        String path = "/product-lock-" + productId;
try {
            zookeeper.delete(path, -1);
            System.out.println("release the lock for product[id=" + productId + "]......");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

/**
     * 建立zk session的watcher
     *
     * @author bingo
     * @since 2018/11/29
     *
     */
private class ZooKeeperWatcher implements Watcher {

public void process(WatchedEvent event) {
            System.out.println("Receive watched event: " + event.getState());

if (KeeperState.SyncConnected == event.getState()) {
                connectedSemaphore.countDown();
            }

if (this.latch != null) {
this.latch.countDown();
            }
        }

    }

/**
     * 封装单例的静态内部类
     *
     * @author bingo
     * @since 2018/11/29
     *
     */
private static class Singleton {

private static ZooKeeperSession instance;

static {
            instance = new ZooKeeperSession();
        }

public static ZooKeeperSession getInstance() {
return instance;
        }

    }

/**
     * 获取单例
     *
     * @return
     */
public static ZooKeeperSession getInstance() {
return Singleton.getInstance();
    }

/**
     * 初始化单例的便捷方法
     */
public static void init() {
        getInstance();
    }

}

public class ZooKeeperDistributedLock implements Watcher {

private ZooKeeper zk;
private String locksRoot = "/locks";
private String productId;
private String waitNode;
private String lockNode;
private CountDownLatch latch;
private CountDownLatch connectedLatch = new CountDownLatch(1);
private int sessionTimeout = 30000;

public ZooKeeperDistributedLock(String productId) {
this.productId = productId;
try {
            String address = "192.168.31.187:2181,192.168.31.19:2181,192.168.31.227:2181";
            zk = new ZooKeeper(address, sessionTimeout, this);
            connectedLatch.await();
        } catch (IOException e) {
throw new LockException(e);
        } catch (KeeperException e) {
throw new LockException(e);
        } catch (InterruptedException e) {
throw new LockException(e);
        }
    }

public void process(WatchedEvent event) {
if (event.getState() == KeeperState.SyncConnected) {
            connectedLatch.countDown();
return;
        }

if (this.latch != null) {
this.latch.countDown();
        }
    }

public void acquireDistributedLock() {
try {
if (this.tryLock()) {
return;
            } else {
                waitForLock(waitNode, sessionTimeout);
            }
        } catch (KeeperException e) {
throw new LockException(e);
        } catch (InterruptedException e) {
throw new LockException(e);
        }
    }

public boolean tryLock() {
try {
// 传入进去的locksRoot + “/” + productId
// 假设productId代表了一个商品id，比如说1
// locksRoot = locks
// /locks/10000000000，/locks/10000000001，/locks/10000000002
            lockNode = zk.create(locksRoot + "/" + productId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

// 看看刚创建的节点是不是最小的节点
// locks：10000000000，10000000001，10000000002
            List locks = zk.getChildren(locksRoot, false);
            Collections.sort(locks);

if(lockNode.equals(locksRoot+"/"+ locks.get(0))){
//如果是最小的节点,则表示取得锁
return true;
            }

//如果不是最小的节点，找到比自己小1的节点
int previousLockIndex = -1;
for(int i = 0; i < locks.size(); i++) {
if(lockNode.equals(locksRoot + “/” + locks.get(i))) {
                     previousLockIndex = i - 1;
break;
        }
       }

this.waitNode = locks.get(previousLockIndex);
        } catch (KeeperException e) {
throw new LockException(e);
        } catch (InterruptedException e) {
throw new LockException(e);
        }
return false;
    }

private boolean waitForLock(String waitNode, long waitTime) throws InterruptedException, KeeperException {
        Stat stat = zk.exists(locksRoot + "/" + waitNode, true);
if (stat != null) {
this.latch = new CountDownLatch(1);
this.latch.await(waitTime, TimeUnit.MILLISECONDS);
this.latch = null;
        }
return true;
    }

public void unlock() {
try {
// 删除/locks/10000000000节点
// 删除/locks/10000000001节点
            System.out.println("unlock " + lockNode);
            zk.delete(lockNode, -1);
            lockNode = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

public class LockException extends RuntimeException {
private static final long serialVersionUID = 1L;

public LockException(String e) {
            super(e);
        }

public LockException(Exception e) {
            super(e);
        }
    }
}