hbase 原子操作cas

Posted 2021-02-15 asker009

在高并发的情况下，对数据row1  column=cf1:qual1, timestamp=1, value=val1的插入或者更新可能会导致非预期的情况，

例如：原本客户端A需要在value=val1的条件下将value更新成val_A，高并发下客户端B可能抢先将数据value=val1更新成value=val_B，这个时候客户端A如果还继续更新将不符合预期。

HBase中的CAS(compare-and-set)API用来解决这个数据一致性问题，简单的说CAS操作可以让你在put数据之前先经过某些条件的验证，只有满足条件的put才会入库。

hbase1.0版本的API：
checkAndPut(byte.md row, byte.md family, byte.md qualifier, byte.md value, Put put)
其中，put方法的最后一个参数是你需要录入的数据的put对象；
value是与服务端check的预期值，只有服务器端对应rowkey的数据与你预期的值相同时，你的put操作才能被提交的服务端。
但是这个API在2.0版本已经被弃用

新的API是checkAndMutate，一个流式API。checkAndMutate在后续的版本中还在持续改进中。

1、创建一个测试表demoTable，模拟一些测试数据

    //清除并插入测试数据
    private static void createDemoTable() throws IOException {
        String tableNameString = "demoTable";
        if (helper.existsTable(tableNameString))
            helper.dropTable(tableNameString);
        helper.createTable(tableNameString, 100, "cf1", "cf2");
        helper.put(tableNameString,
                new String[]{"row1"},
                new String[]{"cf1", "cf2"},
                new String[]{"qual1", "qual2", "qual3"},
                new long[]{1, 2, 3},
                new String[]{"val1", "val2", "val3"});
        System.out.println("Before check and mutate calls...");
        helper.dump("testtable", new String[]{"row1"}, null, null);
    }

结果类似这样：

2、测试cas，checkAndMutate支持流式函数格式

2.1 如果列数据还不存在就插入预期数据

        String tableNameString = "demoTable";
        Table table = helper.getConnection().getTable(TableName.valueOf(tableNameString));

        boolean res = false;
        Put put = null;

        put = new Put(Bytes.toBytes("row1"));
        put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("qual4"), 4, Bytes.toBytes("val1"));
//        如果row1 cf1 qual4 不存在值就插入put数据
        res = table.checkAndMutate(Bytes.toBytes("row1"), Bytes.toBytes("cf1"))
                .qualifier(Bytes.toBytes("qual4"))
                .ifNotExists()
                .thenPut(put);
        System.out.println("1 result is (expected true) :" + res);

执行结果：1 result is (expected true) :true

执行第二遍将返回false

2.2 如果列值相等就更新数据

        put = new Put(Bytes.toBytes("row1"));
        put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("qual1"), 4, Bytes.toBytes("val1"));
        //如果row1 cf1 qual1 val1存在就插入put，因为这个value已经存在所以可以插入，结果返回true，时间戳变为4
        res = table.checkAndMutate(Bytes.toBytes("row1"), Bytes.toBytes("cf1"))
                .qualifier(Bytes.toBytes("qual1")).ifEquals(Bytes.toBytes("val1"))
                .thenPut(put);
        System.out.println("2 result is (expected true) :" + res);

执行结果：2 result is (expected true) :true

2.3 如果列值不等于则更新

        put = new Put(Bytes.toBytes("row1"));
        put.addColumn(Bytes.toBytes("cf1"),Bytes.toBytes("qual1"),5,Bytes.toBytes("val2"));
        ////如果row1 cf1 qual1 不等于val2在就插入put
        res = table.checkAndMutate(Bytes.toBytes("row1"), Bytes.toBytes("cf1"))
                .qualifier(Bytes.toBytes("qual1"))
                .ifMatches(CompareOperator.NOT_EQUAL,Bytes.toBytes("val2"))
                .thenPut(put);
        System.out.println("3 result is (expected true) :" + res);

2.4 多个条件组合判断，多个操作执行

        put = new Put(Bytes.toBytes("row1"));
        put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("qual5"),1,Bytes.toBytes("val1"));
        put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("qual6"),1,Bytes.toBytes("val1"));

        Delete delete = new Delete(Bytes.toBytes("row1"));
        delete.addColumns(Bytes.toBytes("cf1"), Bytes.toBytes("qual4"));

        //RowMutations这个版本还没定型
        RowMutations mutations = new RowMutations(Bytes.toBytes("row1"));
        mutations.add(put);
        mutations.add(delete);

        //row1 cf1 qual4 val1存在,row1 cf1 qual5和row1 cf1 qual6无值则插入qual5和qual6的值，并删除qual4的值
        res = table.checkAndMutate(Bytes.toBytes("row1"),Bytes.toBytes("cf1")).qualifier(Bytes.toBytes("qual4"))
                .ifEquals(Bytes.toBytes("val1"))
                .qualifier(Bytes.toBytes("qual5")).ifNotExists()
                .qualifier(Bytes.toBytes("qual6")).ifNotExists()
                .thenMutate(mutations);
        System.out.println("1 result is (expected true) :" + res);

3、hbase的cas操作针对是同一个行键下的数据。