基于HDFS与HBase的对象存储服务.zip

package com.moon.storagering.service.impl; import com.google.common.base.Strings; import com.moon.storagering.common.util.JsonUtil; import com.moon.storagering.common.util.StorageRingUtil; import com.moon.storagering.entity.ObjectListResult; import com.moon.storagering.entity.ObjectMetaData; import com.moon.storagering.entity.StorageRingObject; import com.moon.storagering.entity.StorageRingObjectSummary; import com.moon.storagering.exception.bussness.FileSystemException; import com.moon.storagering.service.IHdfsService; import com.moon.storagering.service.IStorageRingStore; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.framework.recipes.locks.InterProcessMutex; import org.apache.curator.retry.ExponentialBackoffRetry; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.filter.BinaryComparator; import org.apache.hadoop.hbase.filter.CompareFilter; import org.apache.hadoop.hbase.filter.QualifierFilter; import org.apache.hadoop.hbase.io.ByteBufferInputStream; import org.apache.hadoop.hbase.util.Bytes; import org.apache.log4j.Logger; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; import java.util.*; /** * @author Chanmoey * @date 2023年01月05日 */ public class StorageRingStoreImpl implements IStorageRingStore { private static final Logger LOGGER = Logger.getLogger(StorageRingStoreImpl.class); private Connection connection = null; private IHdfsService hdfsService; private String zkUrls; private CuratorFramework zkClient; public StorageRingStoreImpl(Connection connection, IHdfsService hdfsService, String zkUrls) { this.connection = connection; this.hdfsService = hdfsService; this.zkUrls = zkUrls; zkClient = CuratorFrameworkFactory.newClient(zkUrls, new ExponentialBackoffRetry(20, 5)); this.zkClient.start(); } @Override public void createBucketStore(String bucketName) throws IOException { // 创建目录表 HBaseService.createTable(connection, StorageRingUtil.getDirTableName(bucketName), StorageRingUtil.getDirColumnFamily(), null); // 创建文件表 HBaseService.createTable(connection, StorageRingUtil.getObjTableName(bucketName), StorageRingUtil.getObjColumnFamily(), StorageRingUtil.OBJ_REGIONS); // 将其添加到seq表 Put put = new Put(bucketName.getBytes()); put.addColumn(StorageRingUtil.BUCKET_DIR_SEQ_CF_BYTES, StorageRingUtil.BUCKET_DIR_SEQ_QUALIFIER, Bytes.toBytes(0L)); HBaseService.putRow(connection, StorageRingUtil.BUCKET_DIR_SEQ_TABLE, put); // 创建hdfs目录 hdfsService.mkDir(StorageRingUtil.FILE_STORE_ROOT + "/" + bucketName); } @Override public void deleteBucketStore(String bucketName) throws IOException { // 删除目录表和文件表 HBaseService.deleteTable(connection, StorageRingUtil.getDirTableName(bucketName)); HBaseService.deleteTable(connection, StorageRingUtil.getObjTableName(bucketName)); // 删除seq表中的记录 HBaseService.deleteRow(connection, StorageRingUtil.BUCKET_DIR_SEQ_TABLE, bucketName); // 删除hdfs上的目录 hdfsService.deleteDir(StorageRingUtil.FILE_STORE_ROOT + "/" + bucketName); } @Override public void createSeqTable() throws IOException { HBaseService.createTable(connection, StorageRingUtil.BUCKET_DIR_SEQ_TABLE, new String[]{StorageRingUtil.BUCKET_DIR_SEQ_CF}, null); } @Override public void put(String bucketName, String key, ByteBuffer content, long length, String mediaType, Map<String, String> properties) throws Exception { InterProcessMutex lock = null; // 判断是否创建目录 if (key.endsWith("/")) { putDir(bucketName, key); return; } // 获取seqid String dir = key.substring(0, key.lastIndexOf("/") + 1); String hash = null; while (hash == null) { if (!dirExist(bucketName, dir)) { hash = putDir(bucketName, dir); } else { hash = getDirSeqId(bucketName, dir); } } // 上传文件到文件表 // 获取锁 String lockKey = key.replace("/", "_"); lock = new InterProcessMutex(zkClient, "/stroagering/" + bucketName + "/" + lockKey); lock.acquire(); // 上传文件 String fileKey = hash + "_" + key.substring(key.lastIndexOf("/") + 1); Put contentPut = new Put(fileKey.getBytes()); // 记录文件属性 if (!Strings.isNullOrEmpty(mediaType)) { contentPut.addColumn(StorageRingUtil.OBJ_META_CF_BYTES, StorageRingUtil.OBJ_CONT_QUALIFIER, mediaType.getBytes()); } if (properties != null) { String prop = JsonUtil.toJson(properties); contentPut.addColumn(StorageRingUtil.OBJ_META_CF_BYTES, StorageRingUtil.OBJ_PROPS_QUALIFIER, prop.getBytes()); } contentPut.addColumn(StorageRingUtil.OBJ_META_CF_BYTES, StorageRingUtil.OBJ_LEN_QUALIFIER, Bytes.toBytes(length)); // 判断文件大小，小于20M，存储到hbase，否则存储到hdfs if (length <= StorageRingUtil.FILE_STORE_THRESHOLD) { ByteBuffer buffer = ByteBuffer.wrap(StorageRingUtil.OBJ_CONT_QUALIFIER); contentPut.addColumn(StorageRingUtil.OBJ_CONT_CF_BYTES, buffer, System.currentTimeMillis(), content); buffer.clear(); } else { String fileDir = StorageRingUtil.FILE_STORE_ROOT + "/" + bucketName + "/" + hash; String name = key.substring(key.lastIndexOf("/") + 1); InputStream inputStream = new ByteBufferInputStream(content); hdfsService.saveFile(fileDir, name, inputStream, length, (short) 1); } HBaseService.putRow(connection, StorageRingUtil.getObjTableName(bucketName), contentPut); // 释放锁 if (lock != null) { lock.release(); } } @Override public StorageRingObjectSummary getSummary(String bucketName, String key) throws IOException { // 判断是否时文件夹 if (key.endsWith("/")) { Result result = HBaseService.getRow(connection, StorageRingUtil.getDirTableName(bucketName), key); if (!result.isEmpty()) { // 读取文件夹的基础属性 return this.dirObject2Summary(result, bucketName, key); } return null; } // 获取文件 // 先获取文件的父目录 String dir = key.substring(0, key.lastIndexOf("/") + 1); String seq = getDirSeqId(bucketName, dir); // 父目录不存在，则说明文件也不存在 if (seq == null) { return null; } // 获取文件的rootKey String objectKey = seq + "_" + key.substring(key.lastIndexOf("/") + 1); Result result = HBaseService.getRow(connection, StorageRingUtil.getObjTableName(bucketName), objectKey); if (result.isEmpty()) { return null; } return this.result2ObjectSummary(result, bucketName, dir); } @Override public List<StorageRingObjectSummary> list(String bucketName, String startKey, String endKey) throws IOException { return null; } @Override public ObjectListResult listDir(String bucketName, String dir, String start, int maxCount) throws IOException { // 查询目录表 start = Strings.nullToEmpty(start); Get get = new Get(Bytes.toBytes(dir)); get.addFamily(StorageRingUtil.DIR_SUBDIR_CF_BYTES); if (!Strings.isNullOrEmpty(start)) { get.setFilter(new QualifierFilter(CompareFilter.Com