增加动态配置开关

+package com.yoho.datasync.consumer.common;
+
+public class IndexConstant {
+
+    public interface INDEX_NAME{
+        String GRASS = "grass";
+    }
+
+    public interface INDEX_TYPE{
+        String USER_INFO = "userInfo";
+    }
+}

@@ -69,6 +69,43 @@
             <artifactId>elasticsearch</artifactId>
             <artifactId>elasticsearch</artifactId>
             <version>6.5.2</version>
             <version>6.5.2</version>
         </dependency>
         </dependency>
+        <dependency>
+            <groupId>commons-collections</groupId>
+            <artifactId>commons-collections</artifactId>
+            <version>3.2.2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.netflix.archaius</groupId>
+            <artifactId>archaius-core</artifactId>
+            <version>0.7.3</version>
+        </dependency>
+        <dependency>
+            <groupId>com.netflix.archaius</groupId>
+            <artifactId>archaius-zookeeper</artifactId>
+            <version>0.7.3</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-framework</artifactId>
+            <version>2.9.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-client</artifactId>
+            <version>2.9.1</version>
+        </dependency>
+        <dependency>
+            <groupId>commons-configuration</groupId>
+            <artifactId>commons-configuration</artifactId>
+            <version>1.5</version>
+        </dependency>
 
 
     </dependencies>
     </dependencies>
 
 

+package com.yoho.datasync.consumer.handler.config;
+
+import com.netflix.config.DynamicWatchedConfiguration;
+import com.netflix.config.source.ZooKeeperConfigurationSource;
+import org.apache.curator.framework.CuratorFramework;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PropertiesLoaderUtils;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PreDestroy;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static com.netflix.config.ConfigurationManager.install;
+
+/**
+ * 读取classpath下的config.properties文件，并且写入到zookeeper中
+ */
+@Component
+public class ConfigMangerRegister implements InitializingBean {
+
+    @javax.annotation.Resource
+    private CuratorFramework curatorFramework;
+
+
+    private static final Logger logger = LoggerFactory.getLogger(ConfigMangerRegister.class);
+    private static final String CONFIG_ROOT_PATH = "/yh/config";
+    private static final String LOCAL_CONFIG_FILE = "global.properties";
+    private static final String CACHE_LOCAL_CONFIG_FILE = "cache.properties";
+
+
+    private static AtomicBoolean inited =new AtomicBoolean(false);
+
+
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        start();
+    }
+
+    /**
+     * 1. 注册zookeeper配置的支持到netflix archaius中
+     * 2. 读取config.properties的配置，添加到zookeeper中
+     */
+    public void start() throws Exception {
+        /**
+         * return directly if zk client not be configured
+         */
+        if(null == curatorFramework){
+            return ;
+        }
+
+
+        /**
+         * create path if not exist
+         */
+        if (curatorFramework.checkExists().forPath(CONFIG_ROOT_PATH) == null) {
+            curatorFramework.create().creatingParentContainersIfNeeded().forPath(CONFIG_ROOT_PATH);
+        }
+
+        //register to archaius
+        this.registerZookeeperConfig();
+
+        //register all configurations to zookeeper
+        this.registerConfigurations();
+        
+        // register cache configurations to zookeeper
+        this.registerCacheConfigurations();
+    }
+
+
+
+    /**
+     * register zookeeper config source to netflix
+     *
+     * @see <a href = "http://github.com/Netflix/archaius/wiki/ZooKeeper-Dynamic-Configuration"> netflix archaius </a>
+     */
+    private void registerZookeeperConfig() {
+    	
+    	if(!inited.compareAndSet(false, true)){
+    		return;
+    	}
+        ZooKeeperConfigurationSource zkConfigSource = new ZooKeeperConfigurationSource(curatorFramework, CONFIG_ROOT_PATH);
+        try {
+            zkConfigSource.start();
+        } catch (Exception e) {
+            logger.error("start zk config source error !");
+        }
+
+        DynamicWatchedConfiguration zkDynamicConfig = new DynamicWatchedConfiguration(zkConfigSource);
+        install(zkDynamicConfig);
+    }
+
+    /**
+     * 读取配置，并且注册到zk中
+     */
+    private void registerConfigurations() {
+
+        try {
+
+            //check if resource existed
+            Resource resource = new ClassPathResource(LOCAL_CONFIG_FILE);
+            if(!resource.exists()) {
+                logger.info("can not find {} at classpath.", LOCAL_CONFIG_FILE);
+                return;
+            }
+
+            Properties props = PropertiesLoaderUtils.loadProperties(resource);
+
+            for (String propertyName : props.stringPropertyNames()) {
+                final String path = ConfigMangerRegister.CONFIG_ROOT_PATH + "/" + propertyName;
+                //if path is not existed, create
+                if (this.curatorFramework.checkExists().forPath(path) == null) {
+                    String value = String.valueOf(props.getProperty(propertyName));
+                    this.curatorFramework.create().forPath(path, value.getBytes("UTF-8"));
+
+                    logger.debug("register configuration: {} --> {} success", propertyName, value);
+                }
+            }
+
+            logger.info("register all configurations success");
+
+        }catch (Exception e){
+            logger.error("register all configurations to zookeeper failed with exception.", e);
+        }
+    }
+    
+    
+
+    /**
+     * 读取配置，并且注册到zk中
+     */
+    private void registerCacheConfigurations() {
+
+        try {
+
+            //check if resource existed
+            Resource resource = new ClassPathResource(CACHE_LOCAL_CONFIG_FILE);
+            if(!resource.exists()) {
+                logger.info("can not find {} at classpath.", CACHE_LOCAL_CONFIG_FILE);
+                return;
+            }
+            Properties props = PropertiesLoaderUtils.loadProperties(resource);
+            for (String propertyName : props.stringPropertyNames()) {
+                final String path = ConfigMangerRegister.CONFIG_ROOT_PATH + "/" + propertyName;
+                //if path is not existed, create
+                if (this.curatorFramework.checkExists().forPath(path) == null) {
+                    String value = String.valueOf(props.getProperty(propertyName));
+                    this.curatorFramework.create().forPath(path, value.getBytes("UTF-8"));
+
+                    logger.debug("register configuration: {} --> {} success", propertyName, value);
+                }
+            }
+
+            logger.info("register all configurations success");
+        }catch (Exception e){
+            logger.error("register all configurations to zookeeper failed with exception.", e);
+        }
+    }
+
+
+    //called by spring
+    @PreDestroy
+    public void destroy() {
+        curatorFramework.close();
+    }
+}

