Zookeeper 分布式技术入门

1. 是一个 leader 和多个 follower 来组成的集群
2. 集群中只要有半数以上的节点存活，Zookeeper 就能正常工作（5 台服务器挂 2 台，没问题；4台服务器挂 2 台，就停止）
3. 全局数据一致性，每台服务器都保存一份相同的数据副本，无论 client 连接哪台 server，数据都是一致的
4. 数据更新原子性，一次数据要么成功，要么失败
5. 实时性，在一定时间范围内，client 能读取到最新数据
6. 更新的请求按照顺序执行，会按照发送过来的顺序，逐一执行

tar -zxvf apache-zookeeper-3.6.0-bin.tar.gz

mv apache-zookeeper-3.6.0-bin zookeeper

mkdir zkData
mkdir zkLog

cp zoo_sample.cfg zoo.cfg

dataDir=/opt/zookeeper/zkData
dataLogDir=/opt/zookeeper/zkLog

./zkServer.sh start

jps

./zkServer.sh status

./zkCli.sh

quit

vim myid

########### cluster ##############
server.1=192.168.186.128:2888:3888
server.2=192.168.186.129:2888:3888
server.3=192.168.186.130:2888:3888

firewall-cmd --zone=public --add-port=2181/tcp --permanent
firewall-cmd --zone=public --add-port=2888/tcp --permanent
firewall-cmd --zone=public --add-port=3888/tcp --permanent
firewall-cmd --reload

systemctl stop firewalld.service

./zkServer.sh start

./zkServer.sh status

ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Error contacting service. It is probably not running.

[root@localhost zookeeper]# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower

[root@localhost zookeeper]# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader

./zkCli.sh

help

ls /

ls -s /

[zookeeper]
cZxid = 0x0
ctime = Wed Dec 31 16:00:00 PST 1969
mZxid = 0x0
mtime = Wed Dec 31 16:00:00 PST 1969
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

create /china
create /usa

create /ru "putin"

create /japan/Tokyo "hot"

get /japan/Tokyo

create -e /uk
ls /
# 创建成功之后，quit 退出客户端
quit
# 重新连接，短暂的节点消失
ls /

# 在俄罗斯 ru 下，创建 3 个 city
create -s /ru/city
create -s /ru/city
create -s /ru/city
ls /ru
# [city0000000000, city0000000001, city0000000002]

set /japan/Tokyo "so hot"

addWatch /usa

set /usa "Trump"

WatchedEvent state:SyncConnected type:NodeDataChanged path:/usa

create /usa/NewYork

WatchedEvent state:SyncConnected type:NodeCreatedpath:/usa/NewYork

delete /usa/NewYork

deleteall /ru

<dependencies>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-core</artifactId>
        <version>2.8.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.6.0</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
</dependencies>

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/zk.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

public class TestZK {

    // zookeeper 集群的 ip 和端口
    private String connectString = "192.168.186.128:2181,192.168.186.129:2181,192.168.186.130:2181";

    /**
     * session 超时的时间: 时间不宜设置太小。
     *  因为 zookeeper 和加载集群环境会因为性能等原因而延迟略高，
     *  如果时间太少，还没有创建好客户端，就开始操作节点。会报错的。
     */
    private int sessionTimeout =  * ;

    // zookeeper 客户端对象
    private ZooKeeper zkClient;

    @Test
    public void init() throws Exception {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent watchedEvent) {
                System.out.println("得到监听反馈，再进行的业务处理代码！");
                System.out.println(watchedEvent.getType());
            }
        });
    }

}

- 表示哪个/哪些范围的 Id（Who）在通过了怎样的鉴权（How）之后，就允许进行那些操作（What）
- permission（What）就是一个 int 表示的位码，每一位代表一个对应操作的允许状态。
- 类似 linux 的文件权限，不同的是共有 5 种操作：CREATE、READ、WRITE、DELETE、
- ADMIN(对应更改ACL的权限)：
        + OPEN_ACL_UNSAFE：创建开放节点，允许任意操作 （用的最多，其余的权限用的很少）
        + READ_ACL_UNSAFE：创建只读节点
        + CREATOR_ALL_ACL：创建者才有全部权限

