0. 需求

模拟微博业务，其中包括发布微博、添加关注、取关用户、获取某个用户初始化页面数据、获取某个用户所有微博详情

1. 表结构

1.1 微博内容表

image.png

1.2 用户关系表

image.png

1.3 微博收件箱表

image.png

2. 常量类

  1. package com.jshawn.hbase.guli_weibo.constants;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.hbase.HBaseConfiguration;
  6. /**
  7.  * @author JShawn 2021/6/20 16:47
  8.  */
  9. public class Constants {
  11.     /**
  12.      * HBase的配置信息
  13.      */
  14.     public static final Configuration COFIGURATION = HBaseConfiguration.create();
  16.     /**
  17.      * 命名空间
  18.      */
  19.     public static final String NAMESPACE = "weibo";
  21.     /**
  22.      * 微博内容表
  23.      */
  24.     public static final String CONTENT_TABLE = "weibo:content";
  26.     /**
  27.      * 微博内容表-列族
  28.      */
  29.     public static final String CONTENT_TABLE_CF = "info";
  31.     /**
  32.      * 微博内容表-最大版本
  33.      */
  34.     public static final Integer CONTENT_TABLE_VERSIONS = 1;
  36.     /**
  37.      * 用户关系表
  38.      */
  39.     public static final String RELATION_TABLE = "weibo:relation";
  41.     /**
  42.      * 用户关系表-列族-所有关注
  43.      */
  44.     public static final String RELATION_TABLE_CF_ATTENDS = "attends";
  46.     /**
  47.      * 用户关系表-列族-所有粉丝
  48.      */
  49.     public static final String RELATION_TABLE_CF_FANS = "fans";
  51.     /**
  52.      * 用户关系表-最大版本
  53.      */
  54.     public static final Integer RELATION_TABLE_VERSIONS = 1;
  56.     /**
  57.      * 收件箱表（即用户初始化页面数据表）
  58.      */
  59.     public static final String INBOX_TABLE = "weibo:inbox";
  61.     /**
  62.      * 收件箱表-列族
  63.      */
  64.     public static final String INBOX_TABLE_CF = "info";
  66.     /**
  67.      * 收件箱表-最大版本
  68.      */
  69.     public static final Integer INBOX_TABLE_VERSIONS = 2;
  70. }

