线上有台服务器随时可能会挂掉，所以需要把在这个服务器上hbase的regionserver和hdfs的datanode节点移除。然后重新拿台新服务器部署接管。
 
之前在文章 http://openskill.cn/article/178 中讲到怎么新增一个hdfs的datanode，所以我先讲一下怎么添加一个hbase的regionserver，然后再讲怎么删除！ 

添加hbase regionserver节点

添加步骤如下：
1、在hbase  master上修改regionservers文件

# cd hbase_install_dir/conf

# echo "new_hbase_node_hostname" >> ./regionservers

2、如果你hbase集群使用自身zk集群的话，还需要修改hbase-site.xml文件，反之不用操作！

# cd hbase_install_dir/conf

# vim hbase-site.xml

找到hbase.zookeeper.quorum属性   -->加入新节点

3、同步以上修改的文件到hbase的各个节点上
4、在新节点上启动hbase regionserver

# cd hbase_install_dir/bin/

# ./hbase-daemon.sh start regionserver

5、在hbasemaster启动hbase shell

用status命令确认一下集群情况

hbase新增一个 regionserver节点补充完成了，下面介绍删除hbase和hdfs节点！
 
集群上既部署有Hadoop，又部署有HBase，因为HBase存储是基于Hadoop HDFS的，所以先要移除HBase节点，之后再移除Hadoop节点。添加则反之。

移除hbase regionserver节点

1、在0.90.2之前，我们只能通过在要卸载的节点上执行；我的hbase版本(0.98.7)

# cd hbase_install_dir

# ./bin/hbase-daemon.sh stop regionserver

来实现。这条语句执行后，该RegionServer首先关闭其负载的所有Region而后关闭自己。在关闭时，RegionServer在ZooKeeper中的"Ephemeral Node"会失效。此时，Master检测到RegionServer挂掉并把它作为一个宕机节点，并将该RegionServer上的Region重新分配到其他RegionServer。
 
注意：使用此方法前，一定要关闭HBase Load Balancer。关闭方法：

hbase(main):001:0> balance_switch false

true

0 row(s) in 0.3290 seconds

总结：

这种方法很大的一个缺点是该节点上的Region会离线很长时间。因为假如该RegionServer上有大量Region的话，因为Region的关闭是顺序执行的，第一个关闭的Region得等到和最后一个Region关闭并Assigned后一起上线。这是一个相当漫长的时间。以我这次的实验为例，现在一台RegionServer平均有1000个Region，每个Region Assigned需要4s，也就是说光Assigned就至少需要1个小时。

2、自0.90.2之后，HBase添加了一个新的方法，即"graceful_stop",在你移除的服务器执行：

# cd hbase_install_dir

# ./bin/graceful_stop.sh hostname

该命令会自动关闭Load Balancer，然后Assigned Region，之后会将该节点关闭。除此之外，你还可以查看remove的过程，已经assigned了多少个Region，还剩多少个Region，每个Region 的Assigned耗时。
 
补充graceful stop的一些其他命令参数:

# ./bin/graceful_stop.sh

Usage: graceful_stop.sh [--config &conf-dir>] [--restart] [--reload] [--thrift] [--rest] &hostname>

 thrift      If we should stop/start thrift before/after the hbase stop/start

 rest        If we should stop/start rest before/after the hbase stop/start

 restart     If we should restart after graceful stop

 reload      Move offloaded regions back on to the stopped server

 debug       Move offloaded regions back on to the stopped server

 hostname    Hostname of server we are to stop

最终都需要我们手动打开load balancer：

hbase(main):001:0> balance_switch false

true

0 row(s) in 0.3590 seconds

然后再开启：

hbase(main):001:0> balance_switch true

false

0 row(s) in 0.3290 seconds

对比两种方法，建议使用"graceful_stop"来移除hbase RegionServer节点。
官网说明：http://hbase.apache.org/0.94/book/node.management.html​  http://hbase.apache.org/book.html#decommission​  

移除hdfs datanode节点

1、在core-site.xml文件下新增如下内容

       dfs.hosts.exclude

       /hdfs_install_dir/conf/excludes

2、创建exclude文件，把需要删除节点的主机名写入

# cd hdfs_install_dir/conf

# vim excludes

添加需要删除的节点主机名，比如 hdnode1 保存退出

3、 然后在namenode节点执行如下命令，强制让namenode重新读取配置文件，不需要重启集群。

# cd hdfs_install_dir/bin/

# ./hadoop dfsadmin -refreshNodes

它会在后台进行Block块的移动
 4、 查看状态
等待第三步的操作结束后，需要下架的机器就可以安全的关闭了。

# ./hadoop dfsadmin -report

可以查看到现在集群上连接的节点 

正在执行Decommission，会显示： 

Decommission Status : Decommission in progress  



执行完毕后，会显示： 

Decommission Status : Decommissioned

如下：

Name: 10.0.180.6:50010

Decommission Status : Decommission in progress

Configured Capacity: 917033340928 (10.83 TB)

DFS Used: 7693401063424 (7 TB)

Non DFS Used: 118121652224 (110.00 GB)

DFS Remaining: 4105510625280(3.63 TB)

DFS Used%: 64.56%

DFS Remaining%: 34.45%

Last contact: Mon Nov 29 23:53:52 CST 2015

也可以直接通过Hadoop 浏览器查看：
LIVE的节点可以查看到：http://master_ip:50070/dfsnodelist.jsp?whatNodes=LIVE
查看看到卸载的节点状态是：Decommission in progress
等待节点完成移除后，浏览：http://master_ip:50070/dfsnodelist.jsp?whatNodes=DEAD 结果如下：

完成后，删除的节点显示在dead nodes中。且其上的服务停止。Live Nodes中仅剩had2，had3
以上即为从Hadoop集群中Remove Node的过程，但是，有一点一定要注意：
hdfs-site.xml配置文件中dfs.replication值必须小于或者等于踢除节点后正常datanode的数量，即：

dfs.replication <= 集群所剩节点数

修改备份系数可以参考：http://heylinux.com/archives/2047.html

重载入删除的datanode节点 

1、修改namenode的core-site.xml文件，把我们刚刚加入的内容删除或者注释掉，我这里选择注释掉。

2、 再执行重载namenode的配置文件

# ./bin/hadoop dfsadmin -refreshNodes

