2021-12-22 迈向程序猿的第五十步
目录
1.3.6 修改HbaseUtil工具类：将打印单元格的内容进行封装
一.Hbase的API
1) 准备工作-下载maven并导入maven依赖
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>1.2.1</version>
</dependency>2) 准备工作-修改hosts文件
在window中hosts（要把所有的关于master的映射也配置上，否则无法连接）
192.168.10.101 xxx01
192.168.10.102 xxx02
192.168.10.103 xxx03
192.168.10.104 xxx041.1 namespace的DDL
1.1.1 hbase的连接服务测试API
public class ConnectionDemo {
public static void main(String[] args) throws IOException {
// 1、获取配置对象
Configuration conf = new Configuration();
// 2、配置zookeeper的参数
conf.set("hbase.zookeeper.quorum", "xxx01:2181,xxx02:2181,xxx03:2181");
// 3、获取连接对象
Connection connection = ConnectionFactory.createConnection(conf);
// 4、获取一个DDL操作的hbase客户端Admin
Admin admin = connection.getAdmin();
//5 、创建一个namespace的描述器
NamespaceDescriptor descriptor = NamespaceDescriptor.create("ns3").build();
//6. 客户端进行提交创建
admin.createNamespace(descriptor);
//7. 关闭操作
connection.close();
admin.close();
}
}1.1.2 HbaseUtil的工具类的封装
public class HbaseUtil {
private static Connection conn;
static {
try {
// 1、获取配置对象
Configuration conf = new Configuration();
// 2、配置zookeeper的参数
conf.set("hbase.zookeeper.quorum", "master:2181,slave1:2181,slave2:2181");
// 3、获取连接对象
conn = ConnectionFactory.createConnection(conf);
} catch (Exception e) {
// TODO: handle exception
}
}
/** 获取DDL操作的客户端Admin */
public static Admin getAdmin() {
Admin admin = null;
try {
admin = conn.getAdmin();
} catch (Exception e) {
e.printStackTrace();
}
return admin;
}
/** 关闭客户端Admin */
public static void closeAdmin(Admin admin) {
try {
if (admin != null) {
admin.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}1.1.3 编写好Junit执行代码模板
public class NamespaceHandler {
private Admin admin;
@Before
public void getAdmin() {
admin = HbaseUtil.getAdmin();
}
@After
public void close() {
HbaseUtil.closeAdmin(admin);
}
}1.1.4 创建namespace的API
@Test
public void createNamespace() {
try {
//获取namespace描述器对象
NamespaceDescriptor descriptor = NamespaceDescriptor.create("ns2").build();
//提交到hbase中，进行创建
admin.createNamespace(descriptor);
} catch (IOException e) {
e.printStackTrace();
}
}1.1.5 列出所有的namespace
@Test
public void listNamespace() throws IOException {
//通过客户端，获取namespace的描述器数组
NamespaceDescriptor[] nss = admin.listNamespaceDescriptors();
for (NamespaceDescriptor descriptor : nss) {
//遍历描述器进行打印名称
System.out.println(descriptor.getName());
}
}1.1.6 列出所有namespace中的表名
@Test
public void listNamespaceTable() throws IOException {
//通过客户端，获取namespace的描述器数组
NamespaceDescriptor[] nss = admin.listNamespaceDescriptors();
for (NamespaceDescriptor descriptor : nss) {
System.out.println("-------名称空间：" + descriptor.getName() + "-----------");
//通过客户端，并指定namespace，来获取相应表描述器的数组，进行遍历
HTableDescriptor[] tss = admin
.listTableDescriptorsByNamespace(descriptor.getName());
for (HTableDescriptor t : tss) {
System.out.println(t.getTableName());
}
}
}或者
@Test
public void listNamespaceTable1() throws IOException {
// 通过客户端，获取所有的TableName，进行遍历
TableName[] ts = admin.listTableNames();
for (TableName t : ts) {
System.out.println(t.getNameAsString());
}
}1.1.7 修改namespace的属性信息
@Test
public void alterNamespace() throws IOException {
//获取指定namespace的描述器
NamespaceDescriptor ns2 = admin.getNamespaceDescriptor("ns2");
//设置属性
ns2.setConfiguration("author", "michael");
//进行提交修改
admin.modifyNamespace(ns2);
}1.1.8 删除namespace
@Test
public void deleteNamespace() throws IOException {
//删除指定的namespace
admin.deleteNamespace("ns2");
}1.2 table的DDL
1.2.1 编写Junit执行代码模板
public class Tablehandler {
private Admin admin;
@Before
public void getAdmin() {
admin = HbaseUtil.getAdmin();
}
@After
public void close() {
HbaseUtil.closeAdmin(admin);
}
}1.2.2 创建一个table
@Test
public void createTable() throws IOException {
//获取TableName对象，指定要创建的表名
TableName tableName = TableName.valueOf("ns1:students");
//新建一个表描述器
HTableDescriptor descriptor = new HTableDescriptor(tableName);
//新建一个列族描述器
HColumnDescriptor columnDescriptor = new HColumnDescriptor("f1");
//设置列族的布隆过滤器为列类型
columnDescriptor.setBloomFilterType(BloomType.ROWCOL);
//设置列族支持缓存
columnDescriptor.setInMemory(true);
//设置列族内的单元格支持3个版本
columnDescriptor.setVersions(1, 3);
//设置列族的单元格过期时间为1天
columnDescriptor.setTimeToLive(24*60*60);
//将列族描述器添加到表描述器中, 可以添加多个列族描述器
descriptor.addFamily(columnDescriptor);
//提交到Hbase，进行创建
admin.createTable(descriptor);
}1.2.3 修改table的列族属性信息
@Test
public void alterTable2() throws IOException {
//指定表名
TableName tableName = TableName.valueOf("ns1:students");
//获取表描述器
HTableDescriptor tableDescriptor = admin.getTableDescriptor(tableName);
//获取指定的列族的描述器
HColumnDescriptor columnDescriptor =tableDescriptor.getFamily(Bytes.toBytes("f1"));
//修改属性
columnDescriptor.setBloomFilterType(BloomType.ROWCOL);
columnDescriptor.setVersions(1, 2);
columnDescriptor.setTimeToLive(24*60*60);
//提交到Hbase
admin.modifyColumn(tableName, columnDescriptor);
}1.2.4 向table中添加新的列族
@Test
public void alterTable2() throws IOException {
TableName tableName = TableName.valueOf("ns1:students");
HColumnDescriptor columnDescriptor = new HColumnDescriptor("f2");
columnDescriptor.setBloomFilterType(BloomType.ROWCOL);
columnDescriptor.setVersions(1, 2);
columnDescriptor.setTimeToLive(24*60*60); // 秒为单位
admin.addColumn(tableName, columnDescriptor);
}1.2.5 获取table的描述信息
@Test
public void describeTable() throws IOException, Exception {
TableName tableName = TableName.valueOf("ns1:students");
HTableDescriptor descriptor =admin.getTableDescriptor(tableName);
HColumnDescriptor[] hcs = descriptor.getColumnFamilies();
for(HColumnDescriptor hc:hcs) {
System.out.print("NAME=>"+hc.getNameAsString()+"t");
System.out.print("BLOOMFILTER=>"+hc.getBloomFilterType()+"t");
System.out.print("VERSIONS=>"+hc.getMaxVersions()+"t");
System.out.print("TTL=>"+hc.getTimeToLive()+"t");
System.out.println("BLOCKSIZE=>"+hc.getBlocksize());
}
}1.2.6 删除table
@Test
public void dropTable() throws TableNotFoundException, IOException {
TableName tableName = TableName.valueOf("ns1:students");
if (admin.tableExists(tableName)) {
if (!admin.isTableDisabled(tableName)) {
admin.disableTable(tableName);
}
admin.deleteTable(tableName);
}
}1.3 table的CRUD
1.3.1 修改HbaseUtil工具类型
public class HbaseUtil {
private static Connection conn;
static {
try {
Configuration conf = new Configuration();
conf.set("hbase.zookeeper.quorum", "master:2181,slave1:2181,slave2:2181");
conn = ConnectionFactory.createConnection(conf);
} catch (Exception e) {
// TODO: handle exception
}
}
public static Admin getAdmin() {
// 略
}
public static void closeAdmin(Admin admin) {
// 略
}
/** 获取Table对象的方法*/
public static Table getTable(String name) {
Table table = null;
try {
// 将表名转成TableName对象
TableName tableName = TableName.valueOf(name);
// 通过连接对象，获取Table对象
table = conn.getTable(tableName);
} catch (Exception e) {
e.printStackTrace();
}
return table;
}
/** 关闭表对象 */
public static void closeTable(Table table) {
try {
if (table != null) {
table.close();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}1.3.2 编写Junit执行代码模板
public class TableCRUDandler {
private Table table;
@Before
public void getAdmin() {
table = HbaseUtil.getTable("ns1:students");
}
@After
public void close() {
HbaseUtil.closeTable(table);
}
}1.3.3 向table中添加数据：put
@Test
public void putData() throws IOException {
//1. 获取Put对象,指定rowkey
Put put = new Put(Bytes.toBytes("rk00001"));
//2. 指定列族名，列名，列值
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"), Bytes.toBytes("zhangfei"));
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"), Bytes.toBytes("15"));
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("gender"), Bytes.toBytes("m"));
//3. 提交
table.put(put);
}1.3.4 批量增加数据：
@Test
public void putData2() throws IOException {
List<Put> list = new ArrayList<Put>();
Put put1 = new Put(Bytes.toBytes("rk00002"));
put1.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"), Bytes.toBytes("zhenji"));
put1.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"), Bytes.toBytes("23"));
put1.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("gender"), Bytes.toBytes("f"));
Put put2 = new Put(Bytes.toBytes("rk00003"));
put2.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"), Bytes.toBytes("zhaoyun"));
put2.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"), Bytes.toBytes("25"));
put2.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("gender"), Bytes.toBytes("m"));
Put put3 = new Put(Bytes.toBytes("rk00004"));
put3.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"), Bytes.toBytes("liubei"));
put3.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"), Bytes.toBytes("30"));
put3.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("gender"), Bytes.toBytes("m"));
list.add(put1);
list.add(put2);
list.add(put3);
table.put(list);
}1.3.5 获取指定rowkey的所有列族信息
@Test
public void getData() throws IOException {
//指定rowkey，获取get对象
Get get = new Get(Bytes.toBytes("rk00001"));
//调用get方法，获取结果集
Result result = table.get(get);
//获取结果集的扫描器，即迭代器
CellScanner sc = result.cellScanner();
while (sc.advance()) {
//取出当前单元格对象
Cell cell = sc.current();
//使用克隆的方式，打印出来
System.out.print(new String(CellUtil.cloneRow(cell))+"t");
System.out.print(new String(CellUtil.cloneFamily(cell), "utf8") + "t");
System.out.print(new String(CellUtil.cloneQualifier(cell), "utf8") + "t");
System.out.print(new String(CellUtil.cloneValue(cell), "utf8")+ "t");
System.out.println(CellUtil.getCellKeyAsString(cell));
}
}1.3.6 修改HbaseUtil工具类：将打印单元格的内容进行封装
public static void printResult(Result rs) throws IOException {
CellScanner sc = rs.cellScanner();
while (sc.advance()) {
Cell cell = sc.current();
System.out.print(new String(CellUtil.cloneRow(cell)) + "t");
System.out.print(new String(CellUtil.cloneFamily(cell), "utf8") + "t");
System.out.print(new String(CellUtil.cloneQualifier(cell), "utf8") + "t");
System.out.println(new String(CellUtil.cloneValue(cell), "utf8") + "t");
}
}1.3.7 获取多行的所有列族信息
@Test
public void MultigetData() throws IOException {
List<Get> gets = new ArrayList<Get>();
Get get1 = new Get(Bytes.toBytes("rk00001"));
Get get2 = new Get(Bytes.toBytes("rk00002"));
Get get3 = new Get(Bytes.toBytes("rk00003"));
Get get4 = new Get(Bytes.toBytes("rk00004"));
gets.add(get1);
gets.add(get2);
gets.add(get3);
gets.add(get4);
//获取多行的结果集
Result[] result = table.get(gets);
for (int i = 0; i < result.length; i++) {
HbaseUtil.printResult(result[i]);
}
}1.3.8 使用scan对表进行扫描
@Test
public void scanData() throws Exception {
//获取Scan对象，指定扫描范围：开始行号和结束行号。小技巧00
Scan scan = new Scan(Bytes.toBytes("rk00001"), Bytes.toBytes("rk00003" + "01"));
ResultScanner rScanner = table.getScanner(scan);
Iterator<Result> iterable = rScanner.iterator();
while (iterable.hasNext()) {
Result result = iterable.next();
HbaseUtil.printResult(result);
}
}1.3.9 删除指定行
@Test
public void deleteData() throws Exception {
Delete delete = new Delete(Bytes.toBytes("rk00001"));
table.delete(delete);
}1.3.10 删除指定单元格
@Test
public void deleteDataC() throws Exception {
Delete delete = new Delete(Bytes.toBytes("rk000003"));
delete.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"));
table.delete(delete);
}1.3.11 批量删除
@Test
public void MultideleteData() throws Exception {
List<Delete> list = new ArrayList<Delete>();
Delete delete1 = new Delete(Bytes.toBytes("rk00001"));
Delete delete2 = new Delete(Bytes.toBytes("rk000010"));
Delete delete3 = new Delete(Bytes.toBytes("rk000011"));
Delete delete4 = new Delete(Bytes.toBytes("rk000012"));
list.add(delete1);
list.add(delete2);
list.add(delete3);
list.add(delete4);
table.delete(list);
}1.4 Hbase的Filter
1.4.1 单列过滤器：SingleColumnValueFilter
1.4.2 结构过滤器：FilterList
1.4.3 列值比较器: 四个
正则表达式(可以代替下面三个)：
子串：
二进制：
二进制前缀：
1.4.4 KeyValue元数据过滤器
--列族过滤器
--列名过滤器 select
--列名前缀过滤器
--多列名前缀
--列名范围过滤器
1.4.5 RowKey过滤器
RowFilter
1.4.6 Utility过滤器
FirstKeyOnlyFilter
1.4.7 分页过滤器
PageFilter1.4.1 结构过滤器：FilterList
1) 代码
@Test
public void TestFilter() throws Exception {
// 定义一个结构过滤器，指定满足所有条件 MUST_PASS_ALL/ MUST_PASS_ONE
FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL);
// 使用单列过滤器进行设置条件, 比较关系：CompareOp.EQUAL ,CompareOp.GREATER_OR_EQUAL
SingleColumnValueFilter nameFilter = new SingleColumnValueFilter(
Bytes.toBytes("f1"),
Bytes.toBytes("name"),CompareOp.EQUAL,Bytes.toBytes("zhaoyun"));
SingleColumnValueFilter ageFilter = new SingleColumnValueFilter(
Bytes.toBytes("f1"), Bytes.toBytes("age"),CompareOp.EQUAL, Bytes.toBytes("25"));
/* 设置不满组条件的过滤掉，因为：如果某一行中，没有指定的字段，表示满足条件。
可以设置缺失字段，进行过滤*/
nameFilter.setFilterIfMissing(true);
ageFilter.setFilterIfMissing(true);
// 将单列过滤器添加到结构过滤器上
list.addFilter(nameFilter);
list.addFilter(ageFilter);
// 获取浏览对象，
Scan scan = new Scan();
// 添加过滤条件
scan.setFilter(list);
ResultScanner rScanner = table.getScanner(scan);
Iterator<Result> it = rScanner.iterator();
while (it.hasNext()) {
Result result = it.next();
HbaseUtil.printResult(result);
}
}2) 说明：
- 遍历到某一行键时，如果包含了两个列，age和name，就会对比值，不满足条件就将其过滤，满足条件就获取。
- 如果包含其中一个列，name或者age，只要其中一个列满足条件就会，就会获取。
- 如果name和age都没有，那么视为该行满足条件3) 解决方案：
在某个单值过滤器总添加
ageFilter.setFilterIfMissing(true);
nameFilter.setFilterIfMissing(true);1.4.2 单列过滤器
1) 先修改HbaseUtil工具类，将扫描结果集对象的打印行为，封装到HbaseUtil工具类中
public static void printResultScanner(ResultScanner rScanner) throws IOException {
Iterator<Result> iterator = rScanner.iterator();
while (iterator.hasNext()) {
Result rs = iterator.next();
HbaseUtil.printResult(rs);
}
}2) SingleColumnValueFilter
@Test
public void testSingleColumnValueFilter() throws IOException {
SingleColumnValueFilter filter = new
SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("age"),
CompareFilter.CompareOp.GREATER_OR_EQUAL,
Bytes.toBytes("16"));
filter.setFilterIfMissing(true);
testScan(filter); //调用扫描行为
}
// 将扫描行为再次封装。
private void testScan(Filter filter) throws IOException {
Scan scan = new Scan();
scan.setFilter(filter);
ResultScanner rScanner = table.getScanner(scan);
HbaseUtil.printResultScanner(rScanner);
}1.4.3 列值比较器
1) RegexStringComparator
@Test
public void testRegex() throws IOException {
//正则表达式比较器
RegexStringComparator comp = new RegexStringComparator("^zhao");
SingleColumnValueFilter nameFilter = new
SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("name"),
CompareFilter.CompareOp.EQUAL, comp);
nameFilter.setFilterIfMissing(true);
testScan(nameFilter);
}2) SubstringComparator
@Test
public void testSubstringComparator() throws IOException {
//子串比较器， 在构造器里给一个字符串，如果这个字符串是另外一个字符串的子串，就满足条件
SubstringComparator comp = new SubstringComparator("zh");
SingleColumnValueFilter nameFilter = new
SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("name"),
CompareFilter.CompareOp.EQUAL, comp);
nameFilter.setFilterIfMissing(true);
testScan(nameFilter);
}3) BinaryPrefixComparator
@Test
public void testBinaryPrefixComparator() throws IOException {
//二进制前缀比较器
BinaryPrefixComparator comp = new BinaryPrefixComparator(Bytes.toBytes("zh"));
SingleColumnValueFilter nameFilter = new
SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("name"),
CompareFilter.CompareOp.EQUAL, comp);
nameFilter.setFilterIfMissing(true);
testScan(nameFilter);
}4) BinaryComparator
@Test
public void testBinaryComparator() throws IOException {
//二进制比较器，
BinaryComparator comp = new BinaryComparator(Bytes.toBytes("xiaoniu"));
//如果使用二进制比较器，匹配的是字段的整个值的二进制进行比较，不是模糊匹配。是全匹配
SingleColumnValueFilter nameFilter = new
SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("name"),
CompareFilter.CompareOp.EQUAL, comp);
nameFilter.setFilterIfMissing(true);
testScan(nameFilter);
}1.4.4 KeyValue元数据过滤器
1) FamilyFilter
/**
* 列族过滤器，查询指定列族下的所有单元格
* @throws Exception
*/
@Test
public void testFamily() throws Exception {
// RegexStringComparator comp = new RegexStringComparator("f1");
// SubstringComparator comp = new SubstringComparator("f1");
// BinaryPrefixComparator comp = new BinaryPrefixComparator(Bytes.toBytes("f1"));
BinaryComparator comp = new BinaryComparator(Bytes.toBytes("f2"));
FamilyFilter filter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, comp);
testScan(filter);
}2) QualifierFilter
/**
* QualifierFilter过滤器是用来查询符合条件的单元格数据
* @throws IOException
*/
@Test
public void testQualifierFilter1() throws IOException {
SubstringComparator substringComparator = new SubstringComparator("na");
QualifierFilter qualifierFilter = new QualifierFilter(
CompareFilter.CompareOp.EQUAL, substringComparator);
testScan(qualifierFilter);
}3) ColumnPrefixFilter
/**
* 列名前缀过滤器
* @throws IOException
*/
@Test
public void testColumnPrefixFilter() throws IOException {
// 1. 创建ColumnPrefixFilter
ColumnPrefixFilter columnPrefixFilter = new ColumnPrefixFilter(Bytes.toBytes("a"));
// 2. 打印
testScan(columnPrefixFilter);
}4) MultipleColumnPrefixFilter
@Test
public void testMultipleColumnPrefixFilter() throws IOException {
// 1. 创建ColumnPrefixFilter
byte[][] prefixes = new byte[][] { Bytes.toBytes("a"), Bytes.toBytes("n") };
MultipleColumnPrefixFilter multipleColumnPrefixFilter = new
MultipleColumnPrefixFilter(prefixes);
testScan(multipleColumnPrefixFilter);
}5) ColumnRangeFilter
/**
* 查找以"age"到"name"的列的信息
* minColumnInclusive:true为包含，false为不包含
*/
@Test
public void testColumnRangeFilter() throws IOException {
// 1. 创建ColumnPrefixFilter
ColumnRangeFilter columnRangeFilter = new ColumnRangeFilter(
Bytes.toBytes("age"), true, Bytes.toBytes("name"),false);
// 2. 打印
testScan(columnRangeFilter);
}1.4.5 RowKey过滤器
1）RowFilter
@Test
public void testRowFilter() throws IOException {
// 1. 创建RowFilter
BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("rk00002"));
RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryComparator);
testScan(rowFilter);
}1.4.6 Utility过滤器
1) FirstKeyOnlyFilter
/**
* FirstKeyOnlyFilter
* 功能：查询行内的所有单元格进行排序的第一个单元格的数据
* @throws IOException
*/
@Test
public void testFirstKeyOnlyFilter() throws IOException {
FirstKeyOnlyFilter firstKeyOnlyFilter = new FirstKeyOnlyFilter();
// 2. 打印
testScan(firstKeyOnlyFilter);
}1.4.7 分页过滤器
1) PageFilter
/**
* 需求：每页显示3条记录
* 将全部的数据分页显示出来
* 思路：
* 1. 第一页：
* select * from user_info where rowkey > 01 limit 3;
* 2. 其他页
* select * from user_info where rowkey > 前一页的maxrowkey limit 3;
* 3. 循环什么时候结束？
* while(true) {
* select * from user_info where rowkey > 前一页的maxrowkey limit 3;
* print 3行数据
* 结束条件：count<3
* }
*/
@Test
public void testPageFilter() throws Exception {
PageFilter filter = new PageFilter(3);
Scan scan = new Scan();
scan.setFilter(filter);
String maxkey = "";
while (true) {
int count = 0;
//每三页一查询
ResultScanner rScanner = table.getScanner(scan);
Iterator<Result> iterator = rScanner.iterator();
while (iterator.hasNext()) {
Result result = iterator.next();
count++;//统计打印次数
maxkey = Bytes.toString(result.getRow());
HbaseUtil.printResult(result);
}
System.out.println("------------------------------");
if (count < 3) {
break;//如果不足三页，结束循环
}
//为下一页的查询做准备，下一页的开始rowkey要大于当前页的最后一行的rowkey
scan.setStartRow(Bytes.toBytes(maxkey + "01"));
}
}(=-=,插入数据是真的麻烦,随随便便一个数据库过来API插入得要半小时,累死~~~)
本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。
THE END
二维码