Hbase 实现微博好友关注功能

前言

在社交网站中，比如像QQ,微信，新浪微博之类的，像这种社交属性强的app，里面必然会有好友关注、好友推荐、好友动态等这样的功能；

拿好友功能来说，一般来讲，当用户量还不够大的时候，mysql就可以解决，当用户量继续增长的时候，mysql配合redis也可以解决，当数据继续增长，上升到诸如像微博，QQ这样的体量时，可能需要考虑存储效率更好、承载容量更大的大数据存储引擎了，比如像 hbase这种轻松可以容纳TB甚至PB级别数据的大数据库引擎；

业务需求分析

本例需实现2个功能

好友关注与取关注；
发布微博；

这是两个几乎所有社交类APP都会涉及到的功能，拿出来探讨并通过代码进行实现；

实现过分析

从业务实现的需求出发，既然是使用Habse来存储数据，肯定是先分析表结构，把需要创建的表规划清楚；

1、好友关注功能设计思路

对于好友关注与取关注这个功能，首先想到的，肯定是用户关系表，通过这个表，保存好友之间的关系，具体该如何设计呢？

以上面这张图为例进行说明：

创建一张保存用户关系的表，cf有两个，分别为关注人列表和粉丝列表，这里简化操作，忽略了用户的其他字段，仅保存了用户的主键ID，而rk为每个用户，以user1用户来说，关注人列表中有 user2,user3和user4，而粉丝列表中有 user3,user4和user5；
对于user1来讲，每次关注一个新用户，则往rk为user1的这一栏的关注人列表中，添加新的关注人即可；
同样，当有人关注了user1时，就在rk为user1的这一栏的粉丝列簇中，添加一条数据；
再则，我们看到，当user1关注了user2的时候，对usr1做了操作之后，同样需要对user2这个数据做同样的操作；

上面的分析即为编写代码时候的指导思路，下面就开始编码过程实现吧；

2、发布微博功能设计思路

设想一下，当我们发布一条微博动态的时候，最直接的就是我们的粉丝，粉丝在进入首页刷新页面的时候，我们的粉丝将会接收到我们发布最新的内容，因此，很明显这里需要一张微博内容表；

内容表比较简单，cf为微博内容，rk以user_id即可，当某个用户发布一条微博内容时，往该表插入一条数据即可，问题是，用户会发布多个动态，于是以时间戳（版本）来区分，实际在操作的时候，可以以时间错倒叙进行存储；

3、获取微博动态内容功能设计思路

当我们打开为微博或微信的时候，就能收到我们关注好友的最新的发布内容，这个是怎么做到的呢？

从上面的两张表分析来看，好像是各自独立的，因为好友关注和用户发布微博写到内容表确实是两个不相干的操作，而对于某个具体的用户来讲，刷微博的时候，我们能够联想到，自然是从粉丝表中拿到粉丝，再把这些粉丝最近发布的微博拿出来进行展现不就好了吗？

这个思路其实也不是不可以，问题是，操作繁琐，有没有什么办法可以解决呢？于是，我们想到可以使用一张中间关联表；

从这个表的存储结构来看，这张表冗余了某个用户的关注人的发布的微博内容，以user1用户为例做说明，列簇中，维护了一个个关注人以及关注人对应的内容rowkey，实际中，根据实际微博内容展示的数量，可以冗余5_{10条，而这5}10条内容的，正好按照时间倒叙取过来即可；

那么这个表的数据什么时候存进去呢？

很明显可以想到，这是在某个用户发布一条微博内容的时候，那么梳理下这个过程，大概如下：

user1有个粉丝user2，user1发布一条微博动态；
usr1发布的微博内容将会存储到微博内容表；
由于user2是user1的粉丝，对于user2来说，将会在上面的内容关系表中，将user1发表的内容rowkey拿过来存储到这个表中

环境准备

1、基于centos7或者windows环境，搭建一台单机版的hbase服务环境，我这边已提前准备好并开启了hbase 的shell客户端；

2、添加基础maven依赖

     <dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>1.3.1</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase</artifactId><version>1.3.1</version></dependency>

3、定义一个常量类，用于保存3张表名称以及相关的列簇信息

public class WeiboConstants {/*** 微博命名空间*/public static final String NMAE_SPACE = "weibo";/*** 微博内容表*/public static final String CONTENT_TABLE= "weibo:content";public static final String CONTENT_TABLE_CF= "info";public static final int CONTENT_TABLE_VERSIONS= 1;/*** 用户关系表*/public static final String RELATION_TABLE= "weibo:relation";public static final String RELATION_TABLE_CF1= "attends";public static final String RELATION_TABLE_CF2= "fans";public static final int RELATION_TABLE_VERSIONS= 1;/*** 用户内容关系表*/public static final String INBOX_TABLE= "weibo:inbox";public static final String INBOX_TABLE_CF= "info";public static final int INBOX_TABLE_VERSIONS= 2;}

