Hướng dẫn tích hợp Zookeeper với SpringBoot

Zookeeper và ứng dụng trong hệ thống phân tán

Giới thiệu về Zookeeper

Zookeeper là một dịch vụ điều phối phân tán mã nguồn mở, được duy trì bởi Apache Software Foundation. Nó cung cấp các tính năng mạnh mẽ để quản lý thông tin cấu hình, dịch vụ đặt tên, đồng bộ hóa phân tán và nhóm dịch vụ trong môi trường phân tán.

Các đặc điểm chính của Zookeeper:

  • Tính nhất quán tuần tự: Các yêu cầu cập nhật sẽ được áp dụng theo thứ tự gửi.
  • Tính nguyên tử: Một thao tác hoặc thành công hoặc thất bại hoàn toàn.
  • Giao diện hệ thống đơn nhất: Tất cả các client kết nối đều thấy cùng một giao diện.
  • Tính tin cậy: Khi một bản cập nhật được áp dụng, nó sẽ tồn tại cho đến khi bị ghi đè.
  • Tính thời gian thực: Giao diện mà client nhìn thấy đảm bảo cập nhật trong khoảng thời gian xác định.

Ứng dụng điển hình của Zookeeper:

  1. Quản lý cấu hình tập trung.
  2. Khóa phân tán cho cơ chế loại trừ lẫn nhau giữa JVM.
  3. Theo dõi trạng thái các nút trong cụm.
  4. Dịch vụ đặt tên tương tự DNS.
  5. Đội ngũ phân phối đơn giản.
  6. Lựa chọn leader trong hệ thống phân tán.

Cài đặt và cấu hình Zookeeper

Cài đặt chế độ đơn lẻ

wget https://example.com/zookeeper-3.7.0.tar.gz
tar -zxvf zookeeper-3.7.0.tar.gz
cd zookeeper-3.7.0

echo "tickTime=2000" > conf/zoo.cfg
echo "dataDir=/tmp/zk_data" >> conf/zoo.cfg
echo "clientPort=2181" >> conf/zoo.cfg
echo "initLimit=10" >> conf/zoo.cfg
echo "syncLimit=5" >> conf/zoo.cfg

bin/zkServer.sh start
bin/zkCli.sh -server 127.0.0.1:2181

Cài đặt chế độ cụm

# Tạo file zoo.cfg trên mỗi nút:
echo "tickTime=2000" > conf/zoo.cfg
echo "dataDir=/var/lib/zk" >> conf/zoo.cfg
echo "clientPort=2181" >> conf/zoo.cfg
echo "initLimit=10" >> conf/zoo.cfg
echo "syncLimit=5" >> conf/zoo.cfg
echo "server.1=node1:2888:3888" >> conf/zoo.cfg
echo "server.2=node2:2888:3888" >> conf/zoo.cfg
echo "server.3=node3:2888:3888" >> conf/zoo.cfg

# Tạo file myid trên mỗi nút:
echo 1 > /var/lib/zk/myid # Trên node1
echo 2 > /var/lib/zk/myid # Trên node2
echo 3 > /var/lib/zk/myid # Trên node3

Các thao tác cơ bản với Zookeeper

Thao tác lệnh cơ bản

ls /
create /app "Dữ liệu app"
create -e /app/tempnode "Dữ liệu tạm"
create -s /app/seqnode "Dữ liệu tuần tự"
get /app
set /app "Dữ liệu mới"
delete /app/seqnode0000000001
rmr /app
stat /app

Các loại nút trong Zookeeper

  1. Nút vĩnh viễn (PERSISTENT).
  2. Nút tạm thời (EPHEMERAL).
  3. Nút tuần tự vĩnh viễn (PERSISTENT_SEQUENTIAL).
  4. Nút tuần tự tạm thời (EPHEMERAL_SEQUENTIAL).

Mechanism Watch trong Zookeeper

get /app watch
# Trong một session khác, sửa đổi dữ liệu tại /app
set /app "Dữ liệu đã thay đổi"
# Session trước đó nhận được sự kiện NodeDataChanged.

API Java Client cho Zookeeper

Java Client gốc

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.7.0</version>
</dependency>

import org.apache.zookeeper.*;
import java.util.concurrent.CountDownLatch;