3、最后去启动datanode上的datanode

# ./bin/hadoop-daemon.sh start datanode

starting datanode, logging to /usr/local/hadoop/bin/../logs/hadoop-root-datanode-had1.out

4、查看启动情况

# jps

18653 Jps

19687 DataNode  ---->启动正常

重新载入HBase RegionServer节点
只需要重启regionserver进程即可。
参考：http://www.edureka.co/blog/commissioning-and-decommissioning-nodes-in-a-hadoop-cluster/
           https://pravinchavan.wordpress.com/2013/06/03/removing-node-from-hadoop-cluster/

调优背景

2015年11月上旬，CRMAPP系统所使用的KVM虚拟机的CPU使用率过高异常，达到70%以上，相同业务压力下同性能配置的Vmware虚拟机的负载却非常低，只在10%以内波动，并且发现KVM宿主机的CPU使用率也异常高。

分析与解决

分别从以下几个方面逐一分析排查问题原因：
1、KVM的CPU虚拟模式
首先查看KVM虚拟机CPU信息如下：

与Vmware虚拟机CPU相比较如下：

通过比较Vmware与KVM的虚拟机CPU信息发现KVM虚拟机CPU模式存在如下几个问题：

[]缺少L3 Cache：初步分析是虚拟机CPU虚拟化模式选择不合理所导致。[/][]CPU不是NUMA架构并且CPU Topology不合理：KVM宿主机物理CPU属于NUMA多node的结构，虚拟机的CPU只有一个NUMA node，所有CPU Core都在这一个node中，且虚拟机CPU的Topology是多Socket单Core的形式。[/]

 处理措施>>>> 缺少L3 Cache

针对该问题，检查了Openstack的nova.conf配置文件libvirt部分的cpu_mode的参数配置是host-model，该参数含义是根据物理CPU的特性，选择一个最靠近的标准CPU型号进行虚拟化模拟。 除了host-model外还可以有host-passthrough模式，该模式直接将物理CPU暴露给虚拟机使用，在虚拟机上完全可以看到的就是物理CPU的型号。因Openstack 仍承载业务，选择小范围的修改cpu_mode的参数，通过将Openstack的代码文件 driver.py中cpu_mode的取值修改为host-passthrough并重启宿主机上Nova-computer服务与KVM虚拟机，将自动重新生成虚拟机的 libvirt.xml文件。

>>>> CPU不是NUMA架构并且CPU Topology不合理

经过梳理Openstack虚拟机的创建流程，并查阅Openstack官方文档与代码，发现在JUNO版的Openstack中，KVM的CPU的拓扑可以通过image或者flavor进行元数据传递来定义，如果没有特别的定义此类元数据，则模拟的CPU将是多Socket单Core单NUMA节点的CPU，这样的CPU与物理CPU完全不同。通过nova命令对flavor增加了hw:numa_cpu、hw:numa_nodes、hw:cpu_sockets等属性。

处理结果经过上面两个方面的修改后，新建KVM虚拟机的CPU的信息发生如下改变，CPU的使用率有了明显下降，在10%到20%之间波动。

[]从单个NUMA节点变成4个NUMA节点[/][]具备了L3 cache[/][]CPU的Topology从32个Socket每个Socket 1个 core,变成了4个Socket每个Socket 8个Core[/]

 2、KVM宿主机与NUMA运行状况

在NUMA的CPU内存架构下，无论是物理主机还是虚拟机，如果NUMA的配置不合理对应用程序的性能都有较大的影响，并且不同类型的应用都有不同的配置需求。Vmware ESX 5.0及之后的版本支持一种叫做vNUMA的特性，它将Host的NUMA特征暴露给了GuestOS，从而使得Guest OS可以根据NUMA特征进行更高性能的调度。SUSE与Redhat作为原生的操作系统在NUMA调度上需要人为的根据应用程序的类型的做特殊配置，对云平台来说，这部分的工作是难以做到的。在优化之前，CRM APP的KVM虚拟机的NUMA调度状态非常不理想，表现为所有NUMA 节点的numa_miss统计数值大于numa_hit，这意味着CPU访问内存的路径不是优化的，存在大量CPU访问remote memory的情况。因为宿主机Ubuntu 12.02自身带有Automatic NUMA balancing，所以物理主机NUMA调度运行状态良好。

处理措施

升级KVM虚拟机的操作系统 到SUSE 12 ，因为新版本的SUSE支持Automatic NUMA balancing，并且操作系统检测到硬件属于NUMA架构时将自动开启。

处理结果

在新建的KVM虚拟机的 dmesg日志中可以看到如下信息：Enablingautomatic NUMA balancing. Configure with numa_balancing= or sysctl经过24小时的运行之后，CRMAPP的KVM虚拟机运行状态良好,CPU用率可以稳定在10%左右。

同时NUMA调度也有了较大程度的改善

总结

[list=1][]通过各环节的优化，目前KVM虚拟机的CPU利用率过高问题不再发生，整体运行达到Vmware虚拟机水平。[/][]不同类型的应用程序对于NUMA适应性不同，需要进行针对性优化。JAVA在NUMA方面也可尝试进行优化。参考Oracle官方文档，JAVA7针对并行扫描垃圾回收站（Parallel Scavenger garbage collector）加入了对NUMA体系结构的支持,实现了NUMA感知的内存分配器，由它为Java应用提供自动的内存分配优化。[/][]目前RedHat7与SUSE 12都加入了NUMA自动负载均衡的特性，可以尽量采用较新版本的操作系统，无论是物理机还是虚拟机，对于NUMA调度的优化不仅与KVM虚拟机有关，其实物理主机也应该关注NUMA调度是否是最优的。[/]

场景

在hadoop中的分布式文件系统hdfs中，当存储节点磁盘使用达到预警值是，我们需要新增一个数据存储节点，来存储数据！我这里hdfs的版本是2.2.0！

新增方法：

[]静态添加[/][]动态添加[/]

静态添加

静态新增的方式，就是相当于我们起初部署hdfs集群规划一样，停止NameNode，新增一个DateNode数据节点，这种方法不适用于线上提供服务的场景，具体操作如下：



1、停止NameNode节点

# cd hdfs_install_dir/sbin/

# ./hadoop-deamon.sh stop namenode



2、修改配置文件slaves文件，并修改/etc/hosts记录把新增的节点对应的ip和hostname追加到各节点