@Before
public void init() throws Exception {
    zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            System.out.println("得到监听反馈，再进行的业务处理代码");
            System.out.println(watchedEvent.getType());
        }
    });
}

@Test
public void createNode() throws Exception {
    String str = zkClient.create("/renda", "node1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println("已创建节点: " + str);
}

@Test
public void getNodeData() throws Exception {
    byte[] bytes = zkClient.getData("/renda", false, new Stat());
    String str = new String(bytes);
    System.out.println("/renda 节点的数据：" + str);
}

@Test
public void updateData() throws Exception {
    Stat stat = zkClient.setData("/renda", "nodeA".getBytes(), );
    System.out.println(stat);
}

@Test
public void delete() throws Exception {
    zkClient.delete("/renda", );
    System.out.println("删除成功");
}

@Test
public void getChildren() throws Exception {
    List<String> list = zkClient.getChildren("/china", false);
    for (String child : list) {
        System.out.println(child);
    }
}

@Test
public void watchNode() throws Exception{
    List<String> list = zkClient.getChildren("/", true);
    for (String s : list) {
        System.out.println(s);
    }
    // 让线程无限的等待下去
    System.in.read();
}

@Test
public void exists()throws Exception{
    Stat stat = zkClient.exists("/renda", false);
    if(stat==null)
        System.out.println("不存在");
    else
        System.out.println("存在");
}

public class ShopServer {

    private String connectString = "192.168.186.128:2181,192.168.186.129:2181,192.168.186.130:2181";

    private int sessionTimeout =  * ;

    private ZooKeeper zkClient;

    /**
     * 创建客户端，连接到 zookeeper
     */
    public void connect() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println(watchedEvent.getType());
            }
        });
    }

    /**
     * 注册到 zookeeper
     */
    public void register(String shopName) throws KeeperException, InterruptedException {
        // 要创建 EPHEMERAL_SEQUENTIAL 临时有序的节点（营业）
        // 可以自动编号，而要断开时，节点自动删除（打样）
        String s = zkClient.create("/meituan/shop", shopName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(shopName + " 开始营业，Path：" + s);
    }

    /**
     * 业务逻辑处理
     */
    private void business(String shopName) throws IOException {
        System.out.println(shopName + " 正在营业中");
        System.in.read();
    }

    /**
     * Main 方法
     */
    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        // 1.开一个饭店
        ShopServer shopServer = new ShopServer();

        // 2. 连接 zookeeper 集群（和美团取得联系）
        shopServer.connect();

        // 3.将服务节点注册到 zookeeper（入住美团）
        shopServer.register(args[]);

        // 4.业务逻辑处理（做生意）
        shopServer.business(args[]);
    }

}

public class Customers {

    private String connectString = "192.168.186.128:2181,192.168.186.129:2181,192.168.186.130:2181";

    private int sessionTimeout =  * ;

    private ZooKeeper zkClient;

    /**
     * 创建客户端，连接到 zookeeper
     */
    public void connect() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    // 一旦有数据变化就重新获取商家列表
                    getShopList();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 获取子节点列表（获取商家列表）
     */
    private void getShopList() throws KeeperException, InterruptedException {
        // 获取服务器的子节点信息，并且对父节点进行监听
        List<String> shops = zkClient.getChildren("/meituan", true);
        // 声明存储服务器信息的集合
        ArrayList<String> shopList = new ArrayList<String>();
        for (String shop : shops) {
            byte[] bytes = zkClient.getData("/meituan/" + shop, false, new Stat());
            shopList.add(new String(bytes));
        }
        System.out.println("目前正在营业的商家：" + shopList);
    }

    /**
     * 业务逻辑处理
     */
    private void business() throws IOException {
        System.out.println("用户正在浏览商家");
        System.in.read();
    }

    /**
     * Main 方法
     */
    public static void main(String[] args) throws KeeperException, InterruptedException, IOException {
        Customers client = new Customers();
        // 1.获得 zookeeper 的连接 （用户打开美团 APP）
        client.connect();
        // 2.获取 meituan 下的所有子节点列表（获取商家列表）
        client.getShopList();
        // 3.业务逻辑处理（对比商家，下单点餐）
        client.business();
    }

}

