重构过滤服务器，为diamond存储过滤类做准备

Author: vintagewang

rocketmq-filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvStartup.java

@@ -47,6 +47,8 @@
  * @since 2014-4-10
  */
 public class FiltersrvStartup {
+    public static Logger log;
+
 
     public static Options buildCommandlineOptions(final Options options) {
         Option opt = new Option("c", "configFile", true, "Filter server config properties file");
@@ -62,11 +64,29 @@ public static Options buildCommandlineOptions(final Options options) {
 
 
     public static void main(String[] args) {
-        main0(args);
+        start(createController(args));
     }
 
 
-    public static FiltersrvController main0(String[] args) {
+    public static FiltersrvController start(FiltersrvController controller) {
+        // 启动服务
+        try {
+            controller.start();
+        }
+        catch (Exception e) {
+            e.printStackTrace();
+            System.exit(-1);
+        }
+
+        String tip = "The Filter Server boot success, " + controller.localAddr();
+        log.info(tip);
+        System.out.println(tip);
+
+        return controller;
+    }
+
+
+    public static FiltersrvController createController(String[] args) {
         System.setProperty(RemotingCommand.RemotingVersionKey, Integer.toString(MQVersion.CurrentVersion));
 
         // Socket发送缓冲区大小
@@ -140,7 +160,7 @@ public static FiltersrvController main0(String[] args) {
             configurator.setContext(lc);
             lc.reset();
             configurator.doConfigure(filtersrvConfig.getRocketmqHome() + "/conf/logback_filtersrv.xml");
-            final Logger log = LoggerFactory.getLogger(LoggerName.FiltersrvLoggerName);
+            log = LoggerFactory.getLogger(LoggerName.FiltersrvLoggerName);
 
             // 初始化服务控制对象
             final FiltersrvController controller =
@@ -171,13 +191,6 @@ public void run() {
                 }
             }, "ShutdownHook"));
 
-            // 启动服务
-            controller.start();
-
-            String tip = "The Filter Server boot success, " + controller.localAddr();
-            log.info(tip);
-            System.out.println(tip);
-
             return controller;
         }
         catch (Throwable e) {

rocketmq-filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassFetchMethod.java

@@ -0,0 +1,5 @@
+package com.alibaba.rocketmq.filtersrv.filter;
+
+public interface FilterClassFetchMethod {
+    public String fetch(final String topic, final String consumerGroup, final String className);
+}

rocketmq-filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassManager.java

@@ -15,8 +15,6 @@
 import com.alibaba.rocketmq.common.UtilAll;
 import com.alibaba.rocketmq.common.constant.LoggerName;
 import com.alibaba.rocketmq.common.filter.MessageFilter;
-import com.alibaba.rocketmq.common.utils.HttpTinyClient;
-import com.alibaba.rocketmq.common.utils.HttpTinyClient.HttpResult;
 import com.alibaba.rocketmq.filtersrv.FiltersrvController;
 
 
@@ -35,9 +33,14 @@ public class FilterClassManager {
     private final ScheduledExecutorService scheduledExecutorService = Executors
         .newSingleThreadScheduledExecutor(new ThreadFactoryImpl("FSGetClassScheduledThread"));
 
+    private FilterClassFetchMethod filterClassFetchMethod;
+
 
     public FilterClassManager(FiltersrvController filtersrvController) {
         this.filtersrvController = filtersrvController;
+        this.filterClassFetchMethod =
+                new HttpFilterClassFetchMethod(this.filtersrvController.getFiltersrvConfig()
+                    .getFilterClassRepertoryUrl());
     }
 
 
@@ -65,28 +68,22 @@ private void fetchClassFromRemoteHost() {
             try {
                 Entry<String, FilterClassInfo> next = it.next();
                 FilterClassInfo filterClassInfo = next.getValue();
-
-                String url = this.filtersrvController.getFiltersrvConfig().getFilterClassRepertoryUrl();
-                url += "/";
-                url += filterClassInfo.getClassName();
-                url += ".java";
-
-                HttpResult result = HttpTinyClient.httpGet(url, null, null, "UTF-8", 5000);
-                if (200 == result.code) {
-                    String responseStr = result.content;
-                    byte[] filterSourceBinary = responseStr.getBytes("UTF-8");
-                    int classCRC = UtilAll.crc32(responseStr.getBytes("UTF-8"));
-                    if (classCRC != filterClassInfo.getClassCRC()) {
-                        String javaSource = new String(filterSourceBinary, MixAll.DEFAULT_CHARSET);
-                        Class<?> newClass =
-                                DynaCode.compileAndLoadClass(filterClassInfo.getClassName(), javaSource);
-                        Object newInstance = newClass.newInstance();
-                        filterClassInfo.setMessageFilter((MessageFilter) newInstance);
-                        filterClassInfo.setClassCRC(classCRC);
-
-                        log.info("fetch Remote class File OK, {} {} {}", next.getKey(),
-                            filterClassInfo.getClassName(), url);
-                    }
+                String[] topicAndGroup = next.getKey().split("@");
+                String responseStr =
+                        this.filterClassFetchMethod.fetch(topicAndGroup[0], topicAndGroup[1],
+                            filterClassInfo.getClassName());
+                byte[] filterSourceBinary = responseStr.getBytes("UTF-8");
+                int classCRC = UtilAll.crc32(responseStr.getBytes("UTF-8"));
+                if (classCRC != filterClassInfo.getClassCRC()) {
+                    String javaSource = new String(filterSourceBinary, MixAll.DEFAULT_CHARSET);
+                    Class<?> newClass =
+                            DynaCode.compileAndLoadClass(filterClassInfo.getClassName(), javaSource);
+                    Object newInstance = newClass.newInstance();
+                    filterClassInfo.setMessageFilter((MessageFilter) newInstance);
+                    filterClassInfo.setClassCRC(classCRC);
+
+                    log.info("fetch Remote class File OK, {} {}", next.getKey(),
+                        filterClassInfo.getClassName());
                 }
             }
             catch (Exception e) {
@@ -158,4 +155,14 @@ public boolean registerFilterClass(final String consumerGroup, final String topi
     public FilterClassInfo findFilterClass(final String consumerGroup, final String topic) {
         return this.filterClassTable.get(buildKey(consumerGroup, topic));
     }
+
+
+    public FilterClassFetchMethod getFilterClassFetchMethod() {
+        return filterClassFetchMethod;
+    }
+
+
+    public void setFilterClassFetchMethod(FilterClassFetchMethod filterClassFetchMethod) {
+        this.filterClassFetchMethod = filterClassFetchMethod;
+    }
 }

rocketmq-filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/HttpFilterClassFetchMethod.java

@@ -0,0 +1,38 @@
+package com.alibaba.rocketmq.filtersrv.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.rocketmq.common.constant.LoggerName;
+import com.alibaba.rocketmq.common.utils.HttpTinyClient;
+import com.alibaba.rocketmq.common.utils.HttpTinyClient.HttpResult;
+
+
+public class HttpFilterClassFetchMethod implements FilterClassFetchMethod {
+    private static final Logger log = LoggerFactory.getLogger(LoggerName.FiltersrvLoggerName);
+    private final String url;
+
+
+    public HttpFilterClassFetchMethod(String url) {
+        this.url = url;
+    }
+
+
+    @Override
+    public String fetch(String topic, String consumerGroup, String className) {
+        String thisUrl = String.format("%s/%s.java", this.url, className);
+
+        try {
+            HttpResult result = HttpTinyClient.httpGet(thisUrl, null, null, "UTF-8", 5000);
+            if (200 == result.code) {
+                return result.content;
+            }
+        }
+        catch (Exception e) {
+            log.error(
+                String.format("call <%s> exception, Topic: %s Group: %s", thisUrl, topic, consumerGroup), e);
+        }
+
+        return null;
+    }
+}