# cd hdfs_install_dir/etc/hadoop/

# echo "new_datanode_hostname" >> ./slaves

# echo "new_datanode_ip  new_datanode_hostname" >> /etc/hosts

然后再利用rsync 同步配置文件和hosts文件，到各节点



3、确保Hadoop/HDFS集群的NameNode可以对新节点进行SSH免密码登录。



4、重新启动NameNode节点



5、如果你希望各数据节点磁盘使用量达到一个相对平衡的状态，就是百分比，你还需要执行hadoop balance命令，后面会具体讲到！

动态添加

动态添加，不需要停止启动NameNode节点，具体步骤如下：



1、修改所有hdfs集群机器的配置文件slaves文件，并修改/etc/hosts记录把新增的节点对应的ip和hostname追加到各节点

# cd hdfs_install_dir/etc/hadoop/

# echo "new_datanode_hostname" >> ./slaves

# echo "new_datanode_ip  new_datanode_hostname" >> /etc/hosts



如果你使用ansible管理的话，hdfs集群的集群做一个叫hdfs的分组，两条命令搞定：

# ansible hdfs -m shell -a 'echo "new_datanode_hostname" >> hdfs_install_dir/etc/hadoop/slaves'

# ansible hdfs -m shell -a 'echo "new_datanode_ip new_datanode_hostname" >> /etc/hosts'

这样所有的节点slaves文件和host文件都更新了！



2、启动新增的datanode节点

# cd hdfs_install_dir/sbin

# ./hadoop-daemon.sh start datanode 



3、查看是否正常加入到集群

web查看方式：http://NameNode_ip:50070/dfsnodelist.jsp?whatNodes=LIVE

命令查看方式：cd hdfs_install_dir/bin/ && ./hadoop dfsadmin -report



4、数据再平衡

添加新节点时，HDFS不会自动重新平衡。然而，HDFS提供了一个手动调用的重新平衡（reblancer）工具。这个工具将整个集群中的数据块分布调整到一个可人工配置的百分比阈值。如果在其他现有的节点上有空间存储问题，再平衡就会根据阀值，然后平衡分布数据。



执行再平衡命令，可选参数-threshold指定了磁盘容量的余量百分比，用来判定一个节点的磁盘利用率是过低还是过高。一个利用不足的数据节点其利用率低于平均利用率−阈值。过度利用的数据节点其利用率高于平均利用率+阈值。该参数设置的越小，整个集群越平衡，但会花费更多的时间进行再平衡操作。默认阈值为10％。平衡执行命令如下：

# cd hdfs_install_dir/sbin/

# ./start-balancer.sh -threshold 5



-threshold参数就是是指定平衡的阈值。 

-threshold的默认是10，即每个datanode节点的实际hdfs存储使用量/集群hdfs存储量 



具体解释例子如下：

datanode hdfs使用量1000G； 

集群总hdfs存储量10T即10000G； 

则t值为1000/10000 = 0.1 = 10%

当执行balance的-t参数小于0.1时，集群进行balance； 

命令为：start-balancer.sh -threshold 10   ;

Expecting a number in the range of [1.0, 100.0]: 5%

sh $HADOOP_HOME/bin/start-balancer.sh –t 10%

这个命令中-t参数后面跟的是HDFS达到平衡状态的磁盘使用率偏差值。如果机器与机器之间磁盘使用率偏差小于10%，那么我们就认为HDFS集群已经达到了平衡的状态。

标注知识点

1、 balance命令可以在namenode或者datanode上启动，也可以随时利用stop-balance.sh脚本停止平衡！ 



2、balance的默认带宽是1M/s。 如果你希望修改平衡数据的带宽大小可以用./hdfs dfsadmin -setBalancerBandwidth 124288000命令指定



3、slave文件是用于重启时使用。集群的start和stop需要读取slave文件。启用datanode时只要在hdfs-site中配置了namenode位置，就可以将信息push给namenode。 这就是为什么slaves文件很重要的原因。

HBase是一个分布式的、面向列的开源数据库，该技术来源于 Fay Chang 所撰写的Google论文“Bigtable：一个结构化数据的分布式存储系统”。就像Bigtable利用了Google文件系统（File System）所提供的分布式数据存储一样，HBase在Hadoop之上提供了类似于Bigtable的能力。HBase是Apache的Hadoop项目的子项目。HBase不同于一般的关系数据库，它是一个适合于非结构化数据存储的数据库。另一个不同的是HBase基于列的而不是基于行的模式。

HBase表一般特点：

[]大：一个表可以有上亿行，上百万列[/][]面向列:面向列(族)的存储和权限控制，列(族)独立检索[/][]稀疏:对于为空(null)的列并不占用存储空间，表可以设计非常稀疏[/]

 
Java 调用 Hbase 非关系型数据库，Hbase 中提供了相关的 Java API 访问接口便于使用，下面是本人综合网络总结的通过 Java 操作 HBase 进行创建、修改、删除表以及查询等。具体封装代码如下：

package yoodb.hbase;

 

import java.io.IOException;

 

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.HColumnDescriptor;

import org.apache.hadoop.hbase.HTableDescriptor;

import org.apache.hadoop.hbase.KeyValue;

import org.apache.hadoop.hbase.client.Delete;

import org.apache.hadoop.hbase.client.Get;

import org.apache.hadoop.hbase.client.HBaseAdmin;

import org.apache.hadoop.hbase.client.HTable;

import org.apache.hadoop.hbase.client.HTablePool;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.util.Bytes;

 

public class HBaseTest {

 

    // 声明静态配置

    static Configuration conf = null;

    static final HTablePool tablePool;

    static {

        conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum", "yoodb");

        tablePool = new HTablePool(conf, 15);

    }

 

    /*

     * 创建表

     * @tableName 表名

     * @family 列族数组

     */

    public static void creatTable(String tableName, String[] family)

            throws Exception {

        HBaseAdmin admin = new HBaseAdmin(conf);

        HTableDescriptor desc = new HTableDescriptor(tableName);

        for (int i = 0; i < family.length; i++) {

            desc.addFamily(new HColumnDescriptor(family[i]));

        }

        if (admin.tableExists(tableName)) {

            System.out.println("table Exists!");

            System.exit(0);

        } else {

            admin.createTable(desc);

            System.out.println("create table Success!");

        }

    }

 