create /meituan/KFC "KFC"
create /meituan/BKC "BurgerKing"
create /meituan/baozi "baozi"

delete /meituan/baozi

- 获取锁（创建节点） -> 锁是否已经创建

- 如果锁已经被创建，则监听/等待（addWatch /exclusive/lock）
- 如果锁没有被创建，则创建 lock 临时节点（create -e /exclusive/lock）

- 如果创建 lock 临时节点失败，则进入监听/等待
- 如果创建 lock 临时节点成功，则占用锁，然后完成业务处理，然后释放锁（delete /exclusive/lock）

/lock - [/x0000001, /x0000002, /x0000003, /x0000004]

1. 所有请求进来，在 /lock 下创建临时顺序节点 ，zookeeper 会自动编号排序
2. 判断自己是不是 /lock 下最小的节点
        - 是，获得锁（创建节点）
        - 否，对前面小一级的节点进行监听
3. 获得锁请求，处理完业务逻辑，释放锁（删除节点），后一个节点得到通知
4. 重复步骤 2

CREATE DATABASE IF NOT EXISTS `zkproduct` DEFAULT CHARACTER SET utf8mb4;

-- 商品表
DROP TABLE IF EXISTS `product`;
CREATE TABLE `product` (
  `id` INT PRIMARY KEY AUTO_INCREMENT, # 商品编号
   `product_name` VARCHAR (20) NOT NULL, # 商品名称
   `stock` INT NOT NULL, # 库存
   `version` INT NOT NULL # 版本
);

INSERT INTO product (`product_name`, `stock`, `version`)
VALUES
  ('锦鲤-清空购物车-大奖',5,0);

-- 订单表
 DROP TABLE IF EXISTS `order`;
 CREATE TABLE `order` (
  `id` VARCHAR (100) PRIMARY KEY, # 订单编号
   `pid` INT NOT NULL, # 商品编号
   `userid` INT NOT NULL # 用户编号
);

-- 查询
SELECT * FROM `product`;
SELECT * FROM `order`;

- controller
    + ProductAction
- mapper
    + OrderMapper
    + ProductMapper
- models
    + Order
    + Product
- service
    + impl/OrderServiceImpl
    + OrderService

<groupId>com.renda</groupId>
<artifactId>zk_product</artifactId>
<version>1.0-SNAPSHOT</version>

<packaging>war</packaging>

<!-- 指定编码及版本 -->
<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.encoding>UTF-8</maven.compiler.encoding>
    <java.version>1.11</java.version>
    <maven.compiler.source>11</maven.compiler.source>
    <maven.compiler.target>11</maven.compiler.target>
    <spring.version>5.2.7.RELEASE</spring.version>
</properties>

<dependencies>
    <!-- Spring -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-context</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-webmvc</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jdbc</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <!-- Mybatis -->
    <dependency>
        <groupId>org.mybatis</groupId>
        <artifactId>mybatis</artifactId>
        <version>3.5.5</version>
    </dependency>
    <dependency>
        <groupId>org.mybatis</groupId>
        <artifactId>mybatis-spring</artifactId>
        <version>2.0.5</version>
    </dependency>
    <!-- 连接池 -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid</artifactId>
        <version>1.1.10</version>
    </dependency>
    <!-- 数据库 -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.20</version>
    </dependency>
    <!-- junit -->
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>
</dependencies>

<build>
    <plugins>
        <!-- maven 内嵌的 tomcat 插件 -->
        <plugin>
            <groupId>org.apache.tomcat.maven</groupId>
            <!-- 目前 apache 只提供了 tomcat6 和 tomcat7 两个插件 -->
            <artifactId>tomcat7-maven-plugin</artifactId>
            <configuration>
                <port>8001</port>
                <path>/</path>
            </configuration>
            <executions>
                <execution>
                    <!-- 打包完成后,运行服务 -->
                    <phase>package</phase>
                    <goals>
                        <goal>run</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN" "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>

    <!-- 后台的日志输出：针对开发者 -->
    <settings>
        <setting name="logImpl" value="STDOUT_LOGGING"/>
    </settings>

</configuration>