3. 工具类

  1. package com.jshawn.hbase.guli_weibo.utils;
  3. import com.jshawn.hbase.guli_weibo.constants.Constants;
  4. import org.apache.hadoop.hbase.*;
  5. import org.apache.hadoop.hbase.client.Admin;
  6. import org.apache.hadoop.hbase.client.Connection;
  7. import org.apache.hadoop.hbase.client.ConnectionFactory;
  9. import java.io.IOException;
  11. /**
  12.  * @author JShawn 2021/6/20 16:41
  13.  *
  14.  * 1. 创建命名空间
  15.  * 2. 判断表是否存在
  16.  * 3. 创建表（三张表）
  17.  */
  18. public class HBaseUtils {
  20.     /**
  21.      * 1. 创建命名空间
  22.      * @param nameSpace     命名空间（相当于数据库）
  23.      * @throws IOException
  24.      */
  25.     public static void createNameSpace(String nameSpace) throws IOException {
  27.         // 1.获取Connnection对象
  28.         Connection connection = ConnectionFactory.createConnection(Constants.COFIGURATION);
  30.         // 2. 获取Admin对象
  31.         Admin admin = connection.getAdmin();
  33.         // 3. 构建命名空间描述器
  34.         NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();
  36.         // 4. 创建命名空间
  37.         admin.createNamespace(namespaceDescriptor);
  39.         // 5. 关闭资源
  40.         admin.close();
  41.         connection.close();
  42.     }
  44.     /**
  45.      * 2. 判断表是否存在
  46.      * @param tableName     表名
  47.      * @return
  48.      */
  49.     private static boolean isTableExists(String tableName) throws IOException {
  51.         // 1.获取Connnection对象
  52.         Connection connection = ConnectionFactory.createConnection(Constants.COFIGURATION);
  54.         // 2. 获取Admin对象
  55.         Admin admin = connection.getAdmin();
  57.         // 3. 判断表是否存在
  58.         boolean exists = admin.tableExists(TableName.valueOf(tableName));
  60.         // 4. 关闭资源
  61.         admin.close();
  62.         connection.close();
  64.         // 5. 返回结果
  65.         return exists;
  66.     }
  68.     /**
  69.      * 3. 创建表
  70.      * @param tableName     表名
  71.      * @param versions      最大版本
  72.      * @param cfs           列族信息
  73.      */
  74.     public static void createTable(String tableName, int versions, String... cfs) throws IOException {
  76.         // 1. 判断是否传入了列族信息
  77.         if (cfs.length <= 0) {
  78.             System.out.println("请设置列族信息！！！");
  79.             return;
  80.         }
  81.         // 2. 判断表是否存在
  82.         if (isTableExists(tableName)) {
  83.             System.out.println("表【" + tableName + "】已存在！！！");
  84.             return;
  85.         }
  87.         // 3. 获取connection对象
  88.         Connection connection = ConnectionFactory.createConnection(Constants.COFIGURATION);
  90.         // 4. 获取Admin对象
  91.         Admin admin = connection.getAdmin();
  93.         // 5. 创建表描述器
  94.         HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
  96.         // 6. 添加列族信息
  97.         for (String cf : cfs) {
  99.             HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cf);
  101.             // 7. 设置版本
  102.             hColumnDescriptor.setMaxVersions(versions);
  104.             hTableDescriptor.addFamily(hColumnDescriptor);
  106.         }
  109.         // 8. 创建表
  110.         admin.createTable(hTableDescriptor);
  112.         // 9. 关闭资源
  113.         admin.close();
  114.         connection.close();
  115.     }
  116. }

4. 业务类

4.1 创建名称空间和表结构

  1. /**
  2.  * 0. 创建命名空间和表结构
  3.  */
  4. public static void init() {
  6.     try {
  7.         // 创建命名空间
  8.         HBaseUtils.createNameSpace(Constants.NAMESPACE);
  10.         // 创建微博内容表
  11.         HBaseUtils.createTable(Constants.CONTENT_TABLE, Constants.CONTENT_TABLE_VERSIONS, Constants.CONTENT_TABLE_CF);
  13.         // 创建用户关系表
  14.         HBaseUtils.createTable(Constants.RELATION_TABLE, Constants.RELATION_TABLE_VERSIONS, Constants.RELATION_TABLE_CF_ATTENDS, Constants.RELATION_TABLE_CF_FANS);
  16.         // 创建收件箱表
  17.         HBaseUtils.createTable(Constants.INBOX_TABLE, Constants.INBOX_TABLE_VERSIONS, Constants.INBOX_TABLE_CF);
  19.     } catch (IOException e) {
  20.         e.printStackTrace();
  21.     }
  22. }

4.2 发布微博

