zookeeper实现分布式锁和配置中心
一、Zookeeper实现分布式锁
分布式锁主要用于在分布式环境中保证数据的一致性。
包括跨进程、跨机器、跨网络导致共享资源不一致的问题。
1. 分布式锁的实现思路
说明:
这种实现会有一个缺点,即当有很多进程在等待锁的时候,在释放锁的时候会有很多进程就过来争夺锁,这种现象称为 “惊群效应”
2. 分布式锁优化后的实现思路
3. Zookeeper分布式锁的代码实现
准备工作:
1)安装Zookeeper,具体参考我前面的我文章Zookeeper系列一:Zookeeper介绍、Zookeeper安装配置、ZK Shell的使用
2)新建一个maven项目ZK-Demo,然后在pom.xml里面引入相关的依赖
<dependency> <groupId>com.101tec</groupId> <artifactId>zkclient</artifactId> <version>0.10</version> </dependency>
3.1 Zookeeper分布式锁的核心代码实现
实现逻辑参考“2. 分布式锁优化后的实现思路”中的流程图
package com.study.demo.lock; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import org.I0Itec.zkclient.IZkDataListener; import org.I0Itec.zkclient.ZkClient; import org.I0Itec.zkclient.serialize.SerializableSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * * @Description: Zookeeper分布式锁的核心代码实现 * @author leeSmall * @date 2018年9月4日 * */ public class DistributedLock implements Lock { private static Logger logger = LoggerFactory.getLogger(DistributedLock.class); private static final String ZOOKEEPER_IP_PORT = "192.168.152.130:2181"; private static final String LOCK_PATH = "/LOCK"; private ZkClient client = new ZkClient(ZOOKEEPER_IP_PORT, 4000, 4000, new SerializableSerializer()); private CountDownLatch cdl; private String beforePath;// 当前请求的节点前一个节点 private String currentPath;// 当前请求的节点 // 判断有没有LOCK目录,没有则创建 public DistributedLock() { if (!this.client.exists(LOCK_PATH)) { this.client.createPersistent(LOCK_PATH); } } public void lock() { //尝试去获取分布式锁失败 if (!tryLock()) { //对次小节点进行监听 waitForLock(); lock(); } else { logger.info(Thread.currentThread().getName() + " 获得分布式锁!"); } } public boolean tryLock() { // 如果currentPath为空则为第一次尝试加锁,第一次加锁赋值currentPath if (currentPath == null || currentPath.length() <= 0) { // 创建一个临时顺序节点 currentPath = this.client.createEphemeralSequential(LOCK_PATH + '/', "lock"); System.out.println("---------------------------->" + currentPath); } // 获取所有临时节点并排序,临时节点名称为自增长的字符串如:0000000400 List<String> childrens = this.client.getChildren(LOCK_PATH); //由小到大排序所有子节点 Collections.sort(childrens); //判断创建的子节点/LOCK/Node-n是否最小,即currentPath,如果当前节点等于childrens中的最小的一个就占用锁 if (currentPath.equals(LOCK_PATH + '/' + childrens.get(0))) { return true; } //找出比创建的临时顺序节子节点/LOCK/Node-n次小的节点,并赋值给beforePath else { int wz = Collections.binarySearch(childrens, currentPath.substring(6)); beforePath = LOCK_PATH + '/' + childrens.get(wz - 1); } return false; } //等待锁,对次小节点进行监听 private void waitForLock() { IZkDataListener listener = new IZkDataListener() { public void handleDataDeleted(String dataPath) throws Exception { logger.info(Thread.currentThread().getName() + ":捕获到DataDelete事件!---------------------------"); if (cdl != null) { cdl.countDown(); } } public void handleDataChange(String dataPath, Object data) throws Exception { } }; // 对次小节点进行监听,即beforePath-给排在前面的的节点增加数据删除的watcher this.client.subscribeDataChanges(beforePath, listener); if (this.client.exists(beforePath)) { cdl = new CountDownLatch(1); try { cdl.await(); } catch (InterruptedException e) { e.printStackTrace(); } } this.client.unsubscribeDataChanges(beforePath, listener); } //完成业务逻辑以后释放锁 public void unlock() { // 删除当前临时节点 client.delete(currentPath); } // ========================================== public void lockInterruptibly() throws InterruptedException { } public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { return false; } public Condition newCondition() { return null; } }
3.2 在业务里面使用分布式锁
package com.study.demo.lock; import java.util.concurrent.CountDownLatch; import java.util.concurrent.locks.Lock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * * @Description: 在业务里面使用分布式锁 * @author leeSmall * @date 2018年9月4日 * */ public class OrderServiceImpl implements Runnable { private static OrderCodeGenerator ong = new OrderCodeGenerator(); private Logger logger = LoggerFactory.getLogger(OrderServiceImpl.class); // 同时并发的线程数 private static final int NUM = 10; // 按照线程数初始化倒计数器,倒计数器 private static CountDownLatch cdl = new CountDownLatch(NUM); private Lock lock = new DistributedLock(); // 创建订单接口 public void createOrder() { String orderCode = null; //准备获取锁 lock.lock(); try { // 获取订单编号 orderCode = ong.getOrderCode(); } catch (Exception e) { // TODO: handle exception } finally { //完成业务逻辑以后释放锁 lock.unlock(); } // ……业务代码 logger.info("insert into DB使用id:=======================>" + orderCode); } public void run() { try { // 等待其他线程初始化 cdl.await(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } // 创建订单 createOrder(); } public static void main(String[] args) { for (int i = 1; i <= NUM; i++) { // 按照线程数迭代实例化线程 new Thread(new OrderServiceImpl()).start(); // 创建一个线程,倒计数器减1 cdl.countDown(); } } }
工具类:
package com.study.demo.lock; import java.text.SimpleDateFormat; import java.util.Date; public class OrderCodeGenerator { // 自增长序列 private static int i = 0; // 按照“年-月-日-小时-分钟-秒-自增长序列”的规则生成订单编号 public String getOrderCode() { Date now = new Date(); SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss"); return sdf.format(now) + ++i; } }
二、Zookeeper实现配置中心
1. 首先在zookeeper里面创建一个Jdbc的节点,在下面分别创建4个子节点/Jdbc/url、/Jdbc/uname、/Jdbc/password、/Jdbc/driver
create /Jdbc '' create /Jdbc/url jdbc.mysql://192.168.152.1/dbspread create /Jdbc/uname root create /Jdbc/password 123456 create /Jdbc/driver com.mysql.jdbc.Driver
注意:/Jdbc/url这个节点的值是错的
2. 新建一个zkdemo的maven的web项目
项目结构如下:
2.1 在pom.xml文件里面引入下面依赖:
<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/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.study.demo</groupId> <artifactId>zkdemo</artifactId> <packaging>war</packaging> <version>0.0.1-SNAPSHOT</version> <name>zkdemo Maven Webapp</name> <url>http://maven.apache.org</url> <properties> <spring.version>4.3.8.RELEASE</spring.version> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>3.8.1</version> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.10</version> </dependency> <dependency> <groupId>com.101tec</groupId> <artifactId>zkclient</artifactId> <version>0.10</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>4.0.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>4.0.0</version> </dependency> <dependency> <groupId>javax.servlet</groupId> <artifactId>javax.servlet-api</artifactId> <version>3.1.0</version> </dependency> <dependency> <groupId>org.apache.tomcat</groupId> <artifactId>tomcat-catalina</artifactId> <version>7.0.39</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-core</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-beans</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jdbc</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-web</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-webmvc</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>com.zaxxer</groupId> <artifactId>HikariCP</artifactId> <version>2.7.1</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.41</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.25</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>jcl-over-slf4j</artifactId> <version>1.7.25</version> </dependency> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-core</artifactId> <version>2.9.1</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.9.1</version> </dependency> <dependency> <groupId>javax.servlet</groupId> <artifactId>jstl</artifactId> <version>1.2</version> </dependency> </dependencies> <build> <finalName>zkdemo</finalName> </build> </project>
2.2 新建一个zookeeper配置中心类,从zookeeper动态获取数据库配置
package com.study.demo.config; import java.util.List; import java.util.Properties; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.framework.recipes.cache.TreeCache; import org.apache.curator.framework.recipes.cache.TreeCacheEvent; import org.apache.curator.framework.recipes.cache.TreeCacheListener; import org.apache.curator.retry.ExponentialBackoffRetry; import org.springframework.web.context.ContextLoader; import org.springframework.web.context.WebApplicationContext; import com.zaxxer.hikari.HikariDataSource; /** * * @Description: zookeeper配置中心类,从zookeeper动态获取数据库配置 * @author leeSmall * @date 2018年9月10日 * */ public class ZookeeperConfigurerCentral { //curator客户端 private CuratorFramework zkClient; //curator事件监听 private TreeCache treeCache; //zookeeper的ip和端口 private String zkServers; //zookeeper上的/Jdbc路径 private String zkPath; //超时设置 private int sessionTimeout; //读取zookeeper上的数据库配置文件放到这里 private Properties props; public ZookeeperConfigurerCentral(String zkServers, String zkPath, int sessionTimeout) { this.zkServers = zkServers; this.zkPath = zkPath; this.sessionTimeout = sessionTimeout; this.props = new Properties(); //初始化curator客户端 initZkClient(); //从zookeeper的Jdbc节点下获取数据库配置存入props getConfigData(); //对zookeeper上的数据库配置文件所在节点进行监听,如果有改变就动态刷新props addZkListener(); } //初始化curator客户端 private void initZkClient() { zkClient = CuratorFrameworkFactory.builder().connectString(zkServers).sessionTimeoutMs(sessionTimeout) .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build(); zkClient.start(); } //从zookeeper的Jdbc节点下获取数据库配置存入props private void getConfigData() { try { List<String> list = zkClient.getChildren().forPath(zkPath); for (String key : list) { String value = new String(zkClient.getData().forPath(zkPath + "/" + key)); if (value != null && value.length() > 0) { props.put(key, value); } } } catch (Exception e) { e.printStackTrace(); } } //对zookeeper上的数据库配置文件所在节点进行监听,如果有改变就动态刷新props private void addZkListener() { TreeCacheListener listener = new TreeCacheListener() { public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception { if (event.getType() == TreeCacheEvent.Type.NODE_UPDATED) { getConfigData(); WebApplicationContext ctx = ContextLoader.getCurrentWebApplicationContext(); HikariDataSource dataSource = (HikariDataSource) ctx.getBean("dataSource"); System.out.println("================"+props.getProperty("url")); dataSource.setJdbcUrl(props.getProperty("url")); dataSource.setUsername(props.getProperty("uname")); dataSource.setPassword(props.getProperty("password ")); dataSource.setDriverClassName(props.getProperty("driver ")); } } }; treeCache = new TreeCache(zkClient, zkPath); try { treeCache.start(); treeCache.getListenable().addListener(listener); } catch (Exception e) { e.printStackTrace(); } } public Properties getProps() { return props; } public void setZkServers(String zkServers) { this.zkServers = zkServers; } public void setZkPath(String zkPath) { this.zkPath = zkPath; } public void setSessionTimeout(int sessionTimeout) { this.sessionTimeout = sessionTimeout; } }
2.3 新建一个加载props里面的数据库配置的类
package com.study.demo.config; import java.util.Properties; import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.config.PropertyPlaceholderConfigurer; /** * * @Description: 加载props里面的数据库配置,这个类等价于以前在xml文件里面的配置: * <context:property-placeholder location="classpath:config/jdbc_conf.properties"/> * @author leeSmall * @date 2018年9月10日 * */ public class ZookeeperPlaceholderConfigurer extends PropertyPlaceholderConfigurer { private ZookeeperConfigurerCentral zkConfigurerCentral; @Override protected void processProperties(ConfigurableListableBeanFactory beanFactoryToProcess, Properties props) throws BeansException { System.out.println(zkConfigurerCentral.getProps()); super.processProperties(beanFactoryToProcess, zkConfigurerCentral.getProps()); } public void setzkConfigurerCentral(ZookeeperConfigurerCentral zkConfigurerCentral) { this.zkConfigurerCentral = zkConfigurerCentral; } }
2.4 在/zkdemo/src/main/webapp/WEB-INF/config/applicationContext.xml配置2.2和2.3新建的两个主类
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:annotation-config /> <context:component-scan base-package="com.study.demo" /> <!--通过构造函数注入zkServers、sessionTimeout、zkPath从zookeeper动态获取数据库配置 --> <bean id="zkConfigurerCentral" class="com.study.demo.config.ZookeeperConfigurerCentral"> <constructor-arg name="zkServers" value="192.168.152.130:2181" /> <constructor-arg name="sessionTimeout" value="1000" /> <constructor-arg name="zkPath" value="/Jdbc" /> </bean> <!--这个类等价于以前在xml文件里面的配置: <context:property-placeholder location="classpath:config/jdbc_conf.properties"/> 加载 props里面的数据库配置 --> <bean id="zkPlaceholderConfigurer" class="com.study.demo.config.ZookeeperPlaceholderConfigurer"> <property name="zkConfigurerCentral" ref="zkConfigurerCentral" /> <property name="ignoreUnresolvablePlaceholders" value="true" /> <property name="order" value="1" /> </bean> <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate"> <property name="dataSource"> <ref bean="dataSource" /> </property> </bean> <bean id="dataSource" class="com.zaxxer.hikari.HikariDataSource" destroy-method="shutdown"> <property name="driverClassName" value="${driver}" /> <property name="jdbcUrl" value="${url}" /> <property name="username" value="${uname}" /> <property name="password" value="${password}" /> <!-- 连接只读数据库时配置为true, 保证安全 --> <property name="readOnly" value="false" /> <!-- 等待连接池分配连接的最大时长(毫秒),超过这个时长还没可用的连接则发生SQLException, 缺省:30秒 --> <property name="connectionTimeout" value="30000" /> <!-- 一个连接idle状态的最大时长(毫秒),超时则被释放(retired),缺省:10分钟 --> <property name="idleTimeout" value="600000" /> <!-- 一个连接的生命时长(毫秒),超时而且没被使用则被释放(retired),缺省:30分钟,建议设置比数据库超时时长少30秒,参考MySQL wait_timeout参数(show variables like '%timeout%';) --> <property name="maxLifetime" value="1800000" /> <!-- 连接池中允许的最大连接数。缺省值:10;推荐的公式:((core_count * 2) + effective_spindle_count) --> <property name="maximumPoolSize" value="15" /> </bean> </beans>
2.5 在com.study.demo.controller新建测试类
测试类1:
View Code
测试类2:
View Code
测试类3:
View Code
测试类4:
View Code
2.6 其他附加配置和数据库脚本
/zkdemo/src/main/webapp/WEB-INF/config/log4j.properties
View Code
/zkdemo/src/main/webapp/WEB-INF/config/spring-mvc.xml
View Code
/zkdemo/src/main/webapp/WEB-INF/web.xml
View Code
数据库脚本:
CREATE TABLE `tbl_order` ( `order_id` int(11) NOT NULL AUTO_INCREMENT COMMENT '订单id', `brand_id` int(11) DEFAULT NULL COMMENT '品牌id', PRIMARY KEY (`order_id`) ) ENGINE=MyISAM AUTO_INCREMENT=2 DEFAULT CHARSET=utf8 COMMENT='订单表';INSERT INTO tbl_order VALUES('1','1')
2.7 启动项目在浏览器输入地址http://localhost:8080/zkdemo/test查看效果
可以看到报错了,这是因为我们之前设置了错误的url
create /Jdbc/url jdbc.mysql://192.168.152.1/dbspread
修改url为正确的
set /Jdbc/url jdbc:mysql://192.168.152.1:3306/dbspread
再次输入地址访问查看效果:
http://localhost:8080/zkdemo/test
可以看到在没有重启服务的情况下,可以正常访问获取到值了,这是因为zookeeper的数据库的配置动态刷新到服务了!
相关推荐
周公周金桥 2020-09-06
大象从不倒下 2020-07-31
AlisaClass 2020-07-19
MaureenChen 2020-04-21
xingguanghai 2020-03-13
teresalxm 2020-02-18
木四小哥 2013-05-14
SoShellon 2013-06-01
Simagle 2013-05-31
羽化大刀Chrome 2013-05-31
waterv 2020-01-08
LutosX 2013-07-29
vanturman 2013-06-27
wutongyuq 2013-04-12
luoqu 2013-04-10
today0 2020-09-22
89520292 2020-09-18
bigname 2020-08-25