    /*

     * 表添加数据

     * @rowKey rowKey

     * @tableName 表名

     * @column1 第一个列族数组 realname

     * @value1 第一个列的值的数组

     * @column2 第二个列族数组 address

     * @value2 第二个列的值的数组

     */

    public static void addTableData(String rowKey, String tableName,String[] column1, String[] value1, String[] column2, String[] value2)

            throws IOException {

        Put put = new Put(Bytes.toBytes(rowKey));

        HTable table = (HTable) tablePool.getTable(tableName);

        HColumnDescriptor[] columnFamilies = table.getTableDescriptor()

                .getColumnFamilies();

 

        for (int i = 0; i < columnFamilies.length; i++) {

            String familyName = columnFamilies[i].getNameAsString();

            if (familyName.equals("realname")) {

                for (int j = 0; j < column1.length; j++) {

                    put.add(Bytes.toBytes(familyName),Bytes.toBytes(column1[j]), Bytes.toBytes(value1[j]));

                }

            }

            if (familyName.equals("address")) {

                for (int j = 0; j < column2.length; j++) {

                    put.add(Bytes.toBytes(familyName),Bytes.toBytes(column2[j]), Bytes.toBytes(value2[j]));

                }

            }

        }

        table.put(put);

    }

 

    /*

     * 更新表中的某一列

     * @tableName 表名

     * @rowKey rowKey

     * @familyName 列族名

     * @columnName 列名

     * @value 更新后的值

     */

    public static void updateTable(String tableName, String rowKey,

            String familyName, String columnName, String value)

            throws IOException {

        HTable table = (HTable) tablePool.getTable(tableName);

        Put put = new Put(Bytes.toBytes(rowKey));

        put.add(Bytes.toBytes(familyName), Bytes.toBytes(columnName),Bytes.toBytes(value));

        table.put(put);

    System.out.println("update table Success!");

    }

 

    /*

     * 根据rwokey查询

     * @rowKey rowKey

     * @tableName 表名

     */

    public static Result getResult(String tableName, String rowKey)

            throws IOException {

        Get get = new Get(Bytes.toBytes(rowKey));

        HTable table = (HTable) tablePool.getTable(tableName);

        Result result = table.get(get);

        for (KeyValue kv : result.list()) {

            System.out.println("family==>" + Bytes.toString(kv.getFamily()));

            System.out.println("qualifier==>" + Bytes.toString(kv.getQualifier()));

            System.out.println("value==>" + Bytes.toString(kv.getValue()));

            System.out.println("Timestamp==>" + kv.getTimestamp());

        }

        return result;

    }

 

    /*

     * 遍历查询hbase表数组

     * @tableName 表名

     */

    public static void getResultScann(String tableName) throws IOException {

        Scan scan = new Scan();

        ResultScanner rs = null;

        HTable table = (HTable) tablePool.getTable(tableName);

        try {

            rs = table.getScanner(scan);

            for (Result r : rs) {

                for (KeyValue kv : r.list()) {

                    System.out.println("family==>" + Bytes.toString(kv.getFamily()));

                    System.out.println("qualifier==>" + Bytes.toString(kv.getQualifier()));

                    System.out.println("value==>" + Bytes.toString(kv.getValue()));

                    System.out.println("timestamp==>" + kv.getTimestamp());

                }

            }

        } finally {

            rs.close();

        }

    }

 

    /*

     * 查询表中的某单一列

     * @tableName 表名

     * @rowKey rowKey

     */

    public static void getResultByColumn(String tableName, String rowKey,

            String familyName, String columnName) throws IOException {

        HTable table = (HTable) tablePool.getTable(tableName);

        Get get = new Get(Bytes.toBytes(rowKey));

        get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName)); // 获取指定列族以及列中修饰符对应列名

        Result result = table.get(get);

        for (KeyValue kv : result.list()) {

            System.out.println("family==>" + Bytes.toString(kv.getFamily()));

            System.out.println("qualifier==>" + Bytes.toString(kv.getQualifier()));

            System.out.println("value==>" + Bytes.toString(kv.getValue()));

            System.out.println("Timestamp==>" + kv.getTimestamp());

        }

    }

 

    /*

     * 查询某列数据的多个版本

     * @tableName 表名

     * @rowKey rowKey

     * @familyName 列族名

     * @columnName 列名

     */

    public static void getResultByVersion(String tableName, String rowKey,

            String familyName, String columnName) throws IOException {

        HTable table = (HTable) tablePool.getTable(tableName);

        Get get = new Get(Bytes.toBytes(rowKey));

        get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));

        get.setMaxVersions(5);

        Result result = table.get(get);

        for (KeyValue kv : result.list()) {

            System.out.println("family==>" + Bytes.toString(kv.getFamily()));

            System.out.println("qualifier==>" + Bytes.toString(kv.getQualifier()));

            System.out.println("value==>" + Bytes.toString(kv.getValue()));

            System.out.println("Timestamp==>" + kv.getTimestamp());

        }

         

    }

 

    /*

     * 删除指定的列

     * @tableName 表名

     * @rowKey rowKey

     * @familyName 列族名

     * @columnName 列名

     */

    public static void deleteColumn(String tableName, String rowKey,

            String falilyName, String columnName) throws IOException {

        HTable table = (HTable) tablePool.getTable(tableName);

        Delete deleteColumn = new Delete(Bytes.toBytes(rowKey));

        deleteColumn.deleteColumns(Bytes.toBytes(falilyName),Bytes.toBytes(columnName));

        table.delete(deleteColumn);

        System.out.println(falilyName + "==>" + columnName + "is deleted!");

    }

 

    /*

     * 删除指定的列

     * @tableName 表名

     * @rowKey rowKey

     */

    public static void deleteAllColumn(String tableName, String rowKey) throws IOException {

        HTable table = (HTable) tablePool.getTable(tableName);

        Delete deleteAll = new Delete(Bytes.toBytes(rowKey));

        table.delete(deleteAll);

        System.out.println("all columns are deleted!");

    }

 

    /*

     * 删除表

     * 

     * @tableName 表名

     */

    public static void deleteTable(String tableName) throws IOException {

        HBaseAdmin admin = new HBaseAdmin(conf);

        admin.disableTable(tableName);

        admin.deleteTable(tableName);

        System.out.println(tableName + " is deleted!");

    }

}

