gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #36 from chrislusf/master

Browse Source 
sync
This commit is contained in:
hilimd
2020-11-13 15:34:38 +08:00
committed by GitHub
parent a38efe2e7b a2962604ad
commit e0d5207ed9
39 changed files with 1248 additions and 750 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
other/java/client/src/main/java/seaweedfs/client/ChunkCache.java
View File

@@ -15,7 +15,6 @@ public class ChunkCache {
}
this.cache = CacheBuilder.newBuilder()
.maximumSize(maxEntries)
.weakValues()
.expireAfterAccess(1, TimeUnit.HOURS)
.build();
}

19
other/java/client/src/main/java/seaweedfs/client/SeaweedRead.java
View File

@@ -19,6 +19,7 @@ public class SeaweedRead {
private static final Logger LOG = LoggerFactory.getLogger(SeaweedRead.class);
static ChunkCache chunkCache = new ChunkCache(4);
static VolumeIdCache volumeIdCache = new VolumeIdCache(4 * 1024);
// returns bytesRead
public static long read(FilerGrpcClient filerGrpcClient, List<VisibleInterval> visibleIntervals,
@@ -30,13 +31,19 @@ public class SeaweedRead {
FilerProto.LookupVolumeRequest.Builder lookupRequest = FilerProto.LookupVolumeRequest.newBuilder();
for (ChunkView chunkView : chunkViews) {
String vid = parseVolumeId(chunkView.fileId);
lookupRequest.addVolumeIds(vid);
if (volumeIdCache.getLocations(vid)==null){
lookupRequest.addVolumeIds(vid);
}
}
FilerProto.LookupVolumeResponse lookupResponse = filerGrpcClient
.getBlockingStub().lookupVolume(lookupRequest.build());
Map<String, FilerProto.Locations> vid2Locations = lookupResponse.getLocationsMapMap();
if (lookupRequest.getVolumeIdsCount()>0){
FilerProto.LookupVolumeResponse lookupResponse = filerGrpcClient
.getBlockingStub().lookupVolume(lookupRequest.build());
Map<String, FilerProto.Locations> vid2Locations = lookupResponse.getLocationsMapMap();
for (Map.Entry<String,FilerProto.Locations> entry : vid2Locations.entrySet()) {
volumeIdCache.setLocations(entry.getKey(), entry.getValue());
}
}
//TODO parallel this
long readCount = 0;
@@ -50,7 +57,7 @@ public class SeaweedRead {
startOffset += gap;
}
FilerProto.Locations locations = vid2Locations.get(parseVolumeId(chunkView.fileId));
FilerProto.Locations locations = volumeIdCache.getLocations(parseVolumeId(chunkView.fileId));
if (locations == null || locations.getLocationsCount() == 0) {
LOG.error("failed to locate {}", chunkView.fileId);
// log here!

36
other/java/client/src/main/java/seaweedfs/client/VolumeIdCache.java Normal file
View File

@@ -0,0 +1,36 @@
package seaweedfs.client;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import java.util.concurrent.TimeUnit;
public class VolumeIdCache {
private Cache<String, FilerProto.Locations> cache = null;
public VolumeIdCache(int maxEntries) {
if (maxEntries == 0) {
return;
}
this.cache = CacheBuilder.newBuilder()
.maximumSize(maxEntries)
.expireAfterAccess(1, TimeUnit.HOURS)
.build();
}
public FilerProto.Locations getLocations(String volumeId) {
if (this.cache == null) {
return null;
}
return this.cache.getIfPresent(volumeId);
}
public void setLocations(String volumeId, FilerProto.Locations locations) {
if (this.cache == null) {
return;
}
this.cache.put(volumeId, locations);
}
}

17
other/java/client/src/main/proto/filer.proto
View File

@@ -348,3 +348,20 @@ message KvPutRequest {
message KvPutResponse {
string error = 1;
}
// path-based configurations
message FilerConf {
int32 version = 1;
message PathConf {
string location_prefix = 1;
string collection = 2;
string replication = 3;
string ttl = 4;
enum DiskType {
HDD = 0;
SSD = 1;
}
DiskType disk_type = 5;
}
repeated PathConf locations = 2;
}

32
other/java/unzip/pom.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>unzip</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>com.github.chrislusf</groupId>
<artifactId>seaweedfs-client</artifactId>
<version>1.5.3</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.github.chrislusf</groupId>
<artifactId>seaweedfs-hadoop2-client</artifactId>
<version>1.5.3</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.9.2</version>
<scope>compile</scope>
</dependency>
</dependencies>
</project>

46
other/java/unzip/src/main/java/com/example/test/Example.java Normal file
View File

@@ -0,0 +1,46 @@
package com.example.test;
import seaweed.hdfs.SeaweedInputStream;
import seaweedfs.client.FilerClient;
import seaweedfs.client.FilerGrpcClient;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
public class Example {
public static FilerClient filerClient = new FilerClient("localhost", 18888);
public static FilerGrpcClient filerGrpcClient = new FilerGrpcClient("localhost", 18888);
public static void main(String[] args) throws IOException {
// 本地模式，速度很快
parseZip("/Users/chris/tmp/test.zip");
// swfs读取，慢
SeaweedInputStream seaweedInputStream = new SeaweedInputStream(
filerGrpcClient,
new org.apache.hadoop.fs.FileSystem.Statistics(""),
"/",
filerClient.lookupEntry("/", "test.zip")
);
parseZip(seaweedInputStream);
}
public static void parseZip(String filename) throws IOException {
FileInputStream fileInputStream = new FileInputStream(filename);
parseZip(fileInputStream);
}
public static void parseZip(InputStream is) throws IOException {
ZipInputStream zin = new ZipInputStream(is);
ZipEntry ze;
while ((ze = zin.getNextEntry()) != null) {
System.out.println(ze.getName());
}
}
}