public class ZKClient {
    private static final String CONNECT_STRING = "localhost:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static ZooKeeper zk;
    private static final CountDownLatch connectedSemaphore = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {
        zk = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, event -> {
            if (Event.KeeperState.SyncConnected == event.getState()) {
                connectedSemaphore.countDown();
            }
        });
        connectedSemaphore.await();

        // Thêm nút
        zk.create("/test", "dữ liệu thử nghiệm".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        
        // Lấy dữ liệu nút
        byte[] data = zk.getData("/test", false, null);
        System.out.println("Dữ liệu nút: " + new String(data));
        
        // Sửa đổi dữ liệu nút
        zk.setData("/test", "dữ liệu mới".getBytes(), -1);
        
        // Xóa nút
        zk.delete("/test", -1);
        
        zk.close();
    }
}

Curator Client

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>5.2.0</version>
</dependency>

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorExample {
    private static final String CONNECT_STRING = "localhost:2181";
    private static final int SESSION_TIMEOUT = 5000;

    public static void main(String[] args) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(CONNECT_STRING)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();

        // Tạo nút
        client.create().forPath("/curator_node", "dữ liệu khởi tạo".getBytes());

        // Lấy dữ liệu nút
        byte[] data = client.getData().forPath("/curator_node");
        System.out.println("Dữ liệu nút: " + new String(data));

        // Sửa đổi dữ liệu nút
        client.setData().forPath("/curator_node", "dữ liệu thay đổi".getBytes());

        // Xóa nút
        client.delete().forPath("/curator_node");

        client.close();
    }
}

Tích hợp SpringBoot với Zookeeper

Xây dựng dự án SpringBoot

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>5.2.0</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
</dependencies>

Lớp cấu hình

@Configuration
public class ZKConfig {
    @Value("${zookeeper.connect-string}")
    private String connectString;

    @Bean(initMethod = "start", destroyMethod = "close")
    public CuratorFramework curatorFramework() {
        return CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .namespace("demo-springboot")
                .build();
    }
}

Lớp dịch vụ

@Service
public class ZKService {
    @Autowired
    private CuratorFramework client;

    public String createNode(String path, String data, CreateMode mode) throws Exception {
        return client.create()
                .creatingParentsIfNeeded()
                .withMode(mode)
                .forPath(path, data.getBytes());
    }

    public String getNodeData(String path) throws Exception {
        byte[] data = client.getData().forPath(path);
        return new String(data);
    }

    public Stat updateNodeData(String path, String data) throws Exception {
        return client.setData().forPath(path, data.getBytes());
    }

    public void deleteNode(String path) throws Exception {
        client.delete().guaranteed().forPath(path);
    }
}

Lớp Controller REST

@RestController
@RequestMapping("/zk")
public class ZKController {
    @Autowired
    private ZKService zkService;

    @PostMapping("/node")
    public String createNode(@RequestParam String path, @RequestParam String data,
                             @RequestParam(defaultValue = "PERSISTENT") String mode) throws Exception {
        return zkService.createNode(path, data, CreateMode.valueOf(mode));
    }

    @GetMapping("/node")
    public String getNodeData(@RequestParam String path) throws Exception {
        return zkService.getNodeData(path);
    }

    @PutMapping("/node")
    public String updateNodeData(@RequestParam String path, @RequestParam String data) throws Exception {
        zkService.updateNodeData(path, data);
        return "Cập nhật thành công";
    }

    @DeleteMapping("/node")
    public String deleteNode(@RequestParam String path) throws Exception {
        zkService.deleteNode(path);
        return "Xóa thành công";
    }
}

Ứng dụng nâng cao của Zookeeper

Thực hiện khóa phân tán

@Service
public class DistributedLockService {
    private static final String LOCK_PATH = "/locks";

    @Autowired
    private CuratorFramework client;

    public boolean acquireLock(String lockName, long waitTime, TimeUnit timeUnit) {
        InterProcessMutex lock = new InterProcessMutex(client, LOCK_PATH + "/" + lockName);
        try {
            return lock.acquire(waitTime, timeUnit);
        } catch (Exception e) {
            return false;
        }
    }

    public void releaseLock(String lockName) {
        InterProcessMutex lock = new InterProcessMutex(client, LOCK_PATH + "/" + lockName);
        try {
            if (lock.isAcquiredInThisProcess()) {
                lock.release();
            }
        } catch (Exception ignored) {}
    }
}

Thực hiện trung tâm cấu hình