Java Hbase main函数测试类，具体代码如下：

package com.yoodb;

 

public class Test {

    public static void main(String[] args) throws Exception {

        // 创建表

        String tableName = "yoodbblog";

        String[] family = { "realname","address" };

        HBaseTest.creatTable(tableName,family);

        // 为表添加数据

        String[] column1 = { "title", "author", "content" }; 

        String[] value1 = {"素文宅","yoodb","www.yoodb.com" }; 

        String[] column2 = { "name", "nickname" };

        String[] value2 = { "真实名称", "昵称" }; 

        HBaseTest.addTableData("rowkey1","yoodbblog",column1, value1, column2, value2);

        // 删除一列

        HBaseTest.deleteColumn("yoodbblog", "rowkey1", "realname", "name");

        // 删除所有列

        HBaseTest.deleteAllColumn("yoodbblog", "rowkey1");

        // 删除表

        HBaseTest.deleteTable("yoodbblog");

        // 查询

        HBaseTest.getResult("yoodbblog", "rowkey1");

        // 查询某一列的值

        HBaseTest.getResultByColumn("yoodbblog", "rowkey1", "realname", "nickname");

        // 修改某一列的值

        HBaseTest.updateTable("yoodbblog", "rowkey1", "realname", "nickname","假昵称");

        // 遍历表数据查询

        HBaseTest.getResultScann("yoodbblog");

        // 查询某列的多版本

        HBaseTest.getResultByVersion("yoodbblog", "rowkey1", "realname", "name");

    }

}

分享原文地址
 
 
 

Yii 1.x、thinkPHP、CodeIgniter在PHP 5.3之前，MVC的实现算是比较的足规中矩，大体解决办法是从REQUEST_URI中提取uri，根据uri规则分解出contraller、action、params或者还有app（Yaf）。逻辑也比较清晰，用notepad++就可以了解MVC的结构和主体思想。
 
现在的MVC框架随着PHP版本的升级，支持的特性越来越多，尤其匿名函数这概念的引入，使得服务容器Container在众多一流MVC新版本中极为受宠。laravel的Illuminate/Container使得laravel 5可以让开发者非常灵活组合地使用composer组件。其他如Symfony 2的Component/DependencyInjection/ContainerBuilder和Yii 2的di/Container各家都做了自己的实现。
 
服务容器也叫IoC 容器，或者另外一些说法叫控制反转、依赖注入。暂时叫依赖注入，这名字更贴切的表达服务容器的使命：为解决依赖而生。
 
先来简释Yii2的Container，事实上Yii2的容器实现非常复杂。以Controller::behaviors()这方法说起，先看下：

public function behaviors() {

    return [

        'access' => [

            'class' => AccessControl::className(),

            'rules' => [

                [

                    'actions' => ['login', 'error'],

                    'allow' => true,

                ],

                [

                    'actions' => ['logout', 'index'],

                    'allow' => true,

                    'roles' => ['@'],

                ],

            ],

        ],

        'verbs' => [

            'class' => VerbFilter::className(),

            'actions' => [

                'logout' => ['post'],

            ],

        ],

    ];

}

这就得一路追起来

[]yii\web\Application[/][]yii\base\Application::run()[/][]yii\base\Component::trigger()[/][]yii\base\Component::ensureBehaviors()[/][]yii\base\Component::attachBehaviorInternal()(到现在终于才看到controller::behavior()的影子)[/][]yii\BaseYii::createObject()(终于是服务容器登场了)[/][]yii\di\Container::get()[/]

再细看Container::get($class, $params = , $config = )如何实现服务容器的。三个参数里config其实对于BaseYii::createObject()是暂时没用的，先关注前面两个参数。

# 1、对于单例（对象）来说，无须检查依赖和参数传递

if (isset($this->_singletons[$class])) {

    return $this->_singletons[$class];



# 2、好吧，首次创建这个对象

} elseif (!isset($this->_definitions[$class])) {

    return $this->build($class, $params, $config);

}

Container::build()这个对象得需要知道这个对象的依赖关系：Container::getDependencies($class)

$dependencies = ;

# 先建立对象反射

$reflection = new ReflectionClass($class);



# 获取这个对象的初始化__construct(Foo $foo, $level = 0)依赖条件

$constructor = $reflection->getConstructor();

if ($constructor !== null) {

    foreach ($constructor->getParameters() as $param) {

        # 有默认值的好说，如level = 0

        if ($param->isDefaultValueAvailable()) {

            $dependencies = $param->getDefaultValue();



        # 否则，我们得知道这个依赖的类是什么，如：类Far，并且创建这个对象Instance::of('Foo')

        } else {

            $c = $param->getClass();

            $dependencies = Instance::of($c === null ? null : $c->getName());

        }

    }

}



# 记录$_reflections、$_dependencies并返回

$this->_reflections[$class] = $reflection;

$this->_dependencies[$class] = $dependencies;

此时已经得到一个AccessControl的反射类和相关依赖，好吧，回头一看Controller::behaviors()应该可以由服务容器提供一个AccessControl了吧。细心的同学在上面获取依赖类的过程，有一个细节：创建类Far是用了Instance::of('Far')，只是一个$id = 'Far'的Instance。并不是真正的Far类，而且能想到这个Far实例会不会也像AccessControl一样也有依赖呢？啊，这样下去还有完没完了！
 
好吧，那所有的都走一次Container::get($class, $params)，满足了吧。所以也就有了Container::build($class, $params)方法里先解决一层AccessControl依赖，接着再来一次解决依赖Container::resolveDependencies()

list ($reflection, $dependencies) = $this->getDependencies($class);



...



$dependencies = $this->resolveDependencies($dependencies, $reflection);

具体看Container::resolveDependencies()的实现

/**

 * Resolves dependencies by replacing them with the actual object instances.

 * 以最终实例化的对象来填充类的依赖

 * @param array $dependencies the dependencies

 * @param ReflectionClass $reflection the class reflection associated with the dependencies

 * @return array the resolved dependencies

 * @throws InvalidConfigException if a dependency cannot be resolved or if a dependency cannot be fulfilled.

 */