+package com.yoho.datasync.consumer.handler.config;
+
+import com.netflix.config.*;
+import org.springframework.stereotype.Component;
+
+/**
+ *  读取配置。 当前支持读取zookeeper中的配置
+ */
+@Component
+public class ConfigReader {
+
+    /**
+     * 读取Spring类型的配置
+     * @param key  Key
+     * @param defaultValue 默认值
+     * @return value
+     */
+    public String getString(String key, String defaultValue) {
+        final DynamicStringProperty property = DynamicPropertyFactory
+                .getInstance().getStringProperty(key, defaultValue);
+        return property.get();
+    }
+
+    /**
+     * 读取int类型的配置
+     * @param key  Key
+     * @param defaultValue 默认值
+     * @return value
+     */
+    public int getInt(String key, int defaultValue) {
+        final DynamicIntProperty property = DynamicPropertyFactory
+                .getInstance().getIntProperty(key, defaultValue);
+        return property.get();
+    }
+
+    /**
+     * 读取int类型的配置
+     * @param key  Key
+     * @param defaultValue 默认值
+     * @return value
+     */
+    public double getDouble(String key, double defaultValue) {
+        final DynamicDoubleProperty property = DynamicPropertyFactory
+                .getInstance().getDoubleProperty(key, defaultValue);
+        return property.get();
+    }
+    
+    
+    /**
+     * 读取long类型的配置
+     * @param key  Key
+     * @param defaultValue 默认值
+     * @return value
+     */
+    public long getLong(String key, long defaultValue) {
+        final DynamicLongProperty property = DynamicPropertyFactory
+                .getInstance().getLongProperty(key, defaultValue);
+        return property.get();
+    }
+
+
+    /**
+     * 读取boolean类型的配置
+     * @param key  Key
+     * @param defaultValue 默认值
+     * @return value
+     */
+    public boolean getBoolean(String key, boolean defaultValue) {
+        final DynamicBooleanProperty property = DynamicPropertyFactory
+                .getInstance().getBooleanProperty(key, defaultValue);
+        return property.get();
+    }
+
+
+}