image.png

  1. /**
  2.  * 1. 发布微博
  3.  * @param uid           用户ID
  4.  * @param content      微博内容
  5.  * @throws IOException
  6.  */
  7. public static void publishWeibo(String uid, String content) throws IOException {
  9.     // 获取connection对象
  10.     Connection connection = ConnectionFactory.createConnection(Constants.COFIGURATION);
  12.     // 第一部分：操作微博内容表(增加一条微博内容)
  13.     // 1. 获取微博内容表对象
  14.     Table contentTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
  16.     // 2. 获取当前时间戳
  17.     long timestamp = System.currentTimeMillis();
  19.     // 3. 获取RowKey
  20.     String rowKey = uid + "_" + timestamp;
  22.     // 4. 创建Put对象
  23.     Put contentPut = new Put(Bytes.toBytes(rowKey));
  25.     // 5. 给Put对象赋值
  26.     contentPut.addColumn(Bytes.toBytes(Constants.CONTENT_TABLE_CF), Bytes.toBytes("content"), Bytes.toBytes(content));
  28.     // 6. 执行插入操作
  29.     contentTable.put(contentPut);
  31.     // 第二部分：操作微博收件箱表
  32.     // 获取发布微博的用户有哪些粉丝
  33.     // 1. 获取用户关系表对象
  34.     Table relationTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
  36.     // 2. 获取当前发布微博人的fans列族数据
  37.     Get get = new Get(Bytes.toBytes(uid));
  38.     get.addFamily(Bytes.toBytes(Constants.RELATION_TABLE_CF_FANS));
  39.     Result result = relationTable.get(get);
  41.     // 3. 创建一个集合，用于存放微博内容表的Put对象
  42.     ArrayList<Put> inboxPuts = new ArrayList<>();
  44.     //4. 遍历粉丝
  45.     for (Cell cell : result.rawCells()) {
  47.         // 5. 构建微博收件箱表的Put对象
  48.         Put inboxPut = new Put(CellUtil.cloneQualifier(cell));
  50.         // 6. 给收件箱表的Put对象赋值
  51.         inboxPut.addColumn(Bytes.toBytes(Constants.INBOX_TABLE_CF),Bytes.toBytes(uid),Bytes.toBytes(rowKey));
  53.         // 7. 将收件箱表的Put对象存入集合
  54.         inboxPuts.add(inboxPut);
  55.     }
  58.     // 8. 判断是否有粉丝
  59.     if (!inboxPuts.isEmpty()) {
  61.         // 获取收件箱表对象
  62.         Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
  64.         // 执行收件箱表数据批量插入操作
  65.         inboxTable.put(inboxPuts);
  67.         // 关闭收件箱表
  68.         inboxTable.close();
  69.     }
  71.     // 9. 关闭资源
  72.     contentTable.close();
  73.     relationTable.close();
  74.     connection.close();
  75. }

4.3 添加关注

