wyl/fastdfs-client-java
1
0
Fork 0
mirror of https://github.com/happyfish100/fastdfs-client-java.git synced 2026-09-17 09:20:46 +00:00
Code Issues Projects Releases Wiki Activity

connect to storage server failover with multi IPs

Browse Source
This commit is contained in:
YuQing
2023-12-13 10:06:39 +08:00
parent 020857dbf1
commit bb203bdc71
12 changed files with 404 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -15,3 +15,4 @@ target
*.conf
*.PNG
*.class
*.swp
HISTORY
+2 -1
View File
@@ -1,7 +1,8 @@
Version 1.31 2023-12-07
Version 1.31 2023-12-13
* adapt to FastDFS server V6.11 for IPv6
you must upgrade your FastDFS server to V6.11 or higher version
* connect to storage server failover with multi IPs
Version 1.30 2023-01-29
* support tracker server fail over
fdfs_client.conf
+6 -2
View File
@@ -16,9 +16,13 @@ tracker_server = 10.0.11.247:22122
tracker_server = 10.0.11.248:22122
tracker_server = 10.0.11.249:22122
# connect which ip address first for multi IPs of a storage server, value list:
## tracker: connect to the ip address return by tracker server first
## last-connected: connect to the ip address last connected first
# default value is tracker
connect_first_by = tracker
connection_pool.enabled = true
connection_pool.max_count_per_entry = 500
connection_pool.max_idle_time = 3600
connection_pool.max_wait_time_in_ms = 1000
server_ipv6.enabled = false
src/main/java/org/csource/fastdfs/ClientGlobal.java
+96 -3
View File
@@ -42,7 +42,7 @@ public class ClientGlobal {
public static final String PROP_KEY_HTTP_SECRET_KEY = "fastdfs.http_secret_key";
public static final String PROP_KEY_HTTP_TRACKER_HTTP_PORT = "fastdfs.http_tracker_http_port";
public static final String PROP_KEY_TRACKER_SERVERS = "fastdfs.tracker_servers";
public static final String PROP_KEY_CONNECT_FIRST_BY = "fastdfs.connect_first_by";
public static final String PROP_KEY_CONNECTION_POOL_ENABLED = "fastdfs.connection_pool.enabled";
public static final String PROP_KEY_CONNECTION_POOL_MAX_COUNT_PER_ENTRY = "fastdfs.connection_pool.max_count_per_entry";
@@ -56,6 +56,9 @@ public class ClientGlobal {
public static final String DEFAULT_HTTP_SECRET_KEY = "FastDFS1234567890";
public static final int DEFAULT_HTTP_TRACKER_HTTP_PORT = 80;
public static final int CONNECT_FIRST_BY_TRACKER = 0;
public static final int CONNECT_FIRST_BY_LAST_CONNECTED = 1;
public static final boolean DEFAULT_CONNECTION_POOL_ENABLED = true;
public static final int DEFAULT_CONNECTION_POOL_MAX_COUNT_PER_ENTRY = 100;
public static final int DEFAULT_CONNECTION_POOL_MAX_IDLE_TIME = 3600 ;//second
@@ -67,6 +70,9 @@ public class ClientGlobal {
public static boolean g_anti_steal_token = DEFAULT_HTTP_ANTI_STEAL_TOKEN; //if anti-steal token
public static String g_secret_key = DEFAULT_HTTP_SECRET_KEY; //generage token secret key
public static int g_tracker_http_port = DEFAULT_HTTP_TRACKER_HTTP_PORT;
public static int g_connect_first_by = CONNECT_FIRST_BY_TRACKER;
public static boolean g_multi_storage_ips = false;
public static StorageAddressMap g_storages_address_map;
public static boolean g_connection_pool_enabled = DEFAULT_CONNECTION_POOL_ENABLED;
public static int g_connection_pool_max_count_per_entry = DEFAULT_CONNECTION_POOL_MAX_COUNT_PER_ENTRY;
@@ -78,6 +84,76 @@ public class ClientGlobal {
private ClientGlobal() {
}
private static void loadStoragsFromTracker() throws IOException, MyException {
TrackerClient tracker = new TrackerClient();
StringBuilder builder = tracker.fetchStorageIds();
if (builder.length() == 0) {
return;
}
System.out.println(builder.toString());
boolean without_port = true;
int count = 0;
String[] lines = builder.toString().split("\n");
String[] ipAddresses = new String[lines.length];
for (String line : lines) {
String[] cols = line.split(" ");
if (cols.length != 3) {
throw new MyException("invalid line: " + line);
}
String ipAddrs = cols[2];
if (ipAddrs.indexOf(',') > 0) {
ipAddresses[count++] = ipAddrs;
}
}
if (count == 0) {
return;
}
int startIndex;
if (ipAddresses[0].charAt(0) == '[') { //IPv6
if ((startIndex=ipAddresses[0].indexOf(']')) < 0) {
throw new MyException("invalid IPv6 address: " + ipAddresses[0]);
}
} else {
startIndex = 0;
}
if (ipAddresses[0].indexOf(':', startIndex) > 0) {
without_port = false;
}
g_multi_storage_ips = true;
g_storages_address_map = new StorageAddressMap(without_port);
if (without_port) {
for (String ipAddr: ipAddresses) {
if (ipAddr.charAt(0) == '[') { //IPv6
ipAddr = ipAddr.substring(1, ipAddr.length() - 1);
}
String[] cols = ipAddr.split(",");
g_storages_address_map.puts(cols[0], cols[1]);
}
} else {
for (String ipPort: ipAddresses) {
int colonIndex = ipPort.lastIndexOf(':');
if (colonIndex < 0) {
throw new MyException("invalid ip and port: " + ipPort);
}
String ipAddr = ipPort.substring(0, colonIndex);
int port = Integer.parseInt(ipPort.substring(colonIndex + 1));
if (ipAddr.charAt(0) == '[') { //IPv6
ipAddr = ipAddr.substring(1, ipAddr.length() - 1);
}
String[] cols = ipAddr.split(",");
g_storages_address_map.puts(cols[0], cols[1], port);
}
}
}
/**
* load global variables
*
@@ -114,11 +190,11 @@ public class ClientGlobal {
InetSocketAddress[] tracker_servers = new InetSocketAddress[szTrackerServers.length];
for (int i = 0; i < szTrackerServers.length; i++) {
if(szTrackerServers[i].contains("[")){
if (szTrackerServers[i].contains("[")) {
parts = new String[2];
parts[0] = szTrackerServers[i].substring(1, szTrackerServers[i].indexOf("]"));
parts[1] = szTrackerServers[i].substring(szTrackerServers[i].lastIndexOf(":") + 1);
}else {
} else {
parts = szTrackerServers[i].split("\\:", 2);
}
@@ -130,6 +206,11 @@ public class ClientGlobal {
}
g_tracker_group = new TrackerGroup(tracker_servers);
String connect_first_by = iniReader.getStrValue("connect_first_by");
if (connect_first_by != null && connect_first_by.equalsIgnoreCase("last-connected")) {
g_connect_first_by = CONNECT_FIRST_BY_LAST_CONNECTED;
}
g_tracker_http_port = iniReader.getIntValue("http.tracker_http_port", 80);
g_anti_steal_token = iniReader.getBoolValue("http.anti_steal_token", false);
if (g_anti_steal_token) {
@@ -146,6 +227,8 @@ public class ClientGlobal {
if (g_connection_pool_max_wait_time_in_ms < 0) {
g_connection_pool_max_wait_time_in_ms = DEFAULT_CONNECTION_POOL_MAX_WAIT_TIME_IN_MS;
}
loadStoragsFromTracker();
}
/**
@@ -183,6 +266,8 @@ public class ClientGlobal {
String httpAntiStealTokenConf = props.getProperty(PROP_KEY_HTTP_ANTI_STEAL_TOKEN);
String httpSecretKeyConf = props.getProperty(PROP_KEY_HTTP_SECRET_KEY);
String httpTrackerHttpPortConf = props.getProperty(PROP_KEY_HTTP_TRACKER_HTTP_PORT);
String connectFirstBy = props.getProperty(PROP_KEY_CONNECT_FIRST_BY);
String poolEnabled = props.getProperty(PROP_KEY_CONNECTION_POOL_ENABLED);
String poolMaxCountPerEntry = props.getProperty(PROP_KEY_CONNECTION_POOL_MAX_COUNT_PER_ENTRY);
String poolMaxIdleTime = props.getProperty(PROP_KEY_CONNECTION_POOL_MAX_IDLE_TIME);
@@ -205,6 +290,10 @@ public class ClientGlobal {
if (httpTrackerHttpPortConf != null && httpTrackerHttpPortConf.trim().length() != 0) {
g_tracker_http_port = Integer.parseInt(httpTrackerHttpPortConf);
}
if (connectFirstBy != null && connectFirstBy.equalsIgnoreCase("last-connected")) {
g_connect_first_by = CONNECT_FIRST_BY_LAST_CONNECTED;
}
if (poolEnabled != null && poolEnabled.trim().length() != 0) {
g_connection_pool_enabled = Boolean.parseBoolean(poolEnabled);
}
@@ -217,6 +306,8 @@ public class ClientGlobal {
if (poolMaxWaitTimeInMS != null && poolMaxWaitTimeInMS.trim().length() != 0) {
g_connection_pool_max_wait_time_in_ms = Integer.parseInt(poolMaxWaitTimeInMS);
}
loadStoragsFromTracker();
}
/**
@@ -360,6 +451,8 @@ public class ClientGlobal {
+ "\n g_anti_steal_token = " + g_anti_steal_token
+ "\n g_secret_key = " + g_secret_key
+ "\n g_tracker_http_port = " + g_tracker_http_port
+ "\n g_multi_storage_ips = " + g_multi_storage_ips
+ "\n g_connect_first_by = " + (g_connect_first_by == CONNECT_FIRST_BY_TRACKER ? "tracker" : "last-connected")
+ "\n g_connection_pool_enabled = " + g_connection_pool_enabled
+ "\n g_connection_pool_max_count_per_entry = " + g_connection_pool_max_count_per_entry
+ "\n g_connection_pool_max_idle_time(ms) = " + g_connection_pool_max_idle_time
src/main/java/org/csource/fastdfs/ProtoCommon.java
+29 -12
View File
@@ -26,6 +26,7 @@ import java.util.Arrays;
*/
public class ProtoCommon {
public static final byte FDFS_PROTO_CMD_QUIT = 82;
public static final byte TRACKER_PROTO_CMD_FETCH_STORAGE_IDS = 69;
public static final byte TRACKER_PROTO_CMD_SERVER_LIST_GROUP = 91;
public static final byte TRACKER_PROTO_CMD_SERVER_LIST_STORAGE = 92;
public static final byte TRACKER_PROTO_CMD_SERVER_DELETE_STORAGE = 93;
@@ -310,6 +311,23 @@ public class ProtoCommon {
return headerInfo.errno == 0 ? true : false;
}
/**
* int convert to buff (big-endian)
*
* @param n int number
* @return 4 bytes buff
*/
public static byte[] int2buff(int n) {
byte[] bs;
bs = new byte[4];
bs[0] = (byte) ((n >> 24) & 0xFF);
bs[1] = (byte) ((n >> 16) & 0xFF);
bs[2] = (byte) ((n >> 8) & 0xFF);
bs[3] = (byte) (n & 0xFF);
return bs;
}
/**
* long convert to buff (big-endian)
*
@@ -317,19 +335,18 @@ public class ProtoCommon {
* @return 8 bytes buff
*/
public static byte[] long2buff(long n) {
byte[] bs;
byte[] bs;
bs = new byte[8];
bs[0] = (byte) ((n >> 56) & 0xFF);
bs[1] = (byte) ((n >> 48) & 0xFF);
bs[2] = (byte) ((n >> 40) & 0xFF);
bs[3] = (byte) ((n >> 32) & 0xFF);
bs[4] = (byte) ((n >> 24) & 0xFF);
bs[5] = (byte) ((n >> 16) & 0xFF);
bs[6] = (byte) ((n >> 8) & 0xFF);
bs[7] = (byte) (n & 0xFF);
return bs;
bs = new byte[8];
bs[0] = (byte) ((n >> 56) & 0xFF);
bs[1] = (byte) ((n >> 48) & 0xFF);
bs[2] = (byte) ((n >> 40) & 0xFF);
bs[3] = (byte) ((n >> 32) & 0xFF);
bs[4] = (byte) ((n >> 24) & 0xFF);
bs[5] = (byte) ((n >> 16) & 0xFF);
bs[6] = (byte) ((n >> 8) & 0xFF);
bs[7] = (byte) (n & 0xFF);
return bs;
}
/**
src/main/java/org/csource/fastdfs/StorageAddressMap.java
+62
View File
@@ -0,0 +1,62 @@
/**
* Copyright (C) 2023 Happy Fish / YuQing
* <p>
* FastDFS Java Client may be copied only under the terms of the GNU Lesser
* General Public License (LGPL).
* Please visit the FastDFS Home Page https://github.com/happyfish100/fastdfs for more detail.
*/
package org.csource.fastdfs;
import java.util.HashMap;
import java.net.InetSocketAddress;
/**
* Storage Server Address Map
*
* @author Happy Fish / YuQing
* @version Version 1.31
*/
public class StorageAddressMap {
protected boolean without_port;
protected HashMap<String, InetSocketAddress> storages;
public StorageAddressMap(boolean without_port) {
this.without_port = without_port;
this.storages = new HashMap<String, InetSocketAddress>();
}
protected String getKey(String ipAddr, int port) {
return ipAddr + "@" + port;
}
public void puts(String srcIpAddr, String destIpAddr, int port) {
storages.put(this.getKey(srcIpAddr, port),
new InetSocketAddress(destIpAddr, port));
storages.put(this.getKey(destIpAddr, port),
new InetSocketAddress(srcIpAddr, port));
}
public void puts(String srcIpAddr, String destIpAddr) {
storages.put(srcIpAddr, new InetSocketAddress(destIpAddr, 0));
storages.put(destIpAddr, new InetSocketAddress(srcIpAddr, 0));
}
public InetSocketAddress get(String ipAddr, int port) {
if (this.without_port) {
InetSocketAddress sockAddr;
if ((sockAddr=storages.get(ipAddr)) == null) {
return null;
}
return new InetSocketAddress(sockAddr.getAddress(), port);
} else {
return storages.get(this.getKey(ipAddr, port));
}
}
public InetSocketAddress get(InetSocketAddress sockAddr) {
return this.get(sockAddr.getAddress().getHostAddress(), sockAddr.getPort());
}
}
src/main/java/org/csource/fastdfs/StorageClient.java
-1
View File
@@ -847,7 +847,6 @@ public class StorageClient {
throw ex;
} finally {
releaseConnection(connection, bNewStorageServer);
}
}
src/main/java/org/csource/fastdfs/StorageServer.java
+107 -32
View File
@@ -9,6 +9,9 @@
package org.csource.fastdfs;
import java.io.IOException;
import java.util.Hashtable;
import org.csource.common.MyException;
import org.csource.fastdfs.pool.Connection;
import java.net.InetSocketAddress;
/**
@@ -18,40 +21,112 @@ import java.net.InetSocketAddress;
* @version Version 1.11
*/
public class StorageServer extends TrackerServer {
protected int store_path_index = 0;
protected int store_path_index = 0;
/**
* Constructor
*
* @param ip_addr the ip address of storage server
* @param port the port of storage server
* @param store_path the store path index on the storage server
*/
public StorageServer(String ip_addr, int port, int store_path) throws IOException {
super(new InetSocketAddress(ip_addr, port));
this.store_path_index = store_path;
}
protected static Hashtable<String, InetSocketAddress> sockAddressCache = new Hashtable<String, InetSocketAddress>();
/**
* Constructor
*
* @param ip_addr the ip address of storage server
* @param port the port of storage server
* @param store_path the store path index on the storage server
*/
public StorageServer(String ip_addr, int port, byte store_path) throws IOException {
super(new InetSocketAddress(ip_addr, port));
if (store_path < 0) {
this.store_path_index = 256 + store_path;
} else {
this.store_path_index = store_path;
/**
* Constructor
*
* @param ip_addr the ip address of storage server
* @param port the port of storage server
* @param store_path the store path index on the storage server
*/
public StorageServer(String ip_addr, int port, int store_path) throws IOException {
super(new InetSocketAddress(ip_addr, port));
this.store_path_index = store_path;
}
}
/**
* @return the store path index on the storage server
*/
public int getStorePathIndex() {
return this.store_path_index;
}
/**
* Constructor
*
* @param ip_addr the ip address of storage server
* @param port the port of storage server
* @param store_path the store path index on the storage server
*/
public StorageServer(String ip_addr, int port, byte store_path) throws IOException {
super(new InetSocketAddress(ip_addr, port));
if (store_path < 0) {
this.store_path_index = 256 + store_path;
} else {
this.store_path_index = store_path;
}
}
/**
* @return the store path index on the storage server
*/
public int getStorePathIndex() {
return this.store_path_index;
}
public Connection getConnection() throws MyException, IOException {
Connection connection;
InetSocketAddress sockAddr;
MyException myException = null;
IOException ioException = null;
if (!ClientGlobal.g_multi_storage_ips) {
return super.getConnection();
}
if (ClientGlobal.g_connect_first_by == ClientGlobal.CONNECT_FIRST_BY_TRACKER) {
try {
if ((connection=super.getConnection()) != null) {
return connection;
}
} catch (MyException ex1) {
myException = ex1;
} catch (IOException ex2) {
ioException = ex2;
}
sockAddr = ClientGlobal.g_storages_address_map.get(this.inetSockAddr);
if (sockAddr != null) {
return super.getConnection(sockAddr);
} else if (myException != null) {
throw myException;
} else if (ioException != null) {
throw ioException;
}
} else {
String key = this.inetSockAddr.getAddress().getHostAddress() +
"@" + this.inetSockAddr.getPort();
sockAddr = sockAddressCache.get(key);
try {
if (sockAddr == null) {
sockAddr = this.inetSockAddr;
if ((connection=super.getConnection(sockAddr)) != null) {
sockAddressCache.put(key, sockAddr);
return connection;
}
} else {
if ((connection=super.getConnection(sockAddr)) != null) {
return connection;
}
}
} catch (MyException ex1) {
myException = ex1;
} catch (IOException ex2) {
ioException = ex2;
}
//retry another ip address
if ((sockAddr=ClientGlobal.g_storages_address_map.get(sockAddr)) == null) {
if (myException != null) {
throw myException;
} else if (ioException != null) {
throw ioException;
}
return null;
}
if ((connection=super.getConnection(sockAddr)) != null) {
sockAddressCache.put(key, sockAddr);
return connection;
}
}
return null;
}
}
src/main/java/org/csource/fastdfs/TrackerClient.java
+77 -2
View File
@@ -735,8 +735,9 @@ public class TrackerClient {
* @param storageIpAddr the storage server ip address
* @return true for success, false for fail
*/
public boolean deleteStorage(TrackerGroup trackerGroup,
String groupName, String storageIpAddr) throws IOException, MyException {
public boolean deleteStorage(TrackerGroup trackerGroup, String groupName,
String storageIpAddr) throws IOException, MyException
{
int serverIndex;
int notFoundCount;
TrackerServer trackerServer;
@@ -805,4 +806,78 @@ public class TrackerClient {
return this.errno == 0;
}
/**
* query storage server to upload file
*
* @param trackerServer the tracker server
* @param groupName the group name to upload file to, can be empty
* @return storage server object, return null if fail
*/
public StringBuilder fetchStorageIds() throws IOException, MyException {
byte[] header;
int offset = 0;
int length;
int total_count;
int current_count;
Connection connection = getConnection(null);
try {
OutputStream out = connection.getOutputStream();
StringBuilder builder = new StringBuilder();
header = ProtoCommon.packHeader(ProtoCommon.TRACKER_PROTO_CMD_FETCH_STORAGE_IDS, 5, (byte)0);
byte[] wholePkg = new byte[header.length + 5];
System.arraycopy(header, 0, wholePkg, 0, header.length);
wholePkg[wholePkg.length - 1] = 1;
do {
byte[] bs = ProtoCommon.int2buff(offset);
System.arraycopy(bs, 0, wholePkg, header.length, bs.length);
out.write(wholePkg);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(
connection.getInputStream(),
ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
}
if (pkgInfo.body.length < 8) {
throw new MyException("invalid body length: " + pkgInfo.body.length);
}
total_count = ProtoCommon.buff2int(pkgInfo.body, 0);
current_count = ProtoCommon.buff2int(pkgInfo.body, 4);
if (current_count < 0) {
throw new MyException("invalid current count: " + current_count);
}
length = pkgInfo.body.length - 8;
if (length == 0) {
break;
}
bs = new byte[length];
System.arraycopy(pkgInfo.body, 8, bs, 0, length);
builder.append(new String(bs, ClientGlobal.g_charset));
offset += current_count;
} while (offset < total_count);
return builder;
} catch (IOException e) {
try {
connection.close();
} finally {
connection = null;
}
throw e;
} finally {
if (connection != null) {
connection.release();
}
}
}
}
src/main/java/org/csource/fastdfs/TrackerServer.java
+8 -3
View File
@@ -37,15 +37,20 @@ public class TrackerServer {
this.index = index;
}
public Connection getConnection() throws MyException, IOException {
protected Connection getConnection(InetSocketAddress sockAddr) throws MyException, IOException {
Connection connection;
if (ClientGlobal.g_connection_pool_enabled) {
connection = ConnectionPool.getConnection(this.inetSockAddr);
connection = ConnectionPool.getConnection(sockAddr);
} else {
connection = ConnectionFactory.create(this.inetSockAddr);
connection = ConnectionFactory.create(sockAddr);
}
return connection;
}
public Connection getConnection() throws MyException, IOException {
return this.getConnection(this.inetSockAddr);
}
/**
* get the server info
*
src/main/resources/fastdfs-client.properties.sample
+8
View File
@@ -18,6 +18,14 @@ fastdfs.http_tracker_http_port = 80
#
fastdfs.tracker_servers = 185.245.40.70:22122
# connect which ip address first for multi IPs of a storage server, value list:
## tracker: connect to the ip address return by tracker server first
## last-connected: connect to the ip address last connected first
# default value is tracker
fastdfs.connect_first_by = tracker
## Whether to open the connection pool, if not, create a new connection every time
fastdfs.connection_pool.enabled = true
src/main/resources/fdfs_client.conf.sample
+8
View File
@@ -15,6 +15,14 @@ http.secret_key = FastDFS1234567890
tracker_server = 10.0.11.243:22122
tracker_server = 10.0.11.244:22122
# connect which ip address first for multi IPs of a storage server, value list:
## tracker: connect to the ip address return by tracker server first
## last-connected: connect to the ip address last connected first
# default value is tracker
connect_first_by = tracker
connection_pool.enabled = true
connection_pool.max_count_per_entry = 500
connection_pool.max_idle_time = 3600