4、基础工具类

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;import java.io.IOException;public class CommonUtils {public static Connection connection = null;public static Admin admin = null;static {Configuration conf = HBaseConfiguration.create();//使用 HBaseConfiguration 的单例方法实例化conf = HBaseConfiguration.create();conf.set("hbase.zookeeper.quorum", "127.0.0.1");conf.set("hbase.zookeeper.property.clientPort", "2181");try {connection = ConnectionFactory.createConnection(conf);admin = connection.getAdmin();} catch (IOException e) {e.printStackTrace();}}public static Connection getConnection(){return connection;}/*** 创建表* @param tableName 表名* @param columnFamily 列簇名* @throws Exception*/public static void createTable(String tableName, int versions,String... columnFamily) throws Exception {if (columnFamily.length <= 0) {System.out.println("请传入列簇信息");}//判断表是否存在if (isTableExists(tableName)) {System.out.println("表" + tableName + "已存在");close();return;}//创建表属性对象,表名需要转字节HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));//循环创建多个列族for (String cf : columnFamily) {HColumnDescriptor columnDescriptor = new HColumnDescriptor(cf);columnDescriptor.setMaxVersions(versions);descriptor.addFamily(columnDescriptor);}//根据对表的配置，创建表admin.createTable(descriptor);System.out.println("表" + tableName + "创建成功！");close();}/*** 判断表是否存在* @param tableName* @return* @throws Exception*/public static boolean isTableExists(String tableName) throws Exception {boolean result = admin.tableExists(TableName.valueOf(tableName));return result;}/*** 创建命名空间* @param nameSpace*/public static void createNameSpace(String nameSpace){if(nameSpace == null){System.out.println(nameSpace + ": 不存在 !" );return;}NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();try {admin.createNamespace(namespaceDescriptor);} catch (NamespaceExistException e){System.out.println("命名空间已存在");}catch (IOException e) {e.printStackTrace();}System.out.println(nameSpace + ": 命名空间创建成功");}/*** 关闭连接*/public static void close() {if (admin != null) {try {admin.close();} catch (IOException e) {e.printStackTrace();}}if (connection != null) {try {connection.close();} catch (IOException e) {e.printStackTrace();}}}}

一、发布微博代码实现

以用户1为例，用户1将要发布一条微博，那么要进行什么样的操作呢？

代码逻辑

操作主表：微博内容表，用户1发布的内容插入到微博内容表
在用户关系表中，找到用户1的粉丝列表；
拿到用户1的粉丝列表后，再在内容关系表中，依次将本次的内容插入到内容关系表

具体代码实现过程可参考代码中的注释

public class WeiBoService {private static CommonUtils commonUtils = new CommonUtils();/*** 发布微博** @param uid* @param content*/public static void publish(String uid, String content) throws Exception {//1、操作微博内容表Connection connection = commonUtils.getConnection();Table contentTable = connection.getTable(TableName.valueOf(WeiboConstants.CONTENT_TABLE));//2、操作微博内容表long ts = System.currentTimeMillis();String rowKey = uid + "_" + ts;Put contentPut = new Put(Bytes.toBytes(rowKey));contentPut.addColumn(Bytes.toBytes(WeiboConstants.CONTENT_TABLE_CF),Bytes.toBytes("content"),Bytes.toBytes(content));contentTable.put(contentPut);//3、操作微博内容关联表Table relationTable = connection.getTable(TableName.valueOf(WeiboConstants.RELATION_TABLE));//获取当前发布人的粉丝列表Get get = new Get(Bytes.toBytes(uid));get.addFamily(Bytes.toBytes(WeiboConstants.RELATION_TABLE_CF2));Result result = relationTable.get(get);//4、遍历这个人的粉丝，循环构建微博内容关系列表，往该表插入数据List<Put> inboxPuts = new ArrayList<>();for (Cell cell : result.rawCells()) {Put inboxPut = new Put(CellUtil.cloneQualifier(cell));//将当前用户发布的内容放进去inboxPut.addColumn(Bytes.toBytes(WeiboConstants.INBOX_TABLE_CF),Bytes.toBytes(uid),Bytes.toBytes(rowKey));inboxPuts.add(inboxPut);}//5、将内容关系数据入库if (inboxPuts != null && inboxPuts.size() > 0) {Table inboxTable = connection.getTable(TableName.valueOf(WeiboConstants.INBOX_TABLE));inboxTable.put(inboxPuts);}//6、关闭资源commonUtils.close();}}

二、关注好友功能代码实现

场景，现在用户1要关注用户2，用户3，用户4…

代码逻辑分析：

操作主表，用户关系表；
用户关系表有2个列簇，rw为当前操作人即user1的ID，关注人和粉丝各自一个cf；
首先，给用户1的关注的cf插入数据；
用户1关注了2，3，4，那么，1将作为2，3，4的粉丝，需要将1分别添加到2，3，4用户的粉丝列表对应的cf中；
需要在内容关系表中，新增rk为user1，指定列簇下，2，3，4用户的最新发布的微博内容rk的数据，便于后续获取微博动态时使用；

import com.congge.constants.WeiboConstants;
import com.congge.utils.CommonUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;import java.util.ArrayList;
import java.util.List;public class WeiBoService {private static CommonUtils commonUtils = new CommonUtils();/*** 关注好友** @param uid* @param attends*/public static void attendUser(String uid, String... attends) throws Exception {if (attends.length <= 0) {System.out.println("请选择关注人");return;}Connection connection = commonUtils.getConnection();//1、关系表添加数据Table relationTable = connection.getTable(TableName.valueOf(WeiboConstants.RELATION_TABLE));List<Put> allPuts = new ArrayList<>();Put uidPut = new Put(Bytes.toBytes(uid));for (String attend : attends) {//2、给当前操作人循环添加粉丝uidPut.addColumn(Bytes.toBytes(WeiboConstants.RELATION_TABLE_CF1),Bytes.toBytes(attend), Bytes.toBytes(attend));//3、同时，对于被关注的人来说，他们的粉丝列簇里面需要把当前操作人放进去Put attendPut = new Put(Bytes.toBytes(attend));attendPut.addColumn(Bytes.toBytes(WeiboConstants.RELATION_TABLE_CF2),Bytes.toBytes(uid), Bytes.toBytes(uid));allPuts.add(attendPut);}//将操作人自身的put对象也添加进去allPuts.add(uidPut);//数据写入关系表relationTable.put(allPuts);//2、操作内容表Table contentTable = connection.getTable(TableName.valueOf(WeiboConstants.CONTENT_TABLE));//创建关系内容表的PUT对象Put inboxPut = new Put(Bytes.toBytes(uid));long ts = System.currentTimeMillis();// 获取微博内容表数据，然后依次添加到关系表for (String attend : attends) {//获取这个人近期发布的微博内容Scan scan = new Scan(Bytes.toBytes(attend + "_"), Bytes.toBytes(attend + "|"));ResultScanner resultScanner = contentTable.getScanner(scan);for (Result result : resultScanner) {inboxPut.addColumn(Bytes.toBytes(WeiboConstants.INBOX_TABLE_CF),Bytes.toBytes(attend), ts++, result.getRow());}}//插入数据if (!inboxPut.isEmpty()) {Table inboxTable = connection.getTable(TableName.valueOf(WeiboConstants.INBOX_TABLE));inboxTable.put(inboxPut);}}
}

二、获取某用户微博动态代码实现

场景：假设用户1打开微博，这时候将会出现用户1的好友的最新的微博内容

实现逻辑分析：

以用户1的rk查询微博内容关系表的数据；
第一步中查到的，由于只是其关注用户的rk，需要拿到r再去内容表中回查；
再去内容表中，查到各个关注人的rk对应的微博内容即可；

import com.congge.constants.WeiboConstants;
import com.congge.utils.CommonUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;import java.util.ArrayList;
import java.util.List;public class WeiBoService {private static CommonUtils commonUtils = new CommonUtils();/*** 获取某用户的初始化页面** @param uid*/public static void getInit(String uid) throws Exception {Connection connection = commonUtils.getConnection();Table inboxTable = connection.getTable(TableName.valueOf(WeiboConstants.INBOX_TABLE));Table contentTable = connection.getTable(TableName.valueOf(WeiboConstants.CONTENT_TABLE));//创建内容关联表对象Get inboxGet = new Get(Bytes.toBytes(uid));inboxGet.setMaxVersions();Result result = inboxTable.get(inboxGet);for (Cell cell : result.rawCells()) {//构建微博内容GET对象Get contentGet = new Get(CellUtil.cloneQualifier(cell));Result contentResult = contentTable.get(contentGet);for (Cell contentCell : contentResult.rawCells()) {System.out.println("RK:" + Bytes.toString(CellUtil.cloneRow(contentCell)) +",CF:" + Bytes.toString(CellUtil.cloneFamily(contentCell)) +",CN :" + Bytes.toString(CellUtil.cloneQualifier(contentCell)) +",Value :" + Bytes.toString(CellUtil.cloneValue(contentCell)));}}commonUtils.close();}}