image.png

  1. /**
  2.  * 2. 关注用户
  3.  * @param uid           主用户ID
  4.  * @param attends      被关注人ID
  5.  */
  6. public static void addAttends(String uid, String... attends) throws IOException {
  8.     //  校验是否添加了待关注的人
  9.     if (attends.length <= 0) {
  10.         System.out.println("请添加关注人！！！");
  11.         return;
  12.     }
  14.     // 获取Connection对象
  15.     Connection connection = ConnectionFactory.createConnection(Constants.COFIGURATION);
  17.     // 第一部分：操作用户关系表
  18.     // 1. 获取用户关系表对象
  19.     Table relationTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
  21.     // 2. 创建一个集合用于存放用户关系表的Put对象（主用户attend列族添加1个，被关注人fans列族添加attends.length个）
  22.     LinkedList<Put> puts = new LinkedList<>();
  24.     // 3. 创建主用户的Put对象
  25.     Put uidPut = new Put(Bytes.toBytes(uid));
  27.     // 4. 循环创建被关注者的Put对象
  28.     for (String attend : attends) {
  30.         //5. 给主用户的Put对象赋值
  31.         uidPut.addColumn(Bytes.toBytes(Constants.RELATION_TABLE_CF_ATTENDS),Bytes.toBytes(attend),Bytes.toBytes(attend));
  33.         //6. 创建被关注者的Put对象
  34.         Put put = new Put(Bytes.toBytes(attend));
  36.         // 7. 给被关注者的Put对象赋值
  37.         put.addColumn(Bytes.toBytes(Constants.RELATION_TABLE_CF_FANS), Bytes.toBytes(uid), Bytes.toBytes(uid));
  39.         // 8. 将被关注者的Put对象放入集合
  40.         puts.add(put);
  41.     }
  44.     // 9. 将主用户的Put对象放入集合
  45.     puts.add(uidPut);
  47.     // 10. 执行用户关系表的插入数据操作
  48.     relationTable.put(puts);
  50.     // 第二部分：操作收件箱表
  51.     // 1. 获取微博内容表对象
  52.     Table contentTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
  54.     // 2. 创建微博收件箱表的Put对象
  55.     Put inboxPut = new Put(Bytes.toBytes(uid));
  57.     // 3. 循环attends，获取每个被关注者的近期发布的微博
  58.     for (String attend : attends) {
  60.         // 4. 获取当前被关注者的近期发布的微博（scan）->集合ResultScanner
  61.         // 思路一（此处采用）：先获取出被关注者的所有微博，然后再取前三条（由于发布微博时是rowkey是增加了时间戳，因此HBase中就是顺序存，取出也是顺序的）
  62.         //         由于微博内容表的rowkey设计是uid_时间戳
  63.         //         因此 startRow 可以写成uid_,这样就比该用户所有数据rowkey都小
  64.         //              stopRow 可以写出uid_|,这样就比该用户所有rowkey都大，因为|在unicode编码中数据是最大的
  65.         //         获取到所有微博后再取前三条
  66.         //
  67.         // 思路二：发布微博的时候rowkey不添加时间戳，而是（比如时间戳有13位）用13位数的最大值（13个9）-当前时间戳
  68.         //         这时候存入就是倒序存入的了
  69.         //         然后在这里取值直接就取到前三条即可
  70.         //
  71.         //由于要修改代码，这里选取思路一
  72.         Scan scan = new Scan(Bytes.toBytes(attend + "_"), Bytes.toBytes(attend + "|"));
  73.         ResultScanner resultScanner = contentTable.getScanner(scan);
  75.         // 定义一个时间戳（然后使用++保证时间戳不一样，否则在这里批量写入以后，HBase服务端会自动加时间戳，
  76.         //                 下次再取的时候由于时间戳一样只能取到一条，所有需要手动加一个时间戳）
  77.         long timeStamp = System.currentTimeMillis();
  79.             // 5. 对获取的值进行遍历
  80.         for (Result result : resultScanner) {
  82.                 // 6. 给收件箱表的Put对象赋值
  83.             inboxPut.addColumn(Bytes.toBytes(Constants.INBOX_TABLE_CF), Bytes.toBytes(attend), timeStamp++, result.getRow());
  84.         }
  86.     }
  89.     // 7. 判断当前的Put对象是不为空
  90.     if (!inboxPut.isEmpty()) {
  92.         // 不为空，获取微博收件箱表对象
  93.         Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
  95.         // 插入数据
  96.         inboxTable.put(inboxPut);
  98.         // 关闭收件箱表对象连接
  99.         inboxTable.close();
  100.     }
  103.     // 关闭资源
  104.     relationTable.close();
  105.     contentTable.close();
  106.     connection.close();
  107. }