protected function resolveDependencies($dependencies, $reflection = null) {

    foreach ($dependencies as $index => $dependency) {

        if ($dependency instanceof Instance) {

            if ($dependency->id !== null) {

                # 取回$id = 'Far'值，重新Container::get('Far')得到真正的实例，新的一轮Container::get()又开始了，直到所有依赖的依赖的依赖...都被解决

                $dependencies[$index] = $this->get($dependency->id);

            } elseif ($reflection !== null) {

                $name = $reflection->getConstructor()->getParameters()[$index]->getName();

                $class = $reflection->getName();

                throw new InvalidConfigException("Missing required parameter \"$name\" when instantiating \"$class\".");

            }

        }

    }

    return $dependencies;

}

Container::build($class, $params)已经被打断两次了，好不容易把依赖都解决完了，终于可以创建最开始的实例AccessControl了。

# 打断：解决依赖

list ($reflection, $dependencies) = $this->getDependencies($class);



...



# 打断：解决依赖的依赖...

$dependencies = $this->resolveDependencies($dependencies, $reflection);



# 利用反射类实例对象，顺手把$config数据元素赋到对象的属性

if (!empty($config) && !empty($dependencies) && is_a($class, 'yii\base\Object', true)) {

    // set $config as the last parameter (existing one will be overwritten)

    $dependencies[count($dependencies) - 1] = $config;

    return $reflection->newInstanceArgs($dependencies);

} else {

    $object = $reflection->newInstanceArgs($dependencies);

    foreach ($config as $name => $value) {

        $object->$name = $value;

    }

    return $object;

}

纵观上面服务容器可以支持实例singleton、类名string，但还不能支持闭包clourse，那Yii 2怎么好意思呢？刚才追到了Container::build($class, $params, $config)，现在稍微回溯一级到Container::get($class, $params, $config)

在我们看代码之前，试想下，如果自己要实现一个服务容器，分别支持这三种类型该如何设计？实例不须做工作，先记录保存；字符串类名需要特别细心，通过上述层层依赖的反射最终可以解决；剩下的闭包可以通过call_user_func处理匿名函数就可以得到最终的实例。现在验证下Yii 2是不是也这样的策略。

在我们马上要彻底分析Container::get之前，还有一些工作需要我们理清楚的。get相当于依赖解析和实例化对象，而之前还有一个工作就是注入。到现在我们也还没有对注入进行分析，而在PHP中设计一个服务容器支持上面提到的三种类型，注入是重要的入口，只有注入优雅了，依赖解析才会优雅。

Yii 2的注入是在Container::set中实现，Container::set($class, $params)都支持哪些类注册方式？以达到我们可以随意的Container::get呢？我简单分类说明，代码可以忽略，以下注释就是对Container::set中唯一一个方法normalizeDefinition($class, $definition)对定义进行规范化处理的实例版本。

#A：初级版本的注册，直接一个命名空间的类名，毫无挑战性，甚至都没有注册的必要

$container->set('yii\db\Connection');



// register an interface

// When a class depends on the interface, the corresponding class

// will be instantiated as the dependent object

#B：对于用接口作为类型约束，那实例化时可不能对接口进行实例化，需要根据实际的继承类来实例化。

#   如：__construct(yii\mail\MailInterface $mailer)，而最终实例化的是yii\swiftmailer\Mailer

$container->set('yii\mail\MailInterface', 'yii\swiftmailer\Mailer');



// register an alias name. You can use $container->get('db')

// to create an instance of Connection

#C：如果你觉得yii\db\Connection这货名字太长，可以别名为db，这样在model层就可以随意的$container->get('db')

$container->set('db', 'yii\db\Connection');



// register a class with configuration. The configuration

// will be applied when the class is instantiated by get()

#D：如果对于A版本，没法满足你了，需要在注入时就初始化该的一些属性，使用$params数组即可

$container->set('yii\db\Connection', [

'dsn' => 'mysql:host=127.0.0.1;dbname=demo',

'username' => 'root',

'password' => '',

'charset' => 'utf8',

]);



// register an alias name with class configuration

// In this case, a "class" element is required to specify the class

#E：如果你想要C+D这种结合体，当然也可以，请在$params的key为class标明你原始类名

$container->set('db', [

'class' => 'yii\db\Connection',

'dsn' => 'mysql:host=127.0.0.1;dbname=demo',

'username' => 'root',

'password' => '',

'charset' => 'utf8',

]);



// register a PHP callable

// The callable will be executed when $container->get('db') is called

#F：总会有一些任性的同学，我想要自定义类，那总得支持吧，请使用闭包函数吧

#   虽然它不会像js那些做到真正的回调，但变量的作用域的思想是一致的

$container->set('db', function ($container, $params, $config) {

return new \yii\db\Connection($config);

});

既然注入是这么简单的规则，学习成本小，接下来的最后依赖解析Container::get($class, $params, $config)的完整代码。

/**

 * Returns an instance of the requested class.

 * 所谓好的IoC就是能static::get('啥都有')都能return正确的对象

 */

public function get($class, $params = , $config = ) {

    # 1、对于单例（对象）来说，无须检查依赖和参数传递

    if (isset($this->_singletons[$class])) {

        return $this->_singletons[$class];



    # 2、好吧，首次创建这个对象

    } elseif (!isset($this->_definitions[$class])) {

        return $this->build($class, $params, $config);

    }



    # 3、为什么已定义的对象不直接给返回？

    #    此定义非已经创建过对象这种定义，而是Container::set($class, $definition, $params)注册了一个$class而已，跟laravel的bind相像。

    $definition = $this->_definitions[$class];



    # 4、$definition为闭包函数：

    if (is_callable($definition, true)) {

        $params = $this->resolveDependencies($this->mergeParams($class, $params));

        $object = call_user_func($definition, $this, $params, $config);



    # 5、$definition为数组：

    } elseif (is_array($definition)) {

        # 数组中必须要给出一个key为class的类名

        $concrete = $definition['class'];

        unset($definition['class']);



        $config = array_merge($definition, $config);

        $params = $this->mergeParams($class, $params);



        # 对于没有别名的，可以直接创建该对象

        if ($concrete === $class) {

            $object = $this->build($class, $params, $config);



        # 别名为什么要递归？而不是$this->build(concrete, $params, $config)

        # 

        } else {

            $object = $this->get($concrete, $params, $config);

        }



    # 6、$definition为对象：

    } elseif (is_object($definition)) {

        return $this->_singletons[$class] = $definition;

    } else {

        throw new InvalidConfigException("Unexpected object definition type: " . gettype($definition));

    }



    # 更新单例记录的对象，取最后更新值。假设Foo::__construct($level = 0) {}

    # 当同一进程中有$this->_singletons['Foo'] = new Foo(1)；

    # 现在Container::get('Foo')

    # 此时$this->_singletons[$class] = new Foo(0)；

    if (array_key_exists($class, $this->_singletons)) {

        // singleton

        $this->_singletons[$class] = $object;

    }



    return $object;

}