+package com.yoho.datasync.consumer.handler.config;
+
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * Created by markeloff on 16/10/8.
+ */
+@Component
+public class ConfigReaderUtil  {
+
+    @Resource
+    private ConfigReader configReader;
+
+    public  String getString(String key, String defaultValue) {
+        if(configReader==null){
+            return defaultValue;
+        }
+        return configReader.getString(key, defaultValue);
+    }
+
+    public int getInt(String key, int defaultValue) {
+        if(configReader==null){
+            return defaultValue;
+        }
+        return configReader.getInt(key,defaultValue);
+    }
+    
+    public long getLong(String key, long defaultValue) {
+        if(configReader==null){
+            return defaultValue;
+        }
+        return configReader.getLong(key,defaultValue);
+   }
+
+    public boolean getBoolean(String key, boolean defaultValue) {
+        if(configReader==null){
+            return defaultValue;
+        }
+         return configReader.getBoolean(key,defaultValue);
+    }
+
+}

+package com.yoho.datasync.consumer.handler.config;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.PreDestroy;
+
+
+/**
+ *   获取CuratorFramework 客户端
+ */
+@Configuration
+@ConfigurationProperties(prefix = "zk")
+public class CuratorFrameworkFactoryBean {
+  private final static Logger logger = LoggerFactory.getLogger(CuratorFrameworkFactoryBean.class);
+
+  private CuratorFramework curator;
+  private String connectString;
+  private RetryPolicy retryPolicy;
+  private Integer sessionTimeout;
+  private String namespace;
+
+  @Bean
+  public CuratorFramework curatorFramework(){
+    if(StringUtils.isEmpty(connectString)){
+       return null;
+    }
+
+    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
+
+    builder.connectString(connectString);
+
+    if (retryPolicy == null) {
+      retryPolicy = new ExponentialBackoffRetry(1000, 3);
+    }
+    builder.retryPolicy(retryPolicy);
+
+    if (sessionTimeout != null) {
+      builder.sessionTimeoutMs(sessionTimeout);
+    }
+
+    if (namespace != null) {
+      builder.namespace(namespace);
+    }
+
+    curator = builder.build();
+    curator.start();
+    logger.info("start zookeeper client success. connected to {}", connectString);
+    return curator;
+  }
+
+
+
+  @PreDestroy
+  public void destroy() throws Exception {
+    curator.close();
+  }
+
+  public String getConnectString() {
+    return connectString;
+  }
+
+  public void setConnectString(String connectString) {
+    this.connectString = connectString;
+  }
+
+  public Integer getSessionTimeout() {
+    return sessionTimeout;
+  }
+
+  public void setSessionTimeout(Integer sessionTimeout) {
+    this.sessionTimeout = sessionTimeout;
+  }
+
+  public RetryPolicy getRetryPolicy() {
+    return retryPolicy;
+  }
+  
+  public void setRetryPolicy(RetryPolicy retryPolicy) {
+    this.retryPolicy = retryPolicy;
+  }
+
+  public String getNamespace() {
+    return namespace;
+  }
+
+  public void setNamespace(String namespace) {
+    this.namespace = namespace;
+  }
+
+}

@@ -2,21 +2,18 @@ package com.yoho.datasync.consumer.handler.listener.community;
 
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSON;
 import com.yoho.datasync.consumer.common.EventEnum;
 import com.yoho.datasync.consumer.common.EventEnum;
+import com.yoho.datasync.consumer.common.IndexConstant;
+import com.yoho.datasync.consumer.handler.config.ConfigReaderUtil;
+import com.yoho.datasync.consumer.handler.model.ESBluk;
 import com.yoho.datasync.consumer.handler.mqcomponent.AbstractMqListener;
 import com.yoho.datasync.consumer.handler.mqcomponent.AbstractMqListener;
+import com.yoho.datasync.consumer.handler.processor.ESUpdateProcess;
 import com.yoho.datasync.core.base.annotation.MqConsumerListerner;
 import com.yoho.datasync.core.base.annotation.MqConsumerListerner;
 import com.yoho.datasync.core.base.model.yh_community.UserInfo;
 import com.yoho.datasync.core.base.model.yh_community.UserInfo;