<?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" xmlns:tx="http://www.springframework.org/schema/tx"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/tx
       http://www.springframework.org/schema/tx/spring-tx.xsd">

    <!-- 1.扫描包注解 -->
    <context:component-scan base-package="com.renda.controller,com.renda.service,com.renda.mapper"/>
    <!-- 2.数据连接池 -->
    <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource">
        <property name="url" value="jdbc:mysql://192.168.186.128:3306/zkproduct?serverTimezone=GMT"/>
        <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
        <property name="username" value="root"/>
        <property name="password" value="RendaZhang@666"/>
        <property name="maxActive" value="10"/>
        <property name="minIdle" value="5"/>
    </bean>
    <!-- 3.sqlSessionFactory -->
    <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
        <property name="dataSource" ref="dataSource"/>
        <property name="configLocation" value="classpath:mybatis/mybatis-config.xml"/>
    </bean>
    <!-- 4.事务管理器 -->
    <bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="dataSource"/>
    </bean>
    <!-- 5.开启事务 -->
    <tx:annotation-driven/>

    <!-- MyBatis 扫描 Mapper -->
    <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
        <property name="basePackage" value="mapper"/>
    </bean>

</beans>

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee
http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
         version="3.1">

    <servlet>
        <servlet-name>springMVC</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:spring/spring.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
        <async-supported>true</async-supported>
    </servlet>
    <servlet-mapping>
        <servlet-name>springMVC</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>

</web-app>

@Mapper
@Component
public interface OrderMapper {
    /**
     * 生成订单
     */
    @Insert("insert into `order` (id,pid,userid) values (#{id},#{pid},#{userid})")
    int insert(Order order);
}

@Mapper
@Component
public interface ProductMapper {
    /**
     * 查询商品（目的查询库存）
     */
    @Select("select * from product where id = #{id}")
    Product getProduct(@Param("id") int id);
    /**
     * 减库存
     */
    @Update("update product set stock = stock-1 where id = #{id}")
    int reduceStock(@Param("id") int id);
}

public interface OrderService {
    /**
     * 减库存
     */
    void reduceStock(int id) throws Exception;
}

@Service
public class OrderServiceImpl implements OrderService {

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private OrderMapper orderMapper;

    @Override
    public void reduceStock(int id) throws Exception {
        // 1.获取库存（根据商品 id 查询商品）
        Product product = productMapper.getProduct(id);
        if (product.getStock() <= ) {
            throw new RuntimeException("仓库已无商品");
        }
        // 2.减库存
        int i = productMapper.reduceStock(id);
        if (i == ) {
            // 3.生成订单
            Order order = new Order();
            // 使用 UUID 工具生一个订单号
            order.setId(UUID.randomUUID().toString());
            order.setPid(id);
            order.setUserid();
            orderMapper.insert(order);
        } else {
            throw new RuntimeException("减库存失败！");
        }
    }

}

@Controller
public class ProductAction {

    @Autowired
    private OrderService orderService;

    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduce(int id) throws Exception {
        orderService.reduceStock(id);
        return "ok";
    }

}

upstream renda {
    server 192.168.1.116:8001;
    server 192.168.1.116:8002;
}

server {
    listen       80;
    server_name  www.zookeeper.com;
    location / {
        proxy_pass http://renda;
        index  index.html index.htm;
    }
}

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.2.0</version>
</dependency>

@Controller
public class ProductAction {

    private String zookeeperConnectionString = "192.168.186.128:2181,192.168.186.129:2181,192.168.186.130:2181";

    @Autowired
    private OrderService orderService;

    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduce(int id) throws Exception {
        // 1.创建 curator 工具对象：重试策略 （1000毫秒试1次，最多试3次）
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(, );
        CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
        client.start();

        // 2.根据工具对象创建“内部互斥锁”
        InterProcessMutex lock = new InterProcessMutex(client, "/product_" + id);
        // 3.加锁
        lock.acquire();
        try {
            // 临界区
            orderService.reduceStock(id);
        } catch (Exception e) {
            if (e instanceof RuntimeException) {
                throw e;
            }
        } finally {
            //4.释放锁
            lock.release();
        }

        return "ok";
    }

}