原文作者：花满树
分享原文链接：http://blog.huamanshu.com/?date=2015-06-26

一、hbase的HRegionServer节点启动失败

2015-10-23 17:24:33,147 WARN  [regionserver60020] zookeeper.RecoverableZooKeeper: Node /hbase/rs/SlaveServer,60020,1413095376898 already deleted, retry=false

2015-10-23 17:24:33,147 WARN  [regionserver60020] regionserver.HRegionServer: Failed deleting my ephemeral node

org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /hbase/rs/SlaveServer,60020,1413095376898

	at org.apache.zookeeper.KeeperException.create(KeeperException.java:111)

	at org.apache.zookeeper.KeeperException.create(KeeperException.java:51)

	at org.apache.zookeeper.ZooKeeper.delete(ZooKeeper.java:873)

	at org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper.delete(RecoverableZooKeeper.java:156)

	at org.apache.hadoop.hbase.zookeeper.ZKUtil.deleteNode(ZKUtil.java:1273)

	at org.apache.hadoop.hbase.zookeeper.ZKUtil.deleteNode(ZKUtil.java:1262)

	at org.apache.hadoop.hbase.regionserver.HRegionServer.deleteMyEphemeralNode(HRegionServer.java:1298)

	at org.apache.hadoop.hbase.regionserver.HRegionServer.run(HRegionServer.java:1012)

	at java.lang.Thread.run(Thread.java:662)

2015-10-23 17:24:33,158 INFO  [regionserver60020] zookeeper.ZooKeeper: Session: 0x249020a2cfd0014 closed

2015-10-23 17:24:33,158 INFO  [regionserver60020-EventThread] zookeeper.ClientCnxn: EventThread shut down

2015-10-23 17:24:33,158 INFO  [regionserver60020] regionserver.HRegionServer: stopping server null; zookeeper connection closed.

2015-10-23 17:24:33,158 INFO  [regionserver60020] regionserver.HRegionServer: regionserver60020 exiting

2015-10-23 17:24:33,158 ERROR [main] regionserver.HRegionServerCommandLine: Region server exiting

java.lang.RuntimeException: HRegionServer Aborted

	at org.apache.hadoop.hbase.regionserver.HRegionServerCommandLine.start(HRegionServerCommandLine.java:66)

	at org.apache.hadoop.hbase.regionserver.HRegionServerCommandLine.run(HRegionServerCommandLine.java:85)

	at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70)

	at org.apache.hadoop.hbase.util.ServerCommandLine.doMain(ServerCommandLine.java:126)

	at org.apache.hadoop.hbase.regionserver.HRegionServer.main(HRegionServer.java:2422)

2015-10-23 17:24:33,160 INFO  [Thread-9] regionserver.ShutdownHook: Shutdown hook starting; hbase.shutdown.hook=true; fsShutdownHook=org.apache.hadoop.fs.FileSystem$Cache$ClientFinalizer@8d5aad

2015-10-23 17:24:33,160 INFO  [Thread-9] regionserver.ShutdownHook: Starting fs shutdown hook thread.

2015-10-23 17:24:33,160 INFO  [Thread-9] regionserver.ShutdownHook: Shutdown hook finished.

一般这种情况，是因为集群中节点时间相差太多，时间没有同步导致的，解决方案：

# yum -y install ntpdate  && chkconfig ntpdate off

# crontab -e     #add sync time cron scripts

[i]/2 [/i] [i] [/i] * ntpdate asia.pool.ntp.org

如果遇到是其他原因的同学，下面回答分享一下！

二、主机名配置问题

failed on connection exception: java.net.ConnectException: Connection refused; For more details see:  http://wiki.apache.org/hadoop/ConnectionRefused

根据查看提示链接http://wiki.apache.org/hadoop/ConnectionRefused排查错误，将/etc/hosts中的127.0.0.1 hbase1删除（从节点对应也删除）后程序运行正常。接着尝试运行HBase，没有出现问题！创建表也正常了！
一开始知道得删除hosts文件中127.0.1.1，但是没想到127.0.0.1 主机名也得删除。
 
还有一种情况也会导致集群启动问题，那就是主机名不规范，作为hadoop集群中的主机名，是不支持_和-的，比如hbase_host1这是不支持的！

一、启动虚拟机​Connection reset by peer

# virsh start vmhost1

error: Failed to start domain vmhost1

error: Unable to read from monitor: Connection reset by peer

在虚拟机运行过程中关闭宿主服务器就有可能导致这种情况出现，由于宿主服务器中的kvm虚拟机控制器与安装在kvm中的虚拟机会话被异常重置，所以我们可以如下解决：

# virsh managedsave-remove vmhost1

# virsh start vmhost1

如果启动查看/var/log/libvirt/qemu/vmhost1.log下log还报如下错误：

Cannot set up guest memory 'pc.ram': Cannot allocate memory

这个问题可能是分配给vmhost1分配的内存过大（甚至超过的物理主机的内存大小），或者可能是宿主机没有足够的内存分配给此虚拟机，导致无法启动！

二、重Define虚拟机时无/usr/bin/kvm

error: Failed to define domain from hostname.xml

error: Cannot find QEMU binary /usr/bin/kvm: No such file or directory

解决方法：

# ln -s /usr/libexec/qemu-kvm /usr/bin/kvm

三、error: internal error process exited while connecting to monitor

# virsh start vmhost1  

error: Failed to start domain vmhost1

error: internal error process exited while connecting to monitor: kvm: -drive file=/dev/sp1368155439693/v1368544020461,if=none,id=drive-virtio-disk0,format=qcow2: could not open disk image /dev/sp1368155439693/v1368544020461: Invalid argument

分析：镜像格式错误，用qemu-img info 检查镜像和xml配置文件中指定的type是否一致！

四、Unable to load library 'virt': libvirt.so