4.4 取关用户

  1. /**
  2.  * 3. 取关用户
  3.  * @param uid       主用户ID
  4.  * @param dels      被关注人ID
  5.  */
  6. public static void deleteAttend(String uid,String ...dels) throws IOException {
  8.     if (dels.length <= 0) {
  9.         System.out.println("请添加取关的用户！！！");
  10.         return;
  11.     }
  13.     // 获取Connection对象
  14.     Connection connection = ConnectionFactory.createConnection(Constants.COFIGURATION);
  16.     // 第一部分：操作用户关系表
  17.     // 1. 获取用户关系表对象
  18.     Table relaTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
  20.     // 2. 创建一个集合，用于存放用户关系表的Delete对象
  21.     LinkedList<Delete> relaDeletes = new LinkedList<>();
  23.     // 3. 创建主用户的Delete对象
  24.     Delete uidDelete = new Delete(Bytes.toBytes(uid));
  26.     // 4. 循环创建被取关者的Delete对象
  27.     for (String del : dels) {
  29.         // 5. 给主用户的Delete对象赋值
  30.         // 注意：这里要使用带s的api，防止多次关注（一般不会，会在前端做校验）
  31.         uidDelete.addColumns(Bytes.toBytes(Constants.RELATION_TABLE_CF_ATTENDS),Bytes.toBytes(del));
  33.         // 6. 创建被取关者的Delete对象，删除粉丝
  34.         Delete delete = new Delete(Bytes.toBytes(del));
  36.         // 7. 给被取关者的Delete对象赋值（赋值到列，因为不是删rowkey，只是删被取关的人，不是删全部）
  37.         delete.addColumns(Bytes.toBytes(Constants.RELATION_TABLE_CF_FANS), Bytes.toBytes(uid));
  39.         // 8. 将被取关者的Delete对象添加至集合
  40.         relaDeletes.add(delete);
  41.     }
  43.     // 9. 将操作者的Delete对象添加至结合
  44.     relaDeletes.add(uidDelete);
  46.     //10. 执行用户关系表的批量删除操作
  47.     relaTable.delete(relaDeletes);
  50.     // 第二部分：操作收件箱表
  51.     // 1. 获取收件箱表对象
  52.     Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
  54.     // 2. 创建操作者的Delete对象
  55.     Delete inboxDelete = new Delete(Bytes.toBytes(uid));
  57.     // 3. 循环给操作者的Delete对象赋值
  58.     for (String del : dels) {
  59.         inboxDelete.addColumns(Bytes.toBytes(Constants.INBOX_TABLE_CF), Bytes.toBytes(del));
  60.     }
  62.     // 4. 执行收件箱表的删除操作
  63.     inboxTable.delete(inboxDelete);
  65.     // 关闭资源
  66.     relaTable.close();
  67.     inboxTable.close();
  68.     connection.close();
  69. }

4.5 获取某个用户的初始化页面数据

  1. /**
  2.  * 4. 获取某个用户的初始化页面数据
  3.  * @param uid        用户ID
  4.  */
  5. public static void getInit(String uid) throws IOException {
  7.     // 1. 获取Connection对象
  8.     Connection connection = ConnectionFactory.createConnection(Constants.COFIGURATION);
  10.     // 2. 获取收件箱表对象
  11.     Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
  13.     // 3. 获取微博内容表对象
  14.     Table contentTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
  16.     // 4. 创建收件箱表Get对象，并获取数据（设置最大版本）
  17.     Get inboxGet = new Get(Bytes.toBytes(uid));
  18.     inboxGet.setMaxVersions();
  19.     Result result = inboxTable.get(inboxGet);
  21.     // 5. 遍历获取的数据
  22.     for (Cell cell : result.rawCells()) {
  24.         // 6. 构建微博内容表Get对象
  25.         Get contentGet = new Get(CellUtil.cloneValue(cell));
  27.         // 7. 获取该Get对象的数据内容
  28.         Result contentReult = contentTable.get(contentGet);
  30.         // 8. 解析内容并打印
  31.         for (Cell contentCell : contentReult.rawCells()) {
  32.             System.out.println("RowKey：" + Bytes.toString(CellUtil.cloneRow(contentCell)) +
  33.                     ",CF：" + Bytes.toString(CellUtil.cloneFamily(contentCell)) +
  34.                     "，CN：" + Bytes.toString(CellUtil.cloneQualifier(contentCell)) +
  35.                     "，Value：" + Bytes.toString(CellUtil.cloneValue(contentCell)));
  36.         }
  37.     }
  40.     // 9. 关闭资源
  41.     inboxTable.close();
  42.     contentTable.close();
  43.     connection.close();
  44. }

