Java Zookeeper分布式分片算法超详细讲解流程

@Data
public class ZookeeperConfig {
    
    private String zkAddress;
    
    private String nodePath;
    
    private String serviceName;
    
    private Integer shardinGCount;
    public ZookeeperConfig(String zkAddress, String nodePath, String serviceName, Integer shardingCount) {
        this.zkAddress = zkAddress;
        this.nodePath = nodePath;
        this.serviceName = "/" + serviceName;
        this.shardingCount = shardingCount;
    }
    
    private int baseSleepTimeMilliseconds = 1000;
    
    private int maxSleepTimeMilliseconds = 3000;
    
    private int maxRetries = 3;
    
    private int sessionTimeoutMilliseconds;
    
    private int connectionTimeoutMilliseconds;
}

@AllArgsConstructor
@EqualsAndHashCode()
public class InstanceInfo {
    private String ip;
    public String getInstance() {
        return ip;
    }
}

@Slf4j
public class ZookeeperShardingService {
    public final Map<String, List<Integer>> caches = new HashMap<>(16);
    private final CuratorFramework client;
    private final ZookeeperConfig zkConfig;
    private final ShardingStrategy shardingStrategy;
    private final InstanceInfo instanceInfo;
    private static final CountDownLatch COUNT_DOWN_LATCH = new CountDownLatch(1);
    public ZookeeperShardingService(ZookeeperConfig zkConfig, ShardingStrategy shardingStrategy) {
        this.zkConfig = zkConfig;
        log.info("开始初始化zk, ip列表是: {}.", zkConfig.getZkAddress());
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                .connectString(zkConfig.getZkAddress())
                .retryPolicy(new ExponentialBackoffRetry(zkConfig.getBaseSleepTimeMilliseconds(), zkConfig.getMaxRetries(), zkConfig.getMaxSleepTimeMilliseconds()));
        if (0 != zkConfig.getSessionTimeoutMilliseconds()) {
            builder.sessionTimeoutMs(zkConfig.getSessionTimeoutMilliseconds());
        }
        if (0 != zkConfig.getConnectionTimeoutMilliseconds()) {
            builder.connectionTimeoutMs(zkConfig.getConnectionTimeoutMilliseconds());
        }
        this.shardingStrategy = shardingStrategy;
        HostInfo host = new HostInfo();
        this.instanceInfo = new InstanceInfo(host.getAddress());
        client = builder.build();
        client.getConnectionStateListenable().addListener(new ConnectionListener());
        client.start();
        try {
            COUNT_DOWN_LATCH.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 注册服务节点监听
        reGISterPathChildListener(zkConfig.getNodePath() + zkConfig.getServiceName(), new ChildrenPathListener());
        try {
            if (!client.blockUntilConnected(zkConfig.getMaxSleepTimeMilliseconds() * zkConfig.getMaxRetries(), TimeUnit.MILLISECONDS)) {
                client.close();
                throw new KeeperException.OperationTimeoutException();
            }
        } catch (final Exception ex) {
            ex.printStackTrace();
            throw new RuntimeException(ex);
        }
    }
    
    private void registerPathChildListener(String nodePath, PathChildrenCacheListener listener) {
        try {
            // 1. 创建一个PathChildrenCache
            PathChildrenCache pathChildrenCache = new PathChildrenCache(client, nodePath, true);
            // 2. 添加目录监听器
            pathChildrenCache.getListenable().addListener(listener);
            // 3. 启动监听器
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            log.error("注册子目录监听器出现异常,nodePath:{}",nodePath,e);
            throw new RuntimeException(e);
        }
    }
    
    private void zkOp() throws Exception {
        // 是否存在ruubypay-sharding主节点
        if (null == client.checkExists().forPath(zkConfig.getNodePath())) {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(zkConfig.getNodePath(), Hashing.sha1().hashString("sharding", Charsets.UTF_8).toString().getBytes());
        }
        // 是否存服务主节点
        if (null == client.checkExists().forPath(zkConfig.getNodePath() + zkConfig.getServiceName())) {
            // 创建服务主节点
            client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(zkConfig.getNodePath() + zkConfig.getServiceName());
        }
        // 检查是否存在临时节点
        if (null == client.checkExists().forPath(zkConfig.getNodePath() + zkConfig.getServiceName() + "/" + instanceInfo.getInstance())) {
            System.out.println(zkConfig.getNodePath() + zkConfig.getServiceName() +  "/" + instanceInfo.getInstance());
            // 创建临时节点
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(zkConfig.getNodePath() + zkConfig.getServiceName() +
                    "/" + instanceInfo.getInstance(), zkConfig.getShardingCount().toString().getBytes(StandardCharsets.UTF_8));
        }
        shardingFromZk();
    }
    
    private void shardingFromZk() throws Exception {
        // 从 serviceName 节点下获取所有Ip列表
        final GetChildrenBuilder childrenBuilder = client.getChildren();
        final List<String> instanceList = childrenBuilder.watched().forPath(zkConfig.getNodePath() + zkConfig.getServiceName());
        List<InstanceInfo> res = new ArrayList<>();
        instanceList.forEach(s -> {
            res.add(new InstanceInfo(s));
        });
        Map<InstanceInfo, List<Integer>> shardingResult = shardingStrategy.sharding(res, zkConfig.getShardingCount());
        // 先清一遍缓存
        caches.clear();
        shardingResult.forEach((k, v) -> {
            caches.put(k.getInstance().split("-")[0], v);
        });
    }
    
    private class ConnectionListener implements ConnectionStateListener {
        @Override
        public void stateChanged(CuratorFramework client, ConnectionState newState) {
            if (newState == ConnectionState.CONNECTED || newState == ConnectionState.LOST || newState == ConnectionState.RECONNECTED) {
                try {
                    zkOp();
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new RuntimeException(e);
                } finally {
                    COUNT_DOWN_LATCH.countDown();
                }
            }
        }
    }
    
    private class ChildrenPathListener implements PathChildrenCacheListener {
        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) {
            PathChildrenCacheEvent.Type type = event.getType();
            if (PathChildrenCacheEvent.Type.CHILD_ADDED == type || PathChildrenCacheEvent.Type.CHILD_REMOVED == type) {
                try {
                    shardingFromZk();
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new RuntimeException(e);
                }
            }
        }
    }
}

