分布式之分布式事务、分布式锁、分布式Session

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

<ValveclassName="com.orangefunction.tomcat.redissessions.RedisSessionHandlerValve" /><ManagerclassName="com.orangefunction.tomcat.redissessions.RedisSessionManager"sentinelMaster="mymaster"sentinels="<sentinel1-ip>:26379,<sentinel2-ip>:26379,<sentinel3-ip>:26379"maxInactiveInterval="60"/>

<dependency><groupId>org.springframework.session</groupId><artifactId>spring-session-data-redis</artifactId><version>1.2.1.RELEASE</version></dependency><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.8.1</version></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>springSessionRepositoryFilter</filter-name><filter-class>org.springframework.web.filter.DelegatingFilterProxy</filter-class></filter><filter-mapping><filter-name>springSessionRepositoryFilter</filter-name><url-pattern>/*</url-pattern></filter-mapping>

@RestController@RequestMapping("/test")publicclassTestController{
@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;    }}

if redis.call("get",KEYS[1]) == ARGV[1] thenreturn redis.call("del",KEYS[1])elsereturn0end

publicclassZooKeeperSession {
privatestatic CountDownLatch connectedSemaphore = new CountDownLatch(1);
private ZooKeeper zookeeper;private CountDownLatch latch;
publicZooKeeperSession() {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);returntrue;        } 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);returntrue;                } catch (Exception ee) {continue;                }            }
        }returntrue;    }
/**     * 释放掉一个分布式锁     *     * @param productId     */publicvoidreleaseDistributedLock(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     *     */privateclassZooKeeperWatcherimplementsWatcher {
publicvoidprocess(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     *     */privatestaticclassSingleton {
privatestatic ZooKeeperSession instance;
static {            instance = new ZooKeeperSession();        }
publicstatic ZooKeeperSession getInstance() {return instance;        }
    }
/**     * 获取单例     *     * @return     */publicstatic ZooKeeperSession getInstance() {return Singleton.getInstance();    }
/**     * 初始化单例的便捷方法     */publicstaticvoidinit() {        getInstance();    }
}

publicclassZooKeeperDistributedLockimplementsWatcher {
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);privateint sessionTimeout = 30000;
publicZooKeeperDistributedLock(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) {thrownew LockException(e);        } catch (KeeperException e) {thrownew LockException(e);        } catch (InterruptedException e) {thrownew LockException(e);        }    }
publicvoidprocess(WatchedEvent event) {if (event.getState() == KeeperState.SyncConnected) {            connectedLatch.countDown();return;        }
if (this.latch != null) {this.latch.countDown();        }    }
publicvoidacquireDistributedLock() {try {if (this.tryLock()) {return;            } else {                waitForLock(waitNode, sessionTimeout);            }        } catch (KeeperException e) {thrownew LockException(e);        } catch (InterruptedException e) {thrownew LockException(e);        }    }
public boolean tryLock() {try {// 传入进去的locksRoot + “/” + productId// 假设productId代表了一个商品id，比如说1// locksRoot = locks// /locks/10000000000，/locks/10000000001，/locks/10000000002            lockNode = zk.create(locksRoot + "/" + productId, newbyte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);// 看看刚创建的节点是不是最小的节点// locks：10000000000，10000000001，10000000002            List<String> locks = zk.getChildren(locksRoot, false);            Collections.sort(locks);if(lockNode.equals(locksRoot+"/"+ locks.get(0))){//如果是最小的节点,则表示取得锁returntrue;            }//如果不是最小的节点，找到比自己小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) {thrownew LockException(e);        } catch (InterruptedException e) {thrownew LockException(e);        }returnfalse;    }
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;        }returntrue;    }
publicvoidunlock() {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();        }    }
publicclassLockExceptionextendsRuntimeException {privatestatic final long serialVersionUID = 1L;
publicLockException(String e) {            super(e);        }
publicLockException(Exception e) {            super(e);        }    }}