Unable to load library 'virt': libvirt.so: cannot open shared object file: No such file or directory



Linux下解决：

      ln -s /usr/lib/libvirt.so.0   /usr/lib/libvirt.so



windows下解决：

      将libvirt-0.dll改名为virt.dll

五、error: Refusing to undefine while domain managed save image exists

# virsh undefine vmhost1

error: Refusing to undefine while domain managed save image exists

http://www.redhat.com/archives/libvir-list/2011-July/msg01219.html

解决方法：virsh undefine $domain  --managed-save

六、启动libvirtd进程出错

# /usr/local/sbin/libvirtd -d -l --config /usr/local/etc/libvirt/libvirtd.conf （编译安装的启动方式）

error：/usr/local/sbin/libvirtd: initialization failed

try to install libpcap-devel RPM and rebuild libvirt  http://comments.gmane.org/gmane.comp.emulators.libvirt/58218



apt-get install libpcap-dev

上面的方法好像都没有效果，但是尝试了http://wiki.libvirt.org/page/The_daemon_cannot_be_started说的，把配置文件里的

listen_tls = 0注释取消（更奇怪的问题，在我的客户端链接不对）

七、启动虚拟机报错

# virsh start vmhost1

error: Failed to start domain vmhost1

error: internal error process exited while connecting to monitor: Could not access KVM kernel module: No such file or directory

failed to initialize KVM: No such file or directory

No accelerator found!

上面的提示信息就是因为QEMU在初始化阶段因为无法找到kvm内核模块。

# modprobe kvm   #载入指定的模块

重启电脑，进入bios界面，设置advance选项里面的virtualization标签为Enabled

通过命令 lsmod | grep kvm    #显示已载入的模块

八、虚拟机迁移

# virsh migrate --live 1 qemu+tcp://192.168.0.121 --p2p --tunnelled --unsafe 

error: operation failed: Failed to connect to remote libvirt URI qemu+tcp://192.168.0.121（在URI后面加上/system，‘system’相当于root用户的访问权限)



#virsh migrate --live 2 qemu+tcp://192.168.0.121/system --p2p --tunnelled

 error: Unsafe migration: Migration may lead to data corruption if disks use cache != none（加上--unsafe参数）



#virsh migrate --live 2 qemu+tcp://192.168.0.121/system --p2p --tunnelled --unsafe 

error: Timed out during operation: cannot acquire state change lock (启动虚拟机有时也会遇此错误）,需要重启libvirtd进程

九、virsh

error: Failed to connect socket to '/var/run/libvirt/libvirt-sock': Connection refused(libvirtd 进程没有启动，libvirtd是一个监听客户端请求的进程）



# virsh -c qemu:///system list

error: Failed to connect socket to '/var/run/libvirt/libvirt-sock': Permission denied

error: failed to connect to the hypervisor

（当前用户没有权限，修改/etc/libvirt/libvirtd.conf,unix_sock_rw_perms = 0777,使所有用户都有权限读写）

ZooKeeper 是分布式环境下非常重要的一个中间件，可以完成动态配置推送、分布式 Leader 选举、分布式锁等功能。在运维 ZooKeeper 服务的以来，积累如下经验：

1. 集群数量

3台起，如果是虚拟机，必须分散在不同的宿主机上，以实现容灾的目的。如果长远来看（如2-3年）需求会持续增长，可以直接部署5台。ZooKeeper集群扩容是比较麻烦的事情，因此宁可前期稍微浪费一点。

1. 客户端配置域名而不是 IP

如果有一天你的 ZooKeeper 集群需要做机房迁移，或者其中几个节点机器挂了，需要更换。让你的用户更新 ZooKeeper 服务器配置不是件轻松的事情，因此一开始就配置好域名，到时候更新 DNS 即可。

1. 开启 autopurge.snapRetainCount

ZooKeeper 默认不会自动清理 tx log，所以总会有一天你会收到磁盘报警（如果你有磁盘监控的话）。开启自动清理机制后，就不用担心了，我的配置如下：

autopurge.snapRetainCount=500

autopurge.purgeInterval=24

1. 扩容

如果你可以接受停止服务半个小时，那基本随意玩了，但在比较严肃的环境下，还是不能停服务的。我的做法是这样的：
0. 有节点 A, B, C 处于服务状态

server.3=192.168.12.1:2888:3888

server.4=192.168.12.2:2888:3888

server.5=192.168.12.3:2888:3888

1. 加入节点 D，配置如下：

server.3=192.168.12.1:2888:3888

server.4=192.168.12.2:2888:3888

server.5=192.168.12.3:2888:3888

server.6=192.168.12.4:2888:3888

server.7=192.168.12.5:2888:3888

用 4 字命令检查，保证该节点同步完毕集群数据，处于 Follower 状态：

# echo srvr | nc 192.168.12.4 2181

Zookeeper version: 3.4.5-1392090, built on 09/30/2012 17:52 GMT

Latency min/avg/max: 0/0/13432

Received: ***

Sent: ***

Connections: ***

Outstanding: 0

Zxid: 0x***

Mode: follower

Node count: ***

需要注意的是，这一步加入的节点的 id，必须大于集群中原有的节点的 id，例如 6 > 3,4,5，我也不知道为什么需要这样。

1. 同上一步一样，加入节点 E
2. 更新 A B C 的配置如 D 和 E，并依此重启

1. 机房迁移

例如要把服务从 X 机房的 A B C 迁移到 Y 机房的 A’ B’ C’。

做法是首先把集群扩容成包含6个节点的集群；然后修改域名指向让用户的连接都转到 A’ B’ C’；最后更新集群配置，把 A B C 从集群摘除。

1. 跨机房容灾

由于 ZooKeeper 天生不喜欢偶数（怕脑裂），因此有条件的就三机房部署，但机房之间的网络条件得是类似局域网的条件，否则性能就堪忧了。



双机房做自动容灾基本不可能，加入手动步骤是可以的，和 DB 一样，短时间不可用，立刻启用另外一个机房，平时保证数据同步。



三机房部署，如果一个机房离的比较远，网络延迟较高怎么办？可以 3 + 3 + 1 部署，1 就放在那个网络延迟较高的地方，确保 leader 在 3 + 3 这两个机房中间，那么平时的性能就能保证了。怎么保证 leader 不到 1 呢？目前能想到的办法就是如果发现就重启它