@Service
public class ConfigCenterService {
    private static final String CONFIG_PATH = "/configs";
    private final Map configCache = new ConcurrentHashMap<>();

    @Autowired
    private CuratorFramework client;

    @PostConstruct
    public void init() throws Exception {
        loadAllConfigs();
        watchConfigChanges();
    }

    private void loadAllConfigs() throws Exception {
        List<String> children = client.getChildren().forPath(CONFIG_PATH);
        for (String child : children) {
            String path = CONFIG_PATH + "/" + child;
            byte[] data = client.getData().forPath(path);
            configCache.put(child, new String(data));
        }
    }

    private void watchConfigChanges() {
        CuratorCache cache = CuratorCache.build(client, CONFIG_PATH);
        CuratorCacheListener listener = CuratorCacheListener.builder()
                .forCreates(node -> configCache.put(node.getPath().replace(CONFIG_PATH + "/", ""), new String(node.getData())))
                .forChanges((oldNode, node) -> configCache.put(node.getPath().replace(CONFIG_PATH + "/", ""), new String(node.getData())))
                .forDeletes(node -> configCache.remove(node.getPath().replace(CONFIG_PATH + "/", "")))
                .build();
        cache.listenable().addListener(listener);
        cache.start();
    }

    public String getConfig(String key) {
        return configCache.get(key);
    }
}

Đăng ký và phát hiện dịch vụ

@Service
public class ServiceRegistry {
    private static final String REGISTRY_PATH = "/services";

    @Autowired
    private CuratorFramework client;

    @Value("${server.port}")
    private int port;

    @PostConstruct
    public void register() throws Exception {
        if (client.checkExists().forPath(REGISTRY_PATH) == null) {
            client.create().creatingParentsIfNeeded().forPath(REGISTRY_PATH, "Service Registry".getBytes());
        }

        String ip = InetAddress.getLocalHost().getHostAddress();
        String serviceInstance = ip + ":" + port;

        String servicePath = client.create()
                .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                .forPath(REGISTRY_PATH + "/instance-", serviceInstance.getBytes());
    }
}

@Service
public class ServiceDiscovery {
    private static final String REGISTRY_PATH = "/services";
    private final List<String> serviceInstances = new ArrayList<>();

    @Autowired
    private CuratorFramework client;

    @PostConstruct
    public void init() throws Exception {
        discoverServices();
        watchServices();
    }

    private void discoverServices() throws Exception {
        serviceInstances.clear();
        List<String> instances = client.getChildren().forPath(REGISTRY_PATH);
        for (String instance : instances) {
            String instancePath = REGISTRY_PATH + "/" + instance;
            byte[] data = client.getData().forPath(instancePath);
            serviceInstances.add(new String(data));
        }
    }

    private void watchServices() {
        CuratorCache cache = CuratorCache.build(client, REGISTRY_PATH);
        CuratorCacheListener listener = CuratorCacheListener.builder()
                .forCreates(node -> discoverServices())
                .forChanges((oldNode, node) -> discoverServices())
                .forDeletes(node -> discoverServices())
                .build();
        cache.listenable().addListener(listener);
        cache.start();
    }

    public List<String> getAllServiceInstances() {
        return new ArrayList<>(serviceInstances);
    }
}

Lưu ý trong môi trường sản xuất

Tối ưu hóa hiệu suất Zookeeper

  1. Tách biệt thư mục dữ liệu và nhật ký giao dịch.
  2. Tùy chỉnh JVM heap size và GC parameters.
  3. Cấu hình tự động dọn dẹp snapshot cũ.
  4. Hạn chế số lượng kết nối client.

Theo dõi và vận hành

  1. Sử dụng các lệnh bốn chữ để kiểm tra trạng thái server.
  2. Kích hoạt JMX monitoring.
  3. Quản lý log level và policy cuộn log.

Giải quyết các vấn đề thường gặp

  1. Vấn đề kết nối: Kiểm tra firewall, trạng thái service, phiên bản tương thích.
  2. Lỗi nút đã tồn tại: Sử dụng API có version number, kiểm tra tồn tại trước khi thao tác.
  3. Hết hạn phiên: Tăng timeout, triển khai logic reconnect sau khi hết hạn.

Thẻ: zookeeper SpringBoot Curator Java Microservices

Đăng vào ngày 15 tháng 7 lúc 15:47

Thẻ Phổ Biến

©2026 Thành phố Cuồng loạn