|
文章目录
1、需求2、设计表结构2.1 content表2.2 relation表2.3 receive表
3、代码设计4、测试
1、需求
需求如下：
1、发布微博内容 a. 在微博内容表中 添加一条数据（发布者） b. 在微博内容接收邮件箱表对所有粉丝用户添加数据（订阅者）
2、添加关注用户 a. 在微博用户关系表中 添加新的好友关注(attends) b. 从被关注用户角度来说， 新增粉丝用户（fans） c. 微博邮件箱表添加关注用户发布的微博内容
3、移除或者取消关注用户 a. 在微博用户关系表中 移除新的好友关注(attends) b. 从被关注用户角度来说， 删除粉丝用户(fans) c. 微博邮件箱表删除关注用户发布的微博内容
4、获取关注用户发布的微博内容 a. 从微博内容邮件表中获取该用户其关注用户的微博内容的rowkey b. 根据上面获取到的微博内容的rowkey 获取微博内容 微博展示的内容信息：
（message: 发布者ID , 时间戳 , content 完成相关表的设计，并应用java api 完成业务逻辑代码编写；）
2、设计表结构
2.1 content表
结构值RowKey用户id_时间列族info列blogvalue微博发送的内容
例子：
2.2 relation表
结构值RowKey用户id列族关注的人（attends）、粉丝（fans）列用户idvalue用户id
例子：
2.3 receive表
结构值RowKey用户id列族info列关注的人id（attends_id）valuecontent表的rowkey
例子：
3、代码设计
import com.sun.org.apache.regexp.internal.RE;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.rmi.server.UID;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.Scanner;
/**
* @author zhangtao
*/
public class WeiBoTask {
/**
* message: 发布者ID , 时间戳 ,content
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
private static class Message {
private String uid;
private String timeStamp;
private String content;
private void get() {
System.out.println("用户id：" + this.uid);
System.out.println("发表时间：" + this.timeStamp);
System.out.println("微博内容：" + this.content);
System.out.println("===============================================");
}
}
private static Connection connection = null;
private static Admin admin = null;
static {
try {
//1、获取配置信息
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.quorum", "brace");
//2、创建连接对象
connection = ConnectionFactory.createConnection(configuration);
//3、创建admin对象
admin = connection.getAdmin();
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 基础功能
*/
/**
* 定义命名空间
*/
public static void createNameSpace(String ns){
//1、创建命名空间描述器
NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(ns).build();
//2、创建命名空间
try {
admin.createNamespace(namespaceDescriptor);
System.out.println("命名空间创建成功！");
} catch (NamespaceExistException e) {
System.out.println("命名空间已经存在！");
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 判断表是否存在
*/
public static boolean isTableExist(String tableName) throws IOException {
//判断表是否存在
boolean exists = admin.tableExists(TableName.valueOf(tableName));
return exists;
}
/**
* 判断数据是否存在
*/
public static boolean isDataExit(String tableName,String rowKey,String family,String cl,String value)throws IOException {
Table table = connection.getTable(TableName.valueOf(tableName));
Get get = new Get(Bytes.toBytes(rowKey));
Result result = table.get(get);
for(Cell cell : result.rawCells()){
if(family.equals(Bytes.toString(CellUtil.cloneFamily(cell)))){
if(cl.equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
if(value.equals(Bytes.toString(CellUtil.cloneValue(cell)))){
return true;
}
}
}
}
return false;
}
/**
* 关闭
*/
public static void close() throws IOException {
if (admin != null) {
admin.close();
}
if (connection != null) {
connection.close();
}
}
/**
* 向表里面插入数据
*/
public static void putData(String tableName, String rowKey, String cf, String cn, String value) throws IOException {
//1、获取表对象
Table table = connection.getTable(TableName.valueOf(tableName));
//2、创建put对象
Put put = new Put(Bytes.toBytes(rowKey));
//3、给put对象赋值
put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn), Bytes.toBytes(value));
//4、插入数据
table.put(put);
//5、关闭表连接
table.close();
}
/**
* 获取数据
*/
public static void getData(String tableName, String rowKey, String cf, String cn) throws IOException {
//1、获取表对象
Table table = connection.getTable(TableName.valueOf(tableName));
//2、创建get对象
Get get = new Get(Bytes.toBytes(rowKey));
//3、获取数据
Result result = table.get(get);
//4、解析result并打印
for (Cell cell : result.rawCells()) {
//5、打印数据
System.out.println("CF:" + Bytes.toString(CellUtil.cloneFamily(cell)) +
"，CN:" + Bytes.toString(CellUtil.cloneQualifier(cell)) +
"，Value:" + Bytes.toString(CellUtil.cloneValue(cell)));
}
//6、关闭表链接
table.close();
}
/**
* 创建表
*/
/**
* 创建content表
* rowKey：用户id+时间戳
* 列族：info
* 列：blog
* 值：内容（String）
*/
public static void createContentTable() throws IOException {
String tableName = "content";
//1、TableName：content
if (isTableExist(tableName)) {
return;
}
//创建表
HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
//2、rowKey:用户ID+时间戳
//3、列族：info
//创建列族描述器
HColumnDescriptor hColumnDescriptor = new HColumnDescriptor("info");
//设置版本界限为1
hColumnDescriptor.setMaxVersions(1);
hColumnDescriptor.setMinVersions(1);
//添加具体列族信息
hTableDescriptor.addFamily(hColumnDescriptor);
//4、列：content
//5、值：内容（String）
//创建表
admin.createTable(hTableDescriptor);
}
/**
* 创建relation表
* rowKey：用户id
* 列族：attends、fans
* 列：用户id
* 值：用户id
*/
public static void createRelationTable() throws IOException {
String tableName = "relation";
//1、TableName：content
if (isTableExist(tableName)) {
return;
}
//创建表
HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
//2、rowKey:用户ID
//3、列族：attends、fans
//创建列族描述器
HColumnDescriptor hColumnDescriptor1 = new HColumnDescriptor("attends");
HColumnDescriptor hColumnDescriptor2 = new HColumnDescriptor("fans");
//设置版本界限为1
hColumnDescriptor1.setMaxVersions(1);
hColumnDescriptor1.setMinVersions(1);
hColumnDescriptor2.setMaxVersions(1);
hColumnDescriptor2.setMinVersions(1);
//添加具体列族信息
hTableDescriptor.addFamily(hColumnDescriptor1);
hTableDescriptor.addFamily(hColumnDescriptor2);
//4、列：用户id
//5、值：用户id
//创建表
admin.createTable(hTableDescriptor);
}
/**
* 创建receive表
* rowKey:用户id
* 列族：info
* 列：attends_id
* 值：content的rowKey
*/
public static void createReceiveTable() throws IOException {
String tableName = "receive";
//1、TableName：receive
if (isTableExist(tableName)) {
return;
}
//创建表
HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
//2、rowKey:用户ID
//3、列族：info
//创建列族描述器
HColumnDescriptor hColumnDescriptor = new HColumnDescriptor("info");
//设置版本界限为100，一次显示10000条微博
hColumnDescriptor.setMaxVersions(10000);
hColumnDescriptor.setMinVersions(10000);
//添加具体列族信息
hTableDescriptor.addFamily(hColumnDescriptor);
//4、列：attends
//5、值：微博的rowKey
//创建表
admin.createTable(hTableDescriptor);
}
/**
* 初始化
*/
public static void initialize() throws IOException {
//创建命名空间
// createNameSpace(nameSpace);
//创建内容表
createContentTable();
//创建用户关系表
createRelationTable();
//创建收件箱表
createReceiveTable();
System.out.println("初始化成功！");
}
/**
* 操作：发布微博、添加关注用户、取关用户、获取关注用户的微博内容
*/
/**
* 操作：发布微博
* a. 在微博内容表中添加一条数据（发布者）
* b. 在微博内容接收邮件箱表对所有粉丝用户添加数据（订阅者）
*/
public static void sendContent(String uId, String content) throws IOException {
//1、在content中添加数据
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String timestamp = df.format(new Date());
String rowKey = uId + "_" + timestamp;
putData("content", rowKey, "info", "blog", content);
System.out.println("用户" + uId + "成功发送了一条新微博！");
//2、在receive对fans添加数据
//获得fans的id
Table table = connection.getTable(TableName.valueOf("relation"));
Get get = new Get(Bytes.toBytes(uId));
Result result = table.get(get);
ArrayList fansIds = new ArrayList();
for (Cell cell : result.rawCells()) {
if (("fans").equals(Bytes.toString(CellUtil.cloneFamily(cell)))) {
fansIds.add(Bytes.toString(CellUtil.cloneValue(cell)));
}
}
table.close();
if (fansIds.size()
putData("receive", id, "info", uId, rowKey);
// System.out.println("用户" + id + "收到了" + uId + "刚刚发送的新微博！");
}
}
/**
* 操作：添加关注用户
* a. 在微博用户关系表中添加新的好友关注(attends)
* b. 从被关注用户角度来说，新增粉丝用户（fans）
* c. 微博邮件箱表添加关注用户发布的微博内容
*/
public static void follow(String uid, String attendId) throws IOException {
//1、粉丝：添加新的关注
if(isDataExit("relation",uid,"attends",attendId,attendId)){
System.out.println("该用户你已经关注了！");
return;
}
putData("relation", uid, "attends", attendId, attendId);
System.out.println("关注成功！");
//2、被关注者：添加新的粉丝
putData("relation", attendId, "fans", uid, uid);
//3、粉丝：获取被关注者的rowKeys
//获取被关注者的所有内容
Table table = connection.getTable(TableName.valueOf("content"));
//保存rowKeys
ArrayList rowKeys = new ArrayList();
//扫描前缀为attend_id的所有行
Scan scan = new Scan();
scan.setFilter(new PrefixFilter(Bytes.toBytes(attendId)));
ResultScanner result = table.getScanner(scan);
//添加rowKeys
for (Result r : result) {
for (Cell cell : r.rawCells()) {
rowKeys.add(Bytes.toString(CellUtil.cloneRow(cell)));
}
}
table.close();
if (rowKeys.size()
putData("receive", uid, "info", attendId, rowKey);
}
}
/**
* 操作：取关用户
* a. 在微博用户关系表中 移除新的好友关注(attends)
* b. 从被关注用户角度来说，删除粉丝用户(fans)
* c. 微博邮件箱表删除关注用户发布的微博内容
*/
public static void deleteFollow(String uid, String attendId) throws IOException {
if(!isDataExit("relation",uid,"attends",attendId,attendId)){
System.out.println("该用户你没有关注！");
return;
}
//1、粉丝：移除关注的人
Table table = connection.getTable(TableName.valueOf("relation"));
//删除对象
Delete delete = new Delete(Bytes.toBytes(uid));
delete.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attendId));
table.delete(delete);
//2、被关注者：移除粉丝
Delete delete1 = new Delete(Bytes.toBytes(attendId));
delete1.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid));
table.delete(delete1);
table.close();
//3、收件箱：删除所有内容
Table table1 = connection.getTable(TableName.valueOf("receive"));
Delete delete2 = new Delete(Bytes.toBytes(uid));
delete2.addColumn(Bytes.toBytes("info"), Bytes.toBytes(attendId));
table1.delete(delete2);
table1.close();
}
/**
* 操作：获取关注用户的微博内容
* a. 从微博内容邮件表中获取该用户其关注用户的微博内容的rowKey
* b. 根据上面获取到的微博内容的rowKey 获取微博内容
* 微博展示的内容信息：
* message: 发布者ID , 时间戳 ,content
*/
public static ArrayList getContent(String uid, String attendId) throws IOException {
//存储message
ArrayList messages = new ArrayList ();
//存储rowKeys
ArrayList rowKeys = new ArrayList ();
//1、获取微博内容的rowKeys
Table table = connection.getTable(TableName.valueOf("receive"));
Get get = new Get(Bytes.toBytes(uid));
get.setMaxVersions(1000);
get.addColumn(Bytes.toBytes("info"),Bytes.toBytes(attendId));
Result result = table.get(get);
for(Cell cell:result.rawCells()){
rowKeys.add(Bytes.toString(CellUtil.cloneValue(cell)));
}
table.close();
//2、获取微博的时间戳和内容
Table table1 = connection.getTable(TableName.valueOf("content"));
for(String rowKey :rowKeys){
//创建message对象
Message message = new Message();
//添加id
message.setUid(attendId);
//添加时间戳
message.setTimeStamp(rowKey.split("_")[1]);
//添加内容
Get get1 = new Get(Bytes.toBytes(rowKey));
get1.addColumn(Bytes.toBytes("info"),Bytes.toBytes("blog"));
Result rs = table1.get(get1);
for(Cell cell :rs.rawCells()){
message.setContent(Bytes.toString(CellUtil.cloneValue(cell)));
}
//加入到messages中
messages.add(message);
}
table1.close();
return messages;
}
}
4、测试
public static void main(String[] args) throws IOException {
initialize();
System.out.println("请输入用户名登录账号/注册账号");
Scanner in = new Scanner(System.in);
String id = in.next();
int aa = 0;
while (aa==0){
System.out.println();
System.out.println("请操作：");
System.out.println("1、发表微博");
System.out.println("2、关注用户");
System.out.println("3、取关用户");
System.out.println("4、查看关注用户的微博");
System.out.println("5、退出");
Scanner in2 = new Scanner(System.in);
int a = in2.nextInt();
if(a == 1){
System.out.println("请输入发布微博的内容");
Scanner in3 = new Scanner(System.in);
String text = in3.next();
sendContent(id,text);
}
if(a == 2){
System.out.println("请输入需要关注用户的id");
Scanner in3 = new Scanner(System.in);
String id2 = in3.next();
follow(id,id2);
}
if(a == 3){
System.out.println("请输入需要取消关注用户的id");
Scanner in3 = new Scanner(System.in);
String id2 = in3.next();
deleteFollow(id,id2);
}
if (a == 4) {
System.out.println("请输入关注用户的id");
Scanner in3 = new Scanner(System.in);
String id2 = in3.next();
ArrayList messages = new ArrayList();
messages = getContent(id,id2);
for(Message message : messages){
message.get();
}
}
if(a == 5){
aa = 1;
System.out.println("退出成功");
}
}
close();
}
|