提交 5e259f86 编写于 作者: Y yukon

Avoid using guava Files in FileWatchService

上级 8c0759e3
...@@ -404,7 +404,7 @@ public class BrokerController { ...@@ -404,7 +404,7 @@ public class BrokerController {
((NettyRemotingServer) fastRemotingServer).loadSslContext(); ((NettyRemotingServer) fastRemotingServer).loadSslContext();
} }
}); });
} catch (IOException e) { } catch (Exception e) {
log.warn("FileWatchService created error, can't load the certificate dynamically"); log.warn("FileWatchService created error, can't load the certificate dynamically");
} }
} }
......
...@@ -16,7 +16,6 @@ ...@@ -16,7 +16,6 @@
*/ */
package org.apache.rocketmq.namesrv; package org.apache.rocketmq.namesrv;
import java.io.IOException;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
...@@ -111,7 +110,7 @@ public class NamesrvController { ...@@ -111,7 +110,7 @@ public class NamesrvController {
((NettyRemotingServer) remotingServer).loadSslContext(); ((NettyRemotingServer) remotingServer).loadSslContext();
} }
}); });
} catch (IOException e) { } catch (Exception e) {
log.warn("FileWatchService created error, can't load the certificate dynamically"); log.warn("FileWatchService created error, can't load the certificate dynamically");
} }
} }
......
...@@ -17,12 +17,14 @@ ...@@ -17,12 +17,14 @@
package org.apache.rocketmq.srvutil; package org.apache.rocketmq.srvutil;
import com.google.common.hash.HashCode;
import com.google.common.hash.Hashing;
import com.google.common.io.Files;
import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import org.apache.rocketmq.common.ServiceThread; import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.constant.LoggerName;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
...@@ -30,23 +32,23 @@ import org.slf4j.LoggerFactory; ...@@ -30,23 +32,23 @@ import org.slf4j.LoggerFactory;
public class FileWatchService extends ServiceThread { public class FileWatchService extends ServiceThread {
private static final Logger log = LoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME); private static final Logger log = LoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
private String [] watchFiles; private final String [] watchFiles;
private boolean [] isFileChangedFlag; private final boolean [] isFileChangedFlag;
private HashCode [] fileCurrentHash; private final String [] fileCurrentHash;
private Listener listener; private final Listener listener;
private static final int WATCH_INTERVAL = 100; private static final int WATCH_INTERVAL = 500;
private MessageDigest md = MessageDigest.getInstance("MD5");
public FileWatchService(final String [] watchFiles, public FileWatchService(final String [] watchFiles,
final Listener listener) throws IOException { final Listener listener) throws Exception {
this.watchFiles = watchFiles; this.watchFiles = watchFiles;
this.listener = listener; this.listener = listener;
this.isFileChangedFlag = new boolean[watchFiles.length]; this.isFileChangedFlag = new boolean[watchFiles.length];
this.fileCurrentHash = new HashCode[watchFiles.length]; this.fileCurrentHash = new String[watchFiles.length];
for (int i = 0; i < watchFiles.length; i++) { for (int i = 0; i < watchFiles.length; i++) {
isFileChangedFlag[i] = false; isFileChangedFlag[i] = false;
fileCurrentHash[i] = Files.hash(new File(watchFiles[i]), Hashing.md5()); fileCurrentHash[i] = hash(watchFiles[i]);
} }
} }
...@@ -65,7 +67,7 @@ public class FileWatchService extends ServiceThread { ...@@ -65,7 +67,7 @@ public class FileWatchService extends ServiceThread {
boolean allFileChanged = true; boolean allFileChanged = true;
for (int i = 0; i < watchFiles.length; i++) { for (int i = 0; i < watchFiles.length; i++) {
HashCode newHash = Files.hash(new File(watchFiles[i]), Hashing.md5()); String newHash = hash(watchFiles[i]);
if (!newHash.equals(fileCurrentHash[i])) { if (!newHash.equals(fileCurrentHash[i])) {
isFileChangedFlag[i] = true; isFileChangedFlag[i] = true;
fileCurrentHash[i] = newHash; fileCurrentHash[i] = newHash;
...@@ -86,6 +88,13 @@ public class FileWatchService extends ServiceThread { ...@@ -86,6 +88,13 @@ public class FileWatchService extends ServiceThread {
log.info(this.getServiceName() + " service end"); log.info(this.getServiceName() + " service end");
} }
private String hash(String filePath) throws IOException, NoSuchAlgorithmException {
Path path = Paths.get(filePath);
md.update(Files.readAllBytes(path));
byte[] hash = md.digest();
return UtilAll.bytes2string(hash);
}
public interface Listener { public interface Listener {
/** /**
* Will be called when the target files are changed * Will be called when the target files are changed
......
...@@ -20,6 +20,7 @@ package org.apache.rocketmq.srvutil; ...@@ -20,6 +20,7 @@ package org.apache.rocketmq.srvutil;
import java.io.File; import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.io.PrintWriter; import java.io.PrintWriter;
import java.nio.file.NoSuchFileException;
import java.util.concurrent.Semaphore; import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import org.junit.Rule; import org.junit.Rule;
...@@ -29,6 +30,7 @@ import org.junit.runner.RunWith; ...@@ -29,6 +30,7 @@ import org.junit.runner.RunWith;
import org.mockito.junit.MockitoJUnitRunner; import org.mockito.junit.MockitoJUnitRunner;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Java6Assertions.failBecauseExceptionWasNotThrown;
@RunWith(MockitoJUnitRunner.class) @RunWith(MockitoJUnitRunner.class)
public class FileWatchServiceTest { public class FileWatchServiceTest {
...@@ -36,7 +38,7 @@ public class FileWatchServiceTest { ...@@ -36,7 +38,7 @@ public class FileWatchServiceTest {
public TemporaryFolder tempFolder = new TemporaryFolder(); public TemporaryFolder tempFolder = new TemporaryFolder();
@Test @Test
public void watchSingleFile() throws IOException, InterruptedException { public void watchSingleFile() throws Exception {
File file = tempFolder.newFile(); File file = tempFolder.newFile();
final Semaphore waitSemaphore = new Semaphore(0); final Semaphore waitSemaphore = new Semaphore(0);
FileWatchService fileWatchService = new FileWatchService(new String[] {file.getAbsolutePath()}, new FileWatchService.Listener() { FileWatchService fileWatchService = new FileWatchService(new String[] {file.getAbsolutePath()}, new FileWatchService.Listener() {
...@@ -47,13 +49,49 @@ public class FileWatchServiceTest { ...@@ -47,13 +49,49 @@ public class FileWatchServiceTest {
}); });
fileWatchService.start(); fileWatchService.start();
modifyFile(file); modifyFile(file);
boolean result = waitSemaphore.tryAcquire(1, 100, TimeUnit.MILLISECONDS); boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isTrue(); assertThat(result).isTrue();
}
@Test
public void watchSingleFile_NotExits() throws Exception {
File file = tempFolder.newFile();
final Semaphore waitSemaphore = new Semaphore(0);
try {
FileWatchService fileWatchService = new FileWatchService(new String[] {file.getAbsolutePath() + 123}, new FileWatchService.Listener() {
@Override
public void onChanged() {
waitSemaphore.release();
}
});
failBecauseExceptionWasNotThrown(NoSuchFileException.class);
} catch (Exception e) {
assertThat(e).isInstanceOf(NoSuchFileException.class);
}
}
@Test
public void watchSingleFile_FileDeleted() throws Exception {
File file = tempFolder.newFile();
final Semaphore waitSemaphore = new Semaphore(0);
FileWatchService fileWatchService = new FileWatchService(new String[] {file.getAbsolutePath()}, new FileWatchService.Listener() {
@Override
public void onChanged() {
waitSemaphore.release();
}
});
fileWatchService.start();
file.delete();
boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isFalse();
file.createNewFile();
modifyFile(file);
result = waitSemaphore.tryAcquire(1, 2000, TimeUnit.MILLISECONDS);
assertThat(result).isTrue();
} }
@Test @Test
public void watchTwoFiles_ModifyOne() throws IOException, InterruptedException { public void watchTwoFiles_ModifyOne() throws Exception {
File fileA = tempFolder.newFile(); File fileA = tempFolder.newFile();
File fileB = tempFolder.newFile(); File fileB = tempFolder.newFile();
final Semaphore waitSemaphore = new Semaphore(0); final Semaphore waitSemaphore = new Semaphore(0);
...@@ -67,12 +105,12 @@ public class FileWatchServiceTest { ...@@ -67,12 +105,12 @@ public class FileWatchServiceTest {
}); });
fileWatchService.start(); fileWatchService.start();
modifyFile(fileA); modifyFile(fileA);
boolean result = waitSemaphore.tryAcquire(1, 100, TimeUnit.MILLISECONDS); boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isFalse(); assertThat(result).isFalse();
} }
@Test @Test
public void watchTwoFiles() throws IOException, InterruptedException { public void watchTwoFiles() throws Exception {
File fileA = tempFolder.newFile(); File fileA = tempFolder.newFile();
File fileB = tempFolder.newFile(); File fileB = tempFolder.newFile();
final Semaphore waitSemaphore = new Semaphore(0); final Semaphore waitSemaphore = new Semaphore(0);
...@@ -87,14 +125,14 @@ public class FileWatchServiceTest { ...@@ -87,14 +125,14 @@ public class FileWatchServiceTest {
fileWatchService.start(); fileWatchService.start();
modifyFile(fileA); modifyFile(fileA);
modifyFile(fileB); modifyFile(fileB);
boolean result = waitSemaphore.tryAcquire(1, 100, TimeUnit.MILLISECONDS); boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isTrue(); assertThat(result).isTrue();
} }
private static void modifyFile(File file) { private static void modifyFile(File file) {
try { try {
PrintWriter out = new PrintWriter(file); PrintWriter out = new PrintWriter(file);
out.println(System.currentTimeMillis()); out.println(System.nanoTime());
out.flush(); out.flush();
out.close(); out.close();
} catch (IOException ignore) { } catch (IOException ignore) {
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册