Zookeeper实现分布式应用的(主节点HA)及客户端动态更新主节点状态

某分布式系统中，主节点可以有多台，可以动态上下线
任意一台客户端都能实时感知到主节点服务器的上下线
服务端逻辑：

服务端启动时在zookeeper集群中 /services节点写入一个临时节点，该临时节点内容为该节点的名称。当服务端下线或者宕机时zookeeper会将该节点删除
1

客户端逻辑

客户端启动时会查找/services 下所有节点 并将节点名输出，当节点变化时，会重新查找所有节点，并将节点名输出
1

import org.apache.zookeeper.CreateMode
import org.apache.zookeeper.Watcher
import org.apache.zookeeper.ZooDefs
import org.apache.zookeeper.ZooKeeper

/**
 * 服务端
 */
object DistributedServer {
    //逗号分隔主机：端口对，每个对应一个ZooKeeper 主机名/IP 与配置文件相同
    private val CONNECT_STRING: String = "192.168.84.132:2181,192.168.84.133:2181,192.168.84.134:2181"
    //会话超时（毫秒）
    private val SESSION_TIMEOUT: Int = 2000
    //获取zooKeeper客户端连接
    private val zooKeeper: ZooKeeper by lazy {
        ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, Watcher { event ->
            //Watch回调
            println("${event?.type}-->${event?.path}")
            //创建监听节点变化，非数据
            zooKeeper.getChildren("/", true)
        })
    }

    @JvmStatic
    fun main(args: Array<String>) {
        val serviceName = args[0]
        zooKeeper.create("/services/service", serviceName.toByteArray(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL)
        println("serviceName --> $serviceName create")
        println("$serviceName start working.....")
        Thread.sleep(Long.MAX_VALUE)
    }
}
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

import org.apache.zookeeper.Watcher
import org.apache.zookeeper.ZooKeeper
import org.apache.zookeeper.data.Stat

/**
 * 客户端
 */
object DistributedClient {

    //逗号分隔主机：端口对，每个对应一个ZooKeeper 主机名/IP 与配置文件相同
    private val CONNECT_STRING: String = "192.168.84.132:2181,192.168.84.133:2181,192.168.84.134:2181"
    //会话超时（毫秒）
    private val SESSION_TIMEOUT: Int = 2000
    private val zooKeeper: ZooKeeper by lazy {
        ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, Watcher { event ->
            println("${event?.type}-->${event?.path}")
            getServiceList()
        })

    }
    //Volatile
    @Volatile private lateinit var services: ArrayList<String>

    @JvmStatic
    fun main(args: Array<String>) {
        getServiceList()
        println("client start working.....")
        Thread.sleep(Long.MAX_VALUE)
    }

    private fun getServiceList() {
        //每次/services子节点变化都会调用watch
        val serviceList = zooKeeper.getChildren("/services", true)
        //设置临时集合存储所有service 信息
        val tempList = arrayListOf<String>()
        serviceList
                //取每个service的信息并存入临时集合
                .map { zooKeeper.getData("/services/" + it, false, Stat())}
                .mapTo(tempList) { String(it) }
        services = tempList
        println(services)
    }
}
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

服务端

客户端

服务端上下线 客户端更新