public interface ShardingStrategy {
    Map<InstanceInfo, List<Integer>> sharding(final List<InstanceInfo> list, Integer shardingCount);
}
public class AverageAllocationShardingStrategy implements ShardingStrategy {
    @Override
    public Map<InstanceInfo, List<Integer>> sharding(List<InstanceInfo> list, Integer shardingCount) {
        if (list.isEmpty()) {
            return null;
        }
        Map<InstanceInfo, List<Integer>> result = shardingAliquot(list, shardingCount);
        addAliquant(list, shardingCount, result);
        return result;
    }
    private Map<InstanceInfo, List<Integer>> shardingAliquot(final List<InstanceInfo> instanceInfos, final int shardingTotalCount) {
        Map<InstanceInfo, List<Integer>> result = new LinkedHashMap<>(shardingTotalCount, 1);
        int itemCountPerSharding = shardingTotalCount / instanceInfos.size();
        int count = 0;
        for (InstanceInfo each : instanceInfos) {
            List<Integer> shardingitems = new ArrayList<>(itemCountPerSharding + 1);
            for (int i = count * itemCountPerSharding; i < (count + 1) * itemCountPerSharding; i++) {
                shardingItems.add(i);
            }
            result.put(each, shardingItems);
            count++;
        }
        return result;
    }
    private void addAliquant(final List<InstanceInfo> instanceInfos, final int shardingTotalCount, final Map<InstanceInfo, List<Integer>> shardingResults) {
        int aliquant = shardingTotalCount % instanceInfos.size();
        int count = 0;
        for (Map.Entry<InstanceInfo, List<Integer>> entry : shardingResults.entrySet()) {
            if (count < aliquant) {
                entry.getValue().add(shardingTotalCount / instanceInfos.size() * instanceInfos.size() + count);
            }
            count++;
        }
    }
}