4.6 获取某个用户所有微博详情

  1. /**
  2.  * 5. 获取某个用户所有微博详情
  3.  * @param uid        用户ID
  4.  */
  5. public static void getWeibo(String uid) throws IOException {
  7.     // 1. 获取Connection对象
  8.     Connection connection = ConnectionFactory.createConnection(Constants.COFIGURATION);
  10.     // 2. 获取微博内容表对象
  11.     Table table = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
  13.     // 3. 构建Scan对象
  14.     // 思路一：同添加关注时的思路一
  15.     //         startRow：uid_
  16.     //         stopRow：uid|
  17.     // 思路二（此处采用）：过滤器
  18.     Scan scan = new Scan();
  20.     // 构建过滤器
  21.     // 表示过滤出 RowKey 中有uid_"-"的字串的数据
  22.     RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(uid + "_"));
  23.     scan.setFilter(rowFilter);
  25.     // 4. 获取数据
  26.     ResultScanner resultScanner = table.getScanner(scan);
  28.     // 5. 解析数据并打印
  29.     for (Result result : resultScanner) {
  30.         for (Cell cell : result.rawCells()) {
  31.             System.out.println("RowKey：" + Bytes.toString(CellUtil.cloneRow(cell)) +
  32.                     ",CF：" + Bytes.toString(CellUtil.cloneFamily(cell)) +
  33.                     "，CN：" + Bytes.toString(CellUtil.cloneQualifier(cell)) +
  34.                     "，Value：" + Bytes.toString(CellUtil.cloneValue(cell)));
  35.         }
  36.     }
  38.     // 6. 关闭资源
  39.     table.close();
  40.     connection.close();
  41. }

5. 测试类

  1. package com.jshawn.hbase.guli_weibo.test;
  3. import com.jshawn.hbase.guli_weibo.constants.Constants;
  4. import com.jshawn.hbase.guli_weibo.dao.HBaseDao;
  5. import com.jshawn.hbase.guli_weibo.utils.HBaseUtils;
  7. import java.io.IOException;
  9. /**
  10.  * 谷粒微博测试类
  11.  * @author JShawn 2021/6/20 22:35
  12.  */
  13. public class TestWeibo {
  15.     public static void main(String[] args) throws IOException, InterruptedException {
  17.         // 创建命名空间和表结构
  18.         HBaseDao.init();
  20.         // 1001发布微博
  21.         HBaseDao.publishWeibo("1001","赶紧下课吧");
  23.         // 1002关注1001和1003
  24.         HBaseDao.addAttends("1002","1001","1003");
  26.         // 获取1002初始化页面
  27.         HBaseDao.getInit("1002");
  29.         System.out.println("***************************111*******************************");
  31.         // 1003发布3条微博，同时1001发布2条微博
  32.         HBaseDao.publishWeibo("1003","谁说的赶紧下课！！！");
  33.         Thread.sleep(10);
  34.         HBaseDao.publishWeibo("1001","我没说话！！！");
  35.         Thread.sleep(10);
  36.         HBaseDao.publishWeibo("1003","那谁说的！！！");
  37.         Thread.sleep(10);
  38.         HBaseDao.publishWeibo("1001","反正我没说！！！");
  39.         Thread.sleep(10);
  40.         HBaseDao.publishWeibo("1003","你们爱咋咋地！！！");
  42.         // 获取1002初始化页面
  43.         HBaseDao.getInit("1002");
  44.         System.out.println("***************************222*******************************");
  46.         // 1002取关1003
  47.         HBaseDao.deleteAttend("1002","1003");
  49.         // 获取1002初始化页面
  50.         HBaseDao.getInit("1002");
  51.         System.out.println("***************************333*******************************");
  53.         // 1002再次关注1003
  54.         HBaseDao.addAttends("1002","1003");
  56.         // 获取1002初始化页面
  57.         HBaseDao.getInit("1002");
  58.         System.out.println("***************************444*******************************");
  60.         // 获取1001微博详情
  61.         HBaseDao.getWeibo("1001");
  62.     }
  63. }