Hbase之微博项目
一、微博系统介绍
1. 需求分析
- 微博内容的浏览，数据库表设计
- 用户社交体现：关注用户，取关用户
- 拉取关注的人的微博内容
2. 所需要的表
image
3. 代码实现
创建命名空间以及表名的定义
创建微博内容表
创建用户关系表
创建用户微博内容接收邮件表
发布微博内容
添加关注用户
移除（取关）用户
获取关注的人的微博内容
测试
二、代码实现
2.1 创建命名空间以及表名的定义
//设置配置信息
private static Configuration conf;
private Logger logger = Logger.getLogger(WeiBo.class);
//设置hbase的配置信息——hbase是强依赖于hdfs和zookeeper的
//与hbase-site.xml中一致
static {
conf = HBaseConfiguration.create();
conf.set("hbase.rootdir", "hdfs://master:9000/hbase"); //设置hbase数据在hdfs中存放位置
conf.set("hbase.zookeeper.quorum","master:2181,slave1:2181,slave2:2181"); //配置zookeeper
}
//创建微博namespace——三张表
//定义namespace名称
private static final String NS_WEIBO = "ns_weibo";
private static final byte[] TABLE_CONTENT = Bytes.toBytes("ns_weibo:content");
private static final byte[] TABLE_RELATION = Bytes.toBytes("ns_weibo:relation");
private static final byte[] TABLE_INBOX = Bytes.toBytes("ns_weibo:inbox");
//初始化表
private void init() throws IOException {
initNameSpace(); //初始化NameSpace
initTableContent(); //初始化微博内容表
initTableRelation(); //初始化用户关系表
initTableInbox(); //初始化收件箱表
}
///初始化NameSpace
private void initNameSpace() throws IOException {
logger.info("正在初始化namespace");
// TODO Auto-generated method stub
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
//创建命名空间描述器
NamespaceDescriptor ns_weibo1 = NamespaceDescriptor
.create(NS_WEIBO)
.addConfiguration("author", "anna")
.addConfiguration("create_time", String.valueOf(System.currentTimeMillis()))
.build();
admin.createNamespace(ns_weibo1);
admin.close();
connection.close();
logger.info(NS_WEIBO + "命名空间创建成功");
}
2.2 创建微博内容表
-----------------------------------------------------------------------------------
/**
*
* 初始化微博内容表
*
* 表结构
* TableName : ns_weibo:content
* rowKey:用户ID_时间戳
* ColumnFamily : info
* ColumnQualifier : content
* Value : 微博内容(文字内容;图片URL;视频URL;语音URL)
* Versions : 1
* 此时的版本为1指的是：
* 对于同一rowKey中的数据只能存放一个版本
* 也就是说：当对一个rowKey进行二次添加数据时，则为覆盖
* @throws IOException
*/
private void initTableContent() throws IOException {
// TODO Auto-generated method stub
logger.info("正在初始化content表");
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
//创建表描述器
HTableDescriptor contentTableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_CONTENT));
//创建列描述器
HColumnDescriptor infoColumnDescriptor = new HColumnDescriptor("info");
//设置块缓存
infoColumnDescriptor.setBlockCacheEnabled(true);
//设置块缓存大小
infoColumnDescriptor.setBlocksize(2 * 1024 * 1024);
//设置版本确界
infoColumnDescriptor.setMinVersions(1);
infoColumnDescriptor.setMaxVersions(1);
//将列描述器添加到表描述器中
contentTableDescriptor.addFamily(infoColumnDescriptor);
//创建表
admin.createTable(contentTableDescriptor);
admin.close();
connection.close();
logger.info("content表创建成功");
}
2.3 创建用户关系表
/**
*
* 初始化用户关系表
*
* TableName : ns_weibo:relation
* rowKey:当前用户ID
* ColumnFamily : attends fans 两个列族——两个HColumnDescriptor
* ColumnQualifier : 用户ID
* Value : 无实际含义的空字符串
* Versions : 1
* 此时的版本为1指的是：
* 对于同一rowKey中的数据只能存放一个版本
* 也就是说：当对一个rowKey进行二次添加数据时，则为覆盖
* @throws IOException
*/
private void initTableRelation() throws IOException {
logger.info("正在初始化relation表");
// TODO Auto-generated method stub
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
//创建表描述器
HTableDescriptor relationTableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_RELATION));
//创建列描述器
HColumnDescriptor attendsColumnDescriptor = new HColumnDescriptor("attends");
HColumnDescriptor fansColumnDescriptor = new HColumnDescriptor("fans");
//设置块缓存
attendsColumnDescriptor.setBlockCacheEnabled(true);
fansColumnDescriptor.setBlockCacheEnabled(true);
//设置块缓存大小
attendsColumnDescriptor.setBlocksize(2 * 1024 * 1024);
fansColumnDescriptor.setBlocksize(2 * 1024 * 1024);
//设置版本确界
attendsColumnDescriptor.setMinVersions(1);
attendsColumnDescriptor.setMaxVersions(1);
fansColumnDescriptor.setMinVersions(1);
fansColumnDescriptor.setMaxVersions(1);
//将列描述器添加到表描述器中
relationTableDescriptor.addFamily(attendsColumnDescriptor);
relationTableDescriptor.addFamily(fansColumnDescriptor);
//创建表
admin.createTable(relationTableDescriptor);
admin.close();
connection.close();
logger.info("relation表创建成功");
}
2.4 创建微博收件箱表
/**
*
* 初始化收件箱表
*
* TableName : ns_weibo:inbox
* rowKey:当前用户ID
* ColumnFamily : info
* ColumnQualifier : 关注人的用户ID
* Value : 关注人的微博rowKey
* Versions : 100
* 同一个rowKey中的数据可以存放100个版本
* 若想获取最近的5条微博，则可以通过Get类中的setMaxVersions来设置获取的版本数，新API中使用的是readVersions(int versions)方法
* @throws IOException
*/
private void initTableInbox() throws IOException {
logger.info("正在调用初始化inbox表");
// TODO Auto-generated method stub
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
//创建表描述器
HTableDescriptor inboxTableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_INBOX));
//创建列描述器
HColumnDescriptor infoColumnDescriptor = new HColumnDescriptor("info");
//设置块缓存
infoColumnDescriptor.setBlockCacheEnabled(true);
//设置块缓存大小
infoColumnDescriptor.setBlocksize(2 * 1024 * 1024);
//设置版本确界
infoColumnDescriptor.setMinVersions(100);
infoColumnDescriptor.setMaxVersions(100);
//将列描述器添加到表描述器中
inboxTableDescriptor.addFamily(infoColumnDescriptor);
//创建表
admin.createTable(inboxTableDescriptor);
admin.close();
connection.close();
logger.info("inbox表创建成功");
}
2.5 发布微博内容
/**
* 发布微博
* 传入参数
* 参数一:uid——用户ID
* 参数二:content——微博内容
*
* a. 向ns_weibo:content表中添加数据
* · rowKey:uid_时间戳
* · columnFamily:info
* · column:content
* · Value:参数二:content——微博内容
*
* b. 将发布的内容推送到发布者的粉丝的收件箱中
* 1. 获取粉丝
* · 访问用户关系表
* · rowKey : 参数一:uid——用户ID
* · ColumnFamily : fans
* · 获取Column
*
* 2. 根据粉丝ID访问收件箱表，并添加数据
* · rowKey : 粉丝ID
* · columnFamily: info
* · column:参数一:uid
* · value: 微博rowKey
* @throws IOException
*/
public void publishContent(String uid,String content) throws IOException {
logger.info(uid + "正在调用publishContent方法发布微博");
Connection connection = ConnectionFactory.createConnection(conf);
//获得微博内容表ns_content
Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT));
//a. 向ns_weibo:content表中添加数据
//a.1 封装要添加的数据
long timeWeibo = System.currentTimeMillis();
byte[] rowKey = Bytes.toBytes(uid + "_" + timeWeibo);
Put contentPut = new Put(rowKey);
contentPut.addColumn(Bytes.toBytes("info"),Bytes.toBytes("content"),Bytes.toBytes(content));
//a.2 添加数据
contentTable.put(contentPut);
//b. 将发布的内容推送到发布者的粉丝的收件箱中
//b.1 获取用户关系表
Table relationTable = connection.getTable(TableName.valueOf(TABLE_RELATION));
//b.2从relation表中获取粉丝id
Get get = new Get(Bytes.toBytes(uid));
get.addFamily(Bytes.toBytes("fans"));
//先取出所有fans的用户id，存在集合中
List<byte[]> fans = new ArrayList();
Result result = relationTable.get(get);
Cell[] cells = result.rawCells();
for (Cell cell : cells) {
//取出当前用户所有的粉丝uid
fans.add(CellUtil.cloneQualifier(cell)); //注意：我们是将用户uid存放在column中
}
//判断是否存在粉丝,若不存在，则无需操作粉丝的收件箱表
if(fans.size() < 0){
return;
}else {
//每获取到一个粉丝uid，就去收件箱表中进行设置
//获取收件箱表
Table inboxTable = connection.getTable(TableName.valueOf(TABLE_INBOX));
//封装要添加的数据 rowKey:粉丝ID
List<Put> puts = new ArrayList<>();
for(byte[] fansRowkey : fans) {
Put inboxPut = new Put(fansRowkey);
inboxPut.addColumn(Bytes.toBytes("info"),Bytes.toBytes(uid),timeWeibo,rowKey); //注意要:添加时间戳信息,此时的时间戳是微博发布的时间
puts.add(inboxPut);
}
//向收件箱表中批量添加数据
inboxTable.put(puts);
inboxTable.close();
}
//关闭表与连接器，释放资源
contentTable.close();
relationTable.close();
connection.close();
logger.info(uid + "微博发布成功");
}
2.6 添加关注用户
/**
* 添加关注
*
* 传入参数：
* 参数一 ： 用户id
* 参数二 ： 要关注的人attends
*
* 1. 在用户关系表中,对当前的用户id进行添加关注操作
* · rowKey : 用户id
* · columnFamily : attends
* · column:要关注的人的id
* · value:空字符串
* 2. 在用户关系表中，对被关注的人进行添加粉丝操作
* · rowKey : attend
* · columnFamily:fans
* · column:粉丝id，也就是当前传入参数一的用户的id
* · value:设置为空
* 3. 在收件箱表中，添加当前用户新关注的人的微博rowKey
* @throws IOException
*/
public void addAttends(String uid,String... attends) throws IOException{
logger.info(uid + "正在调用addAttends方法添加关注");
//判断attends是否合法
if (attends == null || attends.length <= 0 || uid == null) {
return;
}
//1. 在用户关系表中,对当前的用户id进行添加关注操作
//1.1 获取用户关系表
Connection connection = ConnectionFactory.createConnection(conf);
Table relationTable = connection.getTable(TableName.valueOf(TABLE_RELATION));
//存放被关注人的粉丝（也就是当前用户）的数据
List<Put> listPutFans = new ArrayList<>();
Put putAttend = new Put(Bytes.toBytes(uid)); //因为都是对当前用户下的单元格进行操作，是同一个rowKey,因此不需要集合
for (String attend : attends) {
//1.2.1封装要添加的数据
putAttend.addColumn(Bytes.toBytes("attends"),Bytes.toBytes(attend),Bytes.toBytes(""));
//2. 在用户关系表中，对被关注的人进行添加粉丝操作
//2.1 对被关注的人进行添加粉丝操作
//2.1.1 封装要添加到relationPut表中的内容
Put putFan = new Put(Bytes.toBytes(attend));
putFan.addColumn(Bytes.toBytes("fans"),Bytes.toBytes(uid),Bytes.toBytes(""));
//2.1.2 对被关注人添加粉丝
listPutFans.add(putFan);
}
relationTable.put(putAttend);
relationTable.put(listPutFans);
//3. 在收件箱表中，添加当前用户新关注的人的微博rowKey
//3.1 获取微博内容表
Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT));
//3.2 获取微博rowKey
Scan scan = new Scan();
//用于存放关注的人的微博rowKey
List<byte[]> weiboRowKeys = new ArrayList<>();
for(String attend:attends) {
//扫描微博rowKey,使用rowKey过滤器rowFilter
//符合类似1001_xxxxx的才取出
RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,new SubstringComparator(attend + "_"));
scan.setFilter(filter);
//通过该scan扫描结果
ResultScanner resultScanner = contentTable.getScanner(scan);
//迭代器遍历
Iterator<Result> iterator = resultScanner.iterator();
while(iterator.hasNext()) {
Result result = iterator.next();
weiboRowKeys.add(result.getRow()); //获取微博rowKey
}
}
//3.3 向收件箱表中rowKey为当前用户的region中添加关注人的微博rowKey
if(weiboRowKeys.size() <= 0) return; //如果所关注的人没有发微博，则直接返回
//3.3.1 获取收件箱表
Table inboxtable = connection.getTable(TableName.valueOf(TABLE_INBOX));
//3.3.2 封装要添加的数据
Put inboxPut = new Put(Bytes.toBytes(uid));
for (byte[] rowKey : weiboRowKeys) {
String attendUID = Bytes.toString(rowKey).split("_")[0]; //获取被关注人的id
String attendTime = Bytes.toString(rowKey).split("_")[1]; //获取该微博对应的时间戳
inboxPut.addColumn(Bytes.toBytes("info"),Bytes.toBytes(attendUID),Long.parseLong(attendTime),rowKey);
}
//3.3.3 向收件箱中添加数据
inboxtable.put(inboxPut);
//关闭释放资源
inboxtable.close();
contentTable.close();
relationTable.close();
connection.close();
}
2.7 移除（取关）用户
/**
* 取关操作
*
* 传入参数
* 参数一：用户ID
* 参数二：要取关的人（多个）
*
* 1. 在用户关系表中,对当前的用户id进行取消关注操作
* rowKey:当前用户ID
* columnFamily:attends
* column:要取关人的id
* 用Delete进行数据的删除
*
* 2. 用户关系表中，删除要取关的人的region中fans为当前用户的列
* rowKey:attend（要取关的用户id）
* columnFamily:fans
* column:当前用户id
* 用Delete进行数据的删除
*
* 3. 收件箱表中，删除当前用户对应的要取关的人的微博
* rowKey:当前用户
* columnFamily:待取关的人的id
* column:微博rowKey
* @throws IOException
*/
public void removeAttends(String uid,String... attends) throws IOException {
//判断attends是否合法
if (attends == null || attends.length <= 0 || uid == null) {
return;
}
//1. 在用户关系表中,对当前的用户id进行取消关注操作
//1.1 获取用户关系表
Connection connection = ConnectionFactory.createConnection(conf);
Table relationTable = connection.getTable(TableName.valueOf(TABLE_RELATION));
//1.2 封装要取关的人数据
Delete deleteAttends = new Delete(Bytes.toBytes(uid));
//封装被取关人中当前用户信息
List<Delete> deleteFansList = new ArrayList<>();
for (String attend : attends) {
//封装取关人的信息并添加到delete对象中
deleteAttends.addColumn(Bytes.toBytes("attends"),Bytes.toBytes(attend));
//2. 用户关系表中，删除要取关的人的region中fans为当前用户的列
Delete deleteFan = new Delete(Bytes.toBytes(attend));
deleteFan.addColumn(Bytes.toBytes("fans"),Bytes.toBytes(uid));
deleteFansList.add(deleteFan);
}
//用户关系表中进行删除
relationTable.delete(deleteAttends);
relationTable.delete(deleteFansList);
//3. 收件箱表中，删除当前用户对应的要取关的人的微博
// * rowKey:当前用户
// * columnFamily:待取关的人的id
// * column:微博rowKey
//3.1 获取收件箱表
Table inboxTable = connection.getTable(TableName.valueOf(TABLE_INBOX));
//封装待删除微博信息
Delete deleteWeibos = new Delete(Bytes.toBytes(uid));
for (String attend : attends) {
deleteWeibos.addColumn(Bytes.toBytes("info"),Bytes.toBytes(attend));
}
inboxTable.delete(deleteWeibos);
//关闭释放资源
inboxTable.close();
relationTable.close();
connection.close();
}
2.8 获取关注的人的微博内容
-
首先创建Message类，用来封装用户ID；发布时间；微博内容
public class Message { private String uid; private long timeStamp; private String content; public String getUid() { return uid; } public void setUid(String uid) { this.uid = uid; } public long getTimeStamp() { return timeStamp; } public void setTimeStamp(long timeStamp) { this.timeStamp = timeStamp; } public String getContent() { return content; } public void setContent(String content) { this.content = content; } @Override public String toString() { // TODO Auto-generated method stub Date date = new Date(this.getTimeStamp()); SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); return "Message 用户ID = " + uid + "\n发布时间 = " + format.format(date) + "\n微博内容 : " + content + "\n"; } } -
实现查看微博内容
/** * 查看微博内容 * 传入参数：用户ID * * 1. 从收件箱中胡群殴所有关注者的发布微博的微博rowKey * rowKey : uid * columnFamliy : info * columnQualifier : attend * * 2. 微博内容表中查询关注人的微博内容 * rowKey : weiboRowKey * columnFamily : info * columnQualifier : content * * 3. 将数据封装为Message对象 * @throws IOException */ public List<Message> getAttendsContent(String uid) throws IOException{ logger.info(uid + "正在刷微博,调用getAttendsContent"); Connection connection = ConnectionFactory.createConnection(conf); List<Message> list = new ArrayList<>(); //1. 从收件箱中获取所有关注者的发布微博的微博rowKey //1.1 获取收件箱表 Table inboxTable = connection.getTable(TableName.valueOf(TABLE_INBOX)); //1.2 获取weiboRowKeys List<byte[]> weiboRowKeys = new ArrayList<>(); //1.2.1 封装要获取的数据 Get get_weiboRowKey = new Get(Bytes.toBytes(uid)); get_weiboRowKey.addFamily(Bytes.toBytes("info")); //只取出最新的5个版本 get_weiboRowKey.setMaxVersions(5); //此设置的意思是：获取一个rowKey下的五条数据 //新版本中 使用的是readVersions(int versions) //To limit the number of versions of each column to be returned Result result = inboxTable.get(get_weiboRowKey); Cell[] cells = result.rawCells(); //1.2.2 获取weiboRowKey for (Cell cell : cells) { byte[] rowKey = CellUtil.cloneValue(cell); weiboRowKeys.add(rowKey); } //2 . 获取微博内容并将数据封装为Message对象 //2. 微博内容表中查询关注人的微博内容 //2.1 获取微博内容表 Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT)); //批量处理 List<Get> contentGets = new ArrayList<>(); for(byte[] rowKey : weiboRowKeys) { Get contentGet = new Get(rowKey); contentGets.add(contentGet); } Result[] result_weiboContent = contentTable.get(contentGets); for (Result r : result_weiboContent) { Cell[] cs = r.rawCells(); for (Cell cell : cs) { byte[] content = CellUtil.cloneValue(cell); String weiboRowKey = Bytes.toString(r.getRow()); //封装Message对象 Message message = new Message(); String[] fields = String.valueOf(weiboRowKey).split("_"); String messageUid = fields[0]; long timeStamp = Long.parseLong(fields[1]); message.setUid(messageUid); message.setTimeStamp(timeStamp); message.setContent(Bytes.toString((content))); list.add(message); } } contentTable.close(); inboxTable.close(); connection.close(); return list; }
2.9 测试方法
/**
* 发微博
* 关注
* 取关
* 查看微博
* @throws IOException
*/
//发微博测试
public static void publishWeiBo(WeiBo weiBo,String uid,String content) throws IOException {
weiBo.publishContent(uid, content);
}
//关注测试
public static void attendTest(WeiBo weiBo, String uid,String... attends) throws IOException {
weiBo.addAttends(uid, attends);
}
//取关测试
public static void removeTest(WeiBo weibo, String uid,String... attends) throws IOException {
weibo.removeAttends(uid, attends);
}
//查看微博
public static void testGetAttensContent(WeiBo weibo, String uid) throws IOException{
List<Message> list = weibo.getAttendsContent(uid);
for (Message message : list) {
System.out.println(message);
}
}
2.10 主方法的实现
public static void main(String[] args) throws Exception {
WeiBo weiBo = new WeiBo();
weiBo.init();
publishWeiBo(weiBo,"1002","hahahhahah,1002在发微博");
publishWeiBo(weiBo,"1002","hahahhahah,1002发了第二条微博");
publishWeiBo(weiBo,"1002","hahahhahah,1002发了第三条微博");
publishWeiBo(weiBo,"1003","hahahhahah,1001发了第一条微博");
attendTest(weiBo, "1001","1002","1003");
testGetAttensContent(weiBo, "1001");
}
/**
* 在jdk1.8中输出结果为：
* ------------------------------------
* Message 用户ID = 1002
发布时间 = 2018-09-17 20:13:18
微博内容 : hahahhahah,1002发了第三条微博
Message 用户ID = 1002
发布时间 = 2018-09-17 20:13:18
微博内容 : hahahhahah,1002发了第二条微博
Message 用户ID = 1002
发布时间 = 2018-09-17 20:13:18
微博内容 : hahahhahah,1002在发微博
Message 用户ID = 1003
发布时间 = 2018-09-17 20:13:18
微博内容 : hahahhahah,1001发了第一条微博
------------------------------------
*/
三、项目总结
明确业务需求
按照业务需求建表
-
具体代码实现
-
明确类的用法
1. 表的管理 HBaseAdmin 管理表（创建/删除） HTableDescriptor 表描述器，用来创建表 HcolumnDescriptor 列描述器，用来构建列族 2. 表中数据操作 Table 用来表中数据的操作（添加数据/删除数据） Put ①用来封装待添加的数据 ②一个rowKey对应一个Put对象，可以对多个列族进行操作 Delete 用来存放待删除的数据 3. 表中数据的获取 Scan 用于设置扫描表的配置信息，默认全部扫描 ResultScanner 通过配置的扫描器，得到一个扫描表的实例扫描器 Result 每一个该类的实例化对象，都对应一个rowkey中若干数据，可直接通过Result得到rowkey Cell ①封装一个row key下所有的单元格中的数据 ②包括rowKey;columnFamily;column;value Get ①用于得到某一具体列数据 ②可通过设置版本号来设置获取yitiaorowKey对应的几个版本的数据
-