-import org.elasticsearch.action.update.UpdateRequest;
-import org.elasticsearch.action.update.UpdateResponse;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.common.xcontent.XContentFactory;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 import org.springframework.stereotype.Component;
 
 
-import java.io.IOException;
+import javax.annotation.Resource;
 
 
 @Component
 @Component
 @MqConsumerListerner(dbName = "yh_community",tableName = "user_info")
 @MqConsumerListerner(dbName = "yh_community",tableName = "user_info")
@@ -25,38 +22,34 @@ public class UserInfoListener extends AbstractMqListener<UserInfo> {
 
 
     private static final Logger logger = LoggerFactory.getLogger(UserInfoListener.class);
     private static final Logger logger = LoggerFactory.getLogger(UserInfoListener.class);
 
 
-    @Qualifier("rhlClient")
-    @Autowired
-    private RestHighLevelClient rhlClient;
+    @Resource
+    private ESUpdateProcess esUpdateProcess;
 
 
-    @Override
-    protected void deleteData(UserInfo sourceObject , Object checkResult) throws Exception {
+    @Resource
+    private ConfigReaderUtil configReaderUtil;
 
 
+    @Override
+    protected void deleteData(UserInfo userInfo , Object checkResult) throws Exception {
+        logger.info("enter UserInfoListener delete userInfo data is {}" , userInfo);
+        boolean syncFlag = configReaderUtil.getBoolean("grass.search.es.flag", false);
+        if(syncFlag){
+            //更新用户信息索引到ES 添加到队列中
+            ESBluk esBluk = new ESBluk(null,String.valueOf(userInfo.getYohoUid()),
+                    IndexConstant.INDEX_NAME.GRASS,IndexConstant.INDEX_TYPE.USER_INFO,true);
+            esUpdateProcess.add(esBluk);
+        }
     }
     }
 
 
     @Override
     @Override
     protected void updateData(UserInfo userInfo , Object checkResult) throws Exception {
     protected void updateData(UserInfo userInfo , Object checkResult) throws Exception {
-        //更新用户信息索引到ES
-        // 批量增加
-        UpdateRequest updateRequest = new UpdateRequest();
-        updateRequest.index("grass");
-        updateRequest.type("userInfo");
-        updateRequest.id(String.valueOf(userInfo.getYohoUid()));
-        updateRequest.doc(XContentFactory.jsonBuilder().startObject()
-                .field("nick_name", userInfo.getNickName())
-                .field("head_ico", userInfo.getHeadIco())
-                .field("yoho_uid", userInfo.getYohoUid())
-                .field("gender", userInfo.getGender())
-                .endObject());
-
-        try {
-            // 2 获取更新后的值
-            UpdateResponse indexResponse = rhlClient.update(updateRequest, RequestOptions.DEFAULT);
-            logger.info("updateIndex: {}" , JSON.toJSONString(indexResponse));
-        } catch (IOException e) {
-            logger.info("updateIndex error :{}" ,e);
+        logger.info("enter UserInfoListener update userInfo data is {}" , userInfo);
+        boolean syncFlag = configReaderUtil.getBoolean("grass.search.es.flag", false);
+        if(syncFlag){
+            //更新用户信息索引到ES 添加到队列中
+            ESBluk esBluk = new ESBluk(JSON.toJSONString(userInfo),String.valueOf(userInfo.getYohoUid()),
+                    IndexConstant.INDEX_NAME.GRASS,IndexConstant.INDEX_TYPE.USER_INFO,false);
+            esUpdateProcess.add(esBluk);
         }
         }
-
     }
     }
 
 
 
 

+package com.yoho.datasync.consumer.handler.model;
+
+/**
+ * Created by apple on 16/8/10.
+ */
+public class ESBluk {
+	
+	private String indexName;
+	private String type;
+	private String id;
+	private String dataJson;
+	private boolean isDelete;
+
+	public ESBluk(String dataJson, String id, String indexName, String type, boolean isDelete) {
+		this.dataJson = dataJson;
+		this.id = id;
+		this.indexName = indexName;
+		this.isDelete = isDelete;
+		this.type = type;
+	}
+
+	@Override
+	public String toString() {
+		return "ESBluk{" + "indexName='" + indexName + '\'' + ", type='" + type + '\'' + ", id='" + id + '\'' + ", dataJson='" + dataJson + '\'' + ", isDelete=" + isDelete + '}';
+	}
+
+	public void replace(ESBluk bluk) {
+		this.dataJson = bluk.getDataJson();
+		this.id = bluk.getId();
+		this.indexName = bluk.getIndexName();
+		this.isDelete = bluk.isDelete();
+		this.type = bluk.getType();
+	}
+
+	public String getDataJson() {
+		return dataJson;
+	}
+
+	public void setDataJson(String dataJson) {
+		this.dataJson = dataJson;
+	}
+
+	public String getId() {
+		return id;
+	}
+
+	public void setId(String id) {
+		this.id = id;
+	}
+
+	public String getIndexName() {
+		return indexName;
+	}
+
+	public void setIndexName(String indexName) {
+		this.indexName = indexName;
+	}
+
+	public boolean isDelete() {
+		return isDelete;
+	}
+
+	public void setDelete(boolean delete) {
+		isDelete = delete;
+	}
+
+	public String getType() {
+		return type;
+	}
+
+	public void setType(String type) {
+		this.type = type;
+	}
+}

+package com.yoho.datasync.consumer.handler.processor;
+
+import com.alibaba.fastjson.JSON;
+import com.yoho.datasync.consumer.handler.model.ESBluk;
+import org.apache.commons.collections.CollectionUtils;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Component
+public class ESUpdateProcess {
+    private static final Logger logger = LoggerFactory.getLogger(ESUpdateProcess.class);
+
+    private final ArrayBlockingQueue<ESBluk> queue = new ArrayBlockingQueue<>(200);
+
+
+    @Qualifier("rhlClient")
+    @Resource
+    private RestHighLevelClient rhlClient;
+
+    @PostConstruct
+    void init() {
+        // 批量更新ES
+        ExecutorService executorService = Executors.newSingleThreadExecutor();
+        executorService.submit(() -> {
+            while (true) {
+                doBulk();
+            }
+        });
+    }
+
+    private void doBulk() {
+        try {
+            long begin = System.currentTimeMillis();
+
+            //1、从队列中获取全部数据
+            List<ESBluk> blukList = new ArrayList<>();
+            queue.drainTo(blukList);
+
+            //2、批量更新Es
+            if (CollectionUtils.isNotEmpty(blukList)) {
+                BulkRequest bulkARequest = new BulkRequest();
+                for (ESBluk eSBluk : blukList) {
+                    if(!eSBluk.isDelete()){
+                        //新增或者更新
+                        IndexRequest indexRequest = new IndexRequest(eSBluk.getIndexName(), eSBluk.getType(), eSBluk.getId());
+                        indexRequest.source(eSBluk.getDataJson(), XContentType.JSON);
+                        bulkARequest.add(indexRequest);
+                    }else{
+                        //删除
+                        DeleteRequest deleteRequest = new DeleteRequest(eSBluk.getIndexName(), eSBluk.getType(), eSBluk.getId());
+                        bulkARequest.add(deleteRequest);
+                    }
+
+                }
+                BulkResponse bulkResponse = null;
+                try {
+                    bulkResponse = rhlClient.bulk(bulkARequest, RequestOptions.DEFAULT);
+                    if(bulkResponse.hasFailures()){
+                        logger.warn("bulk failure[{}]", bulkResponse.buildFailureMessage());
+                    }
+                    logger.info("writeData: {}" , JSON.toJSONString(bulkResponse));
+
+                } catch (IOException e) {
+                    logger.info("writeData error :{}" ,e);
+                }
+                logger.info("doBulk, the blukList size is {} and cost {} ms,", blukList.size(), System.currentTimeMillis() - begin);
+                Thread.sleep(50);
+            } else {
+                Thread.sleep(1000);
+            }
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
+        }
+    }
+
+    public void add(ESBluk esBluk) {
+        try {
+            queue.put(esBluk);
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
+        }
+    }
+
+}

+sync.es.flag=false

@@ -27,5 +27,8 @@ elasticSearch:
     connectNum: 10
     connectNum: 10
     connectPerRoute: 50
     connectPerRoute: 50
 
 
+zk:
+  connectString: 192.168.102.45:2181
+
 logging:
 logging:
   config: classpath:logback-boot.xml
   config: classpath:logback-boot.xml

@@ -29,3 +29,6 @@ elasticSearch:
   client:
   client:
     connectNum: 10
     connectNum: 10
     connectPerRoute: 50
     connectPerRoute: 50
+
+zk:
+  connectString: 192.168.104.246:2181