Dawn's Blogs

分享技术 记录成长

0%

发表于 更新于 分类于

用 Rand7() 实现 Rand10()

用 Rand7() 实现 Rand10()

解题思路

原理

1
2
3
4
5
6
已知 rand_N() 可以等概率的生成[1, N]范围的随机数
那么：
(rand_X() - 1) × Y + rand_Y() ==> 可以等概率的生成[1, X * Y]范围的随机数，即实现了rand_XY()


rand_N() % Y +1 ==> 只要N是Y的倍数，就可以实现利用rand_N()实现rand_Y()

要实现rand10()，就需要实现rand_N()，并且保证N大于10且是10的倍数。这样就可以通过rand_N() % 10 + 1得到[1, 10]范围内的随机数了。

(rand7()-1) × 7 + rand7() ==> rand49()，但是49不是10的倍数。这里就涉及到了“拒绝采样”的知识了，也就是说，如果某个采样结果不在要求的范围内，则丢弃它。

解题代码

1
2
3
4
5
6
7
8
func rand10() int {
    for {
        num := (rand7()-1) * 7 + rand7()    // rand49()
        if num <= 40 {  // 拒绝采样
            return num % 10 + 1
        }
    }
}
  • 优化
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
func rand10() int {
    for {
        a := rand7()
        b := rand7()
        num := (a-1) * 7 + b    // rand49
        if num <= 40 {  // 拒绝采样
            return num % 10 + 1
        }

        a = num - 40    // rand9
        b = rand7()
        num = (a-1) * 7 + b      // rand63
        if num <= 60 {
            return num % 10 + 1
        }

        a = num - 60    // rand3
        b = rand7()
        num = (a-1) * 7 + b      // rand21
        if num <= 20 {
            return num % 10 + 1
        }
    }
}
阅读全文 »

发表于 更新于 分类于

生产者

选择Partition原则

在kafka中，一个topic指定了多个partition，生产者选择partition的方式：

  1. partition在写⼊的时候可以指定需要写⼊的partition，如果有指定，则写⼊对应的partition。
  2. 如果没有指定partition，但是设置了数据的key，则会根据key的值hash出⼀个partition。
  3. 如果既没指定partition，⼜没有设置key，则会采⽤轮询⽅式，即每次取⼀小段时间的数据写⼊某个partition，下⼀⼩段的时间写⼊下⼀个partition。

同步发送和异步发送

同步发送

⽣产者同步发消息，在收到kafka的ACK告知发送成功之前⼀直处于阻塞状态。

异步发送

⽣产者发消息，发送完后不用等待broker给回复，直接执⾏下面的业务逻辑。

注意：异步发送可能会丢失消息。

生产者ACK机制

同步发送的前提下，生产者向集群发送消息时，关于ACK应答机制有3个配置：

  • 0代表producer往集群发送数据不需要等待集群的返回。也就是说，集群不需要任何broker收到消息，就立刻返回ACK给生产者。不确保消息发送成功。安全性最低但是效率最⾼。
  • 1代表producer往集群发送数据只要leader应答（Leader将消息写入本地日志中）就可以发送下⼀条，只确保leader发送成功。默认选项。
  • all代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下⼀条，确保leader发送成功和所有的副本都完成备份。安全性最⾼，但是效率最低。
阅读全文 »

发表于 更新于 分类于

副本

副本是为了为主题中的分区创建多个备份，多个副本在kafka集群的多个broker中，会有⼀个副本作为leader，其他是follower。

下图是拥有2个分区，3个副本的主题：

副本

  • leader：kafka的写和读的操作，都发⽣在leader上。leader负责把数据同步给follower。当leader挂了，经过主从选举，从多个follower中选举产⽣⼀个新的leader。
  • follower：接收leader同步的数据，在leader宕机后参与选举。
  • isr：可以同步和已同步的节点会被存入到isr集合中。如果isr中的节点性能较差，会被踢出isr集合（不能参与选举）。

集群消费

集群消费

图中Kafka集群有两个broker，每个broker中有多个partition。

  • 一个partition只能被一个消费组里的某一个消费者消费，从而保证消费顺序。

  • Kafka只在partition的范围内保证消息消费的局部顺序性，不能在同⼀个topic中的多个partition中保证总的消费顺序性。

  • 一个消费者可以消费多个partition，但是消费者组中的消费者数量不能比一个topic中的partition数量多，否则多出来的消费者消费不到消息。

  • 如果消费者宕机，会触发rebalance机制，让其他消费者来消费该分区。

发表于 更新于 分类于

对角线遍历

对角线遍历

解题思路

划分为两种遍历形式：朝着右上方遍历和朝着左下方遍历。

解题代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
func findDiagonalOrder(mat [][]int) []int {
    m := len(mat)
    n := len(mat[0])
    res := make([]int, m*n)     // 记录遍历序列
    count := 0  // 记录已经遍历的元素个数
    row, col := 0, 0    // 待遍历元素所在行、列
    for count < m*n {
        if row == m-1 && col == n-1 {
            // 遍历到最后一个元素（右下角的元素）
            res [count] = mat[row][col]
            break
        }
        // 向斜右上方遍历
        for row >= 0 && col < n {
            res[count] = mat[row][col]
            count++
            row--
            col++
        }
        if col < n {
            // 在遍历矩阵的左上部分
            row++
        } else {
            // 在遍历矩阵的右下部分
            row += 2
            col--
        }
        // 向斜左下方遍历
        for row < m && col >= 0 {
            res[count] = mat[row][col]
            count++
            row++
            col--
        }
        if row < m {
            // 在遍历矩阵的左上部分
            col++
        } else {
            // 在遍历矩阵的右下部分
            col += 2
            row--
        }
    }

    return res
}

长度最小的子数组

长度最小的子数组

解题思路

利用滑动窗口的思想，start指向窗口的起始位置，end指向窗口的终止位置的后一位，sum为窗口内数字之和：

  • 如果sum < target，end需要向后移动
  • 如果sum > target，start需要向前移动

解题代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
func minSubArrayLen(target int, nums []int) int {
    start, end := 0, 0  // 滑动窗口的起始位置、终止位置的后一位
    minLen := math.MaxInt64     // 连续子数组的最小长度
    sum := 0
    for end < len(nums) {
        // end向后移动
        for sum < target && end < len(nums) {
            sum += nums[end]
            end++    
        }
        // start向前移动
        for sum >= target {
            if end-start < minLen {
                minLen = end-start
            }
            sum -= nums[start]
            start++
        }
    }

    if minLen == math.MaxInt64 {
        // 不存在符合条件的子数组
        return 0
    } else {
        // 存在
        return minLen
    }
}

删除二叉搜索树中的节点

删除二叉搜索树中的节点

解题思路

  • 当待删除节点是叶子节点时，直接删除

  • 待删除节点只有一个分支时，用其分支节点代替待删除的节点

  • 待删除节点的左右子树均存在时：

    • 找到其后继节点，交换后继节点和当前待删除节点的值
    • 删除右子树中值为key的节点

解题代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
/**
 * Definition for a binary tree node.
 * type TreeNode struct {
 *     Val int
 *     Left *TreeNode
 *     Right *TreeNode
 * }
 */
func deleteNode(root *TreeNode, key int) *TreeNode {
    if root != nil {
        if root.Val == key {
            // 找到待删除节点
            if root.Left == nil && root.Right == nil {
                // 叶子节点直接删除
                return nil
            }
            if root.Left != nil && root.Right == nil {
                // 只有左子树，用左子树代替
                return root.Left
            }
            if root.Right != nil && root.Left == nil {
                // 只有右子树，用右子树代替
                return root.Right
            }
            // 左右子树都存在
            // 找到后继节点
            next := nextNode(root)
            // 交换后继节点与当前节点的值
            root.Val, next.Val = next.Val, root.Val
            // 在右子树中删除值为key的节点
            root.Right = deleteNode(root.Right, key)
        } else if root.Val > key {
            // 进入左子树查找
            root.Left =  deleteNode(root.Left, key)
        } else {
            // 进入右子树查找
            root.Right = deleteNode(root.Right, key)
        }
    }

    return root
}

// 返回以root为根节点，中序遍历中的后继节点
func nextNode(root *TreeNode) *TreeNode {
    // 后继节点为root右子树的最左边的节点
    next := root.Right
    for next.Left != nil {
        next = next.Left
    }
    return next
}

发表于 更新于 分类于

CAP理论

CAP定理

CAP理论为：⼀个分布式系统最多只能同时满足⼀致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance）这三项中的两项。

  • 一致性（Consistency）：即更新操作成功并返回客户端后，所有节点在同一时间的数据完全一致。
  • 可用性（Availability）：服务⼀直可用，而且是正常响应时间。
  • 分区容错性（Partition tolerance）：分布式系统在遇到某节点或网络分区故障的时候，仍然能够对外提供满足一致性或者可用性的服务。

BASE理论

BASE理论是对CAP理论的延申，核心思想就是即使无法做到强一致性（CAP的一致性就是强一致性），但可以采取适当的方式达到最终一致性。

  • 基本可用（Basically Available）：指分布式系统在出现故障的时候，允许损失部分可用性，即保证核心可用。
  • 软状态（Soft State）：允许系统存在中间状态，而该中间状态不会影响系统整体可⽤性。分布式存储中⼀般⼀份数据至少会有三个副本，允许不同节点间副本同步的延时就是软状态的体现。
  • 最终一致性（Eventual Consistency）：指系统中的所有数据副本经过⼀定时间后，最终能够达到⼀致的状态。

ZooKeeper追求的一致性

ZooKeeper保证的最终一致性又叫做顺序一致性，即每个结点的数据都是严格按事务的发起顺序生效的。

ZooKeeper如何保证事务的顺序

zookeeper中的事务ID为ZXID，ZXID由Leader节点生成，有新写入事件时，Leader生成新ZXID并随提案一起广播，每个结点本地都保存了当前最近一次事务的ZXID，ZXID是递增的，所以谁的ZXID越大，就表示谁的数据是最新的。

ZXID的生成规则如下：

ZXID

ZXID由两部分组成：

  • 任期：完成本次选举后，直到下次选举前，由同一Leader负责协调写入
  • 事务计数器：单调递增，每生效一次写入，计数器加一

ZXID的低32位是计数器，所以同一任期内，ZXID是连续的，每个结点又都保存着自身最新生效的ZXID，通过对比新提案的ZXID与自身最新ZXID是否相差1，来保证事务严格按照顺序生效的。

发表于 更新于 分类于

消息队列

应用场景

  • 缓存/消峰：将消息暂存在消息队列中，以生产消息和消费消息的处理速度不一致的情况。
  • 解耦：可以修改或者扩充生产者和消费者，只需要遵守消息格式的一致即可。

消息队列解耦

  • 异步通信：允许用户把一个消息放入队列，但并不立即处理它，然后在需要的时候再去处理它们。

消息队列异步通信

消息队列两种模式

点对点模式

类似于队列，消费者消费消息后会清除消息。

发布/订阅模式

消息⽣产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。和点对点⽅式不同，发布到topic的消息会被所有订阅者消费。

Kafka介绍

Kafka是一种高吞吐量的分布式发布订阅消息系统。

同时Kafka结合点对点模型和发布订阅模型：

  • 对于同一个消费者组Consumer Group，一个message只有被一个consumer消费 - 点对点模型
  • 同一个消息可以被广播到不同的Consumer Group中 - 发布订阅模型

架构

Kafka架构

  • Producer：Producer即⽣产者，消息的产⽣者，是消息的⼊⼝。
  • kafka cluster：kafka集群，⼀台或多台服务器组成
    • Broker：Broker是指部署了Kafka实例的服务器节点。每个服务器上有⼀个或多个kafka的实例，我们姑且认为每个broker对应⼀台服务器。每个kafka集群内的broker都有⼀个不重复的编号。
    • Topic：消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。实际应⽤中通常是⼀个业务线建⼀个topic。
    • Partition：Topic的分区，每个topic可以有多个分区，分区的作⽤是做负载，提⾼kafka的吞吐量。同⼀个topic在不同的分区的数据是不重复的，partition的表现形式就是⼀个⼀个的⽂件夹。为了提高可靠性，提出分区的副本，分为Leader和Follower，生产和消费只针对Leader。
    • Replication：每⼀个分区都有多个副本，副本的作⽤是做备份。当主分区（Leader）故障的时候会选择⼀个备胎（Follower）上位，成为Leader。在kafka中默认副本的最⼤数量是10个，且副本的数量不能⼤于Broker的数量，follower和leader绝对是在不同的机器，同⼀机器对同⼀个分区也只可能存放⼀个副本（包括自己）。
  • Consumer：消费者，即消息的消费⽅，是消息的出⼝。
    • Consumer Group：我们可以将多个消费组组成⼀个消费者组，在kafka的设计中同⼀个分区的数据只能被消费者组中的某⼀个消费者消费。同⼀个消费者组的消费者可以消费同⼀个topic的不同分区的数据，这也是为了提⾼kafka的吞吐量！
阅读全文 »

发表于 更新于 分类于

zookeeper集群中的节点共有三种角色：

  • Leader：处理集群中的所有事务请求，包括读和写，集群中只有一个Leader
  • Follower：只处理读请求，可以参与Leader选举
  • Observer：只处理读请求，但是不能参与Leader选举

ZAB协议

介绍

zookeeper作为⾮常重要的分布式协调组件，需要进行集群部署，集群中通常会以一主多从的方式进行部署。zookeeper为了保证数据的一致性，使用ZAB（zookeeper atomic broadcast）协议，这一协议解决了zookeeper的崩溃恢复和主从数据同步的问题。

zookeeper集群

ZAB协议的四种节点状态

ZAB中协议定义了四种节点状态：

  • Looking：选举状态
  • Following：Follower节点的状态
  • Leading：Leader节点的状态
  • Observing：Observer节点的状态

Leader选举过程

集群上线时的Leader选举

若进行Leader选举，则至少需要两台机器，这里选取3台机器组成的服务器集群为例。在集群初始化阶段，当有一台服务器Server1启动时，其单独无法进行和完成Leader选举，当第二台服务器Server2启动时，此时两台机器可以相互通信，每台机器都试图找到Leader，于是进入Leader选举过程。选举过程如下：

  1. 每个Server发出一个投票。由于是初始情况，Server1和Server2都会将自己作为Leader服务器来进行投票，每次投票会包含所推举的服务器的myid和ZXID，使用(myid, ZXID)来表示，其中myid为服务器的唯一标识，ZXID为事务号。此时Server1的投票为(1, 0)，Server2的投票为(2, 0)，然后各自将这个投票发给集群中其他机器。

  2. 接受来自各个服务器的投票。集群的每个服务器收到投票后，首先判断该投票的有效性，如检查是否是本轮投票、是否来自LOOKING状态的服务器。

  3. 处理投票。针对每一个投票，服务器都需要将别人的投票和自己的投票进行PK，PK规则如下：

    • 优先检查ZXID。ZXID比较大的服务器优先作为Leader。
    • 如果ZXID相同，那么就比较myid。myid较大的服务器作为Leader服务器。

    对于对于Server1而言，它的投票是(1, 0)，接收Server2的投票为(2, 0)，首先会比较两者的ZXID，均为0，再比较myid，此时Server2的myid最大，于是更新自己的投票为(2, 0)，然后重新投票，对于Server2而言，其无须更新自己的投票，只是再次向集群中所有机器发出上一次投票信息即可。

  4. 统计投票。每次投票后，服务器都会统计投票信息，判断是否已经有过半机器接受到相同的投票信息，对于Server1、Server2而言，都统计出集群中已经有两台机器接受了(2, 0)的投票信息，此时便认为已经选出了Leader。

  5. 改变服务器状态。一旦确定了Leader，每个服务器就会更新自己的状态。

崩溃恢复时的Leader选举

Leader建立完成后，Leader会周期性的不断向Follower发送心跳。当Leader崩溃后，Follower会进入Looking状态，重新进行Leader选举，此时集群不能对外服务。

主从服务器之间的数据同步

  1. 客户端向Leader节点写数据
  2. Leader节点把数据写到自己的数据文件中，并给自己返回一个ACK
  3. Leader把数据发送给Follower
  4. Follower将数据写到本地数据文件中
  5. Follower返回ACK给Leader节点
  6. Leader收到半数以上的ACK后，向Follower发送Commit
  7. 从节点收到Commit后把数据文件中的数据写到内存中

zookeeper数据同步

发表于 更新于 分类于

基本操作

连接zookeeper

1
2
3
4
5
6
// 连接zookeeper服务器
func initZk() (*zk.Conn, error) {
	servers := []string{"127.0.0.1:2181"}
	conn, _, err := zk.Connect(servers, time.Second*5)
	return conn, err
}

增

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/*
conn zookeeper连接
path 节点
data 节点数据
flags有4种取值：
   0:永久，除非手动删除
   zk.FlagEphemeral = 1:短暂，session断开则改节点也被删除
   zk.FlagSequence  = 2:会自动在节点后面添加序号
   3:Ephemeral和Sequence，即，短暂且自动添加序号
*/
func add(conn *zk.Conn, path string, data []byte, flags int32) error {
	acls := zk.WorldACL(zk.PermAll)
	s, err := conn.Create(path, data, flags, acls)
	if err != nil {
		log.Println("create failed, err:", err)
	} else {
		log.Println("create success:", s)
	}
	return err
}

删

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 删除
func delete(conn *zk.Conn, path string) error {
	// 获取版本
	_, stat, err := conn.Get(path)
	if err != nil {
		log.Println("get version failed, err:", err)
		return err
	}
	// 删除
	err = conn.Delete(path, stat.Version)
	if err != nil {
		log.Printf("delete failed for path %s, err: %v\n", path, err)
	}
	return err
}

改

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 更新节点
func set(conn *zk.Conn, path string, data []byte) error {
	// 获取版本
	_, stat, err := conn.Get(path)
	if err != nil {
		log.Println("get version failed, err:", err)
		return err
	}
	// 更新
	stat, err = conn.Set(path, data, stat.Version)
	if err != nil {
		log.Printf("set failed for path %s, err: %v\n", path, err)
	} else {
		log.Printf("set success ,stat: %v\n", stat)
	}
	return err
}

查

1
2
3
4
5
6
7
8
9
10
// 查询
func get(conn *zk.Conn, path string) ([]byte, error) {
	b, stat, err := conn.Get(path)
	if err != nil {
		log.Println("get failed, err:", err)
	} else {
		log.Printf("info of node for %v: %v\n", path, stat)
	}
	return b, err
}

Watch机制

只监听一次

调用zk.WithEventCallback(callback)设置回调

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package main

import (
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"log"
	"time"
)

// zk watch 回调函数
func callback(event zk.Event) {
	// zk.EventNodeCreated
	// zk.EventNodeDeleted
	fmt.Println("###########################")
	fmt.Println("path: ", event.Path)
	fmt.Println("type: ", event.Type.String())
	fmt.Println("state: ", event.State.String())
	fmt.Println("---------------------------")
}

func initZk() (*zk.Conn, error) {
	eventCallbackOption := zk.WithEventCallback(callback)
	servers := []string{"127.0.0.1:2181"}
	conn, _, err := zk.Connect(servers, time.Second*5, eventCallbackOption)
	return conn, err
}

func add(conn *zk.Conn, path string, data []byte, flags int32) error {
	acls := zk.WorldACL(zk.PermAll)
	_, err := conn.Create(path, data, flags, acls)

	return err
}

// 删除
func delete(conn *zk.Conn, path string) error {
	// 获取版本
	_, stat, err := conn.Get(path)
	if err != nil {
		return err
	}
	// 删除
	err = conn.Delete(path, stat.Version)
	return err
}

func listenOne(conn *zk.Conn, path string) error {
	//调用conn.ExistsW(path) 或GetW(path)为对应节点设置监听，该监听只生效一次
	_, _, _, err := conn.ExistsW(path)
	return err
}

func main() {
	conn, err := initZk()
	defer conn.Close()
	if err != nil {
		fmt.Println(err)
		return
	}

	path := "/dawn111"
	data := []byte("hello world")

	// 开始监听path，只监听一次
	err = listenOne(conn, path)
	if err != nil {
		log.Println(err)
	}
	// 触发创建数据操作
	err = add(conn, path, data, 0)
	if err != nil {
		log.Println("create failed, err:", err)
	} else {
		log.Println("创建数据成功！")
	}

	//再次监听path
	err = listenOne(conn, path)
	if err != nil {
		log.Println(err)
	}
	// 触发删除数据操作
	err = delete(conn, path)
	if err != nil {
		log.Printf("delete failed for path %s, err: %v\n", path, err)
	} else {
		log.Println("删除数据成功！")
	}

}

开启一个channel处理

可以开一一个协程来处理chanel中传来的event事件，可以持续监听

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
package main

import (
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"log"
	"time"
)

func initZk() (*zk.Conn, error) {
	servers := []string{"127.0.0.1:2181"}
	conn, _, err := zk.Connect(servers, time.Second*5)
	return conn, err
}

func add(conn *zk.Conn, path string, data []byte, flags int32) error {
	acls := zk.WorldACL(zk.PermAll)
	_, err := conn.Create(path, data, flags, acls)

	return err
}

// 删除
func delete(conn *zk.Conn, path string) error {
	// 获取版本
	_, stat, err := conn.Get(path)
	if err != nil {
		return err
	}
	// 删除
	err = conn.Delete(path, stat.Version)
	return err
}

func listenOneChannel(conn *zk.Conn, path string) error {
	_, _, ch, err := conn.ExistsW(path)

	go func() {
		event := <-ch
		fmt.Println("*******************")
		fmt.Println("path:", event.Path)
		fmt.Println("type:", event.Type.String())
		fmt.Println("state:", event.State.String())
		fmt.Println("-------------------")
	}()

	return err
}

func main() {
	conn, err := initZk()
	defer conn.Close()
	if err != nil {
		fmt.Println(err)
		return
	}

	path := "/dawn111"
	data := []byte("hello world")

	// 开启一个管道持续监听
	err = listenOneChannel(conn, path)
	if err != nil {
		log.Println(err)
	}
	// 创建数据触发事件
	err = add(conn, path, data, 0)
	if err != nil {
		log.Println("create failed, err:", err)
	} else {
		log.Println("创建数据成功！")
	}
	// 删除数据触发事件
	err = delete(conn, path)
	if err != nil {
		log.Printf("delete failed for path %s, err: %v\n", path, err)
	} else {
		log.Println("删除数据成功！")
	}

}

发表于 更新于 分类于

ZooKeeper实现分布式锁

zookeeper中可以实现两种锁：

  • 读锁（共享锁）：可以一起读，但是与写锁互斥。
  • 写锁（互斥锁）：与读锁和写锁都互斥。

实现读锁

  • 创建⼀个临时序号节点，节点的数据是read，表示是读锁
  • 获取当前zookeeper中序号比自己小的所有节点。若这些节点有写锁，则加锁不成功。

读锁

实现写锁

  • 创建⼀个临时序号节点，节点的数据是write，表示是写锁
  • 获取zookeeper中所有的子节点
  • 判断自己是否是最小的节点：
    • 如果是，则上锁成功
    • 如果不是，说明前面还有锁，上锁失败，为最小节点设置监听。阻塞等待，watch机制会当最小节点发生变化时通知当前节点，再执行第二步

写锁

羊群效应

羊群效应：如果采用上述的上锁方式，只要有节点发生变化，就会触发其他节点的监听事件，这样对于zookeeper的压力非常大。

可以调整成链式监听：

链式监听

阅读全文 »

发表于 更新于 分类于

ZooKeeper介绍

ZooKeeper是一种分布式协调服务，用于管理大型分布式集群。在分布式环境中协调和管理服务是一个复杂的过程，但是ZooKeeper通过其简单的架构和API解决了这个问题。

应用场景

分布式协调组件

在分布式系统中，需要zookeeper作为分布式协调组件，协调分布式系统中的状态。如下图，用户更改了一台服务器A-2上的状态，但是另一台主机A-1上的flag没有做出及时的更改，这就需要zookeeper进行协调，更改A-1的状态，以协调分布式系统中所有服务器的状态。

zk作为分布式协调组件

分布式锁

zookeeper在实现分布式锁上，可以做到强⼀致性。

分布式锁就是将锁放在zookeeper上，当需要上锁时到zookeeper中获取锁。

无状态化的实现

对于分布式系统，将Session或者Cookie等状态信息保存在各个主机上是不现实的。可以利用zookeeper实现分布式系统的无状态化，将登录信息统一保存在zookeeper中，各个分布式主机在zookeeper中获取状态信息。

zk作为无状态化的实现

阅读全文 »