日志中心迁移elasticsearch实现

日志中心迁移elasticsearch
1.引入依赖
pom.xml

    <dependency>
       <groupId>org.elasticsearch</groupId>
       <artifactId>elasticsearch</artifactId>
       <version>6.6.0</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
        <version>6.6.0</version>
    </dependency>

2、ElasticsearchCmbsConfig装配RestHighLevelClient

import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 类作用: Elasticsearch配置类
 * 类名称: ElasticsearchMonitorConfig
 * 类描述: Elasticsearch配置类
 */
@Configuration
public class ElasticsearchMonitorConfig {
    @Autowired
    private ElasticsearchMonitorProperties esProperties;

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        String[] urlArr = esProperties.getEsAddress().split(",");
        HttpHost[] httpPostArr = new HttpHost[urlArr.length];
        for (int i = 0; i < urlArr.length; i++) {
            HttpHost httpHost = new HttpHost(urlArr[i].split(":")[0].trim(),
                    Integer.parseInt(urlArr[i].split(":")[1].trim()), "http");
            httpPostArr[i] = httpHost;
        }
        RestClientBuilder builder = RestClient.builder(httpPostArr);

        // 异步httpclient连接延时配置
        builder.setRequestConfigCallback(requestConfigBuilder -> {
            requestConfigBuilder.setConnectTimeout(esProperties.getConnectTimeOut());
            requestConfigBuilder.setSocketTimeout(esProperties.getSocketTimeOut());
            requestConfigBuilder.setConnectionRequestTimeout(esProperties.getConnectionRequestTimeOut());
            return requestConfigBuilder;
        });

        // 异步httpclient配置
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            // httpclient连接数配置
            httpClientBuilder.setMaxConnTotal(esProperties.getMaxConnectNum());
            httpClientBuilder.setMaxConnPerRoute(esProperties.getMaxConnectPerRoute());
            return httpClientBuilder;
        });
        return new RestHighLevelClient(builder);
    }
}

3、ElasticsearchMonitorProperties

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Component
@ConfigurationProperties(prefix = "elasticsearch")
@Data
public class ElasticsearchMonitorProperties {

    /**
     *  esAddress.
     */
    private String esAddress;

    /**
     * 连接超时时间
     */
    private int connectTimeOut;
    /**
     * 连接超时时间
     */
    private int socketTimeOut;
    /**
     * 获取连接的超时时间
     */
    private int connectionRequestTimeOut;
    /**
     * 最大连接数
     */
    private int maxConnectNum;
    /**
     * 最大路由连接数
     */
    private int maxConnectPerRoute;
}

4、application.yml

elasticsearch:
  esAddress: 127.0.0.1:9200
  connectTimeOut: 3000
  socketTimeOut: 30000
  connectionRequestTimeOut: 3000
  maxConnectNum: 30
  maxConnectPerRoute: 10

5、后端代码交互
restHighLevelClient.indexAsync
restHighLevelClient.bulkAsync

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * 类作用: Elasticsearch客户端模板
 * 类名称: ElasticsearchTemplate
 * 类描述: Elasticsearch客户端模板
 */
@Slf4j
@Component
public class ElasticsearchTemplate {
    @Autowired
    RestHighLevelClient restHighLevelClient;

    public void createIndex(String indexName, Object data) {
        log.info("ElasticsearchTemplate | createIndex | indexName : {}", indexName);
        XContentBuilder sourceBuilder = null;
        try {
            // 7X之后版本可以不需要"_doc"
            IndexRequest indexRequest = new IndexRequest(indexName,"_doc");
            // 构造入参执行请求
            sourceBuilder = constructIndexRequest(indexName, data, indexRequest);
            ActionListener<IndexResponse> listener = new ActionListener<IndexResponse>() {
                @Override
                public void onResponse(IndexResponse indexResponse) {

                }

                @Override
                public void onFailure(Exception e) {
                    log.error("ElasticsearchTemplate.createIndex.onFailure error : {}",
                            ExceptionUtils.getFullStackTrace(e));                    
                }
            };
            restHighLevelClient.indexAsync(indexRequest, RequestOptions.DEFAULT, listener);
        } catch (Exception e) {
            log.error("ElasticsearchTemplate.createIndex data : {}, error : {}",
                    data.toString(), ExceptionUtils.getFullStackTrace(e));
        } finally {
            closeSourceBuilder(sourceBuilder);
        }
    }

    public void bulkPutIndex(String indexName, List<BossInsServiceLogDtl> bossLogDtlList) {
        log.info("ElasticsearchTemplate | bulkPutIndex | indexName : {}", indexName);
        List<XContentBuilder> builderList = null;
        try {
            // 7X版本之后不需要"_doc"
            BulkRequest request = new BulkRequest(indexName,"_doc");
            builderList = constructBulkRequest(bossLogDtlList, request, indexName);
            ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
                @Override
                public void onResponse(BulkResponse indexResponse) {

                }
                @Override
                public void onFailure(Exception e) {
                    log.error("ElasticsearchTemplate.bulkPutIndex onFailure, error : {}",
                            ExceptionUtils.getFullStackTrace(e));
                }
            };
            restHighLevelClient.bulkAsync(request, RequestOptions.DEFAULT, listener);
        } catch (Exception e) {
            log.error("ElasticsearchTemplate.bulkIndex data : {}, error : {}",
                    bossLogDtlList, ExceptionUtils.getFullStackTrace(e));
        } finally {
            closeBtachSourceBuilder(builderList);
        }
    }

    public XContentBuilder constructIndexRequest(String indexName, Object data, IndexRequest indexRequest)
            throws IOException {
        XContentBuilder sourceBuilder = XContentFactory.jsonBuilder().startObject();
        String[] fieldNames = getFiledName(data);
        for(int j=0 ; j<fieldNames.length ; j++){           
            sourceBuilder.field(EsLogFieldEnum.getEsFieldByFieldName(fieldNames[j]),
                    getFieldValueByName(fieldNames[j], data));
        }
        sourceBuilder.endObject();
        indexRequest.source(sourceBuilder);
        return sourceBuilder;
    }

    /**
     * 获取属性名数组
     * */
    private static String[] getFiledName(Object o){
        Field[] fields=o.getClass().getDeclaredFields();
        String[] fieldNames=new String[fields.length];
        for(int i=0;i<fields.length;i++){
            fieldNames[i]=fields[i].getName();
        }
        return fieldNames;
    }

    /* 根据属性名获取属性值
     * */
    private static Object getFieldValueByName(String fieldName, Object o) {
        try {
            String firstLetter = fieldName.substring(0, 1).toUpperCase();
            String getter = "get" + firstLetter + fieldName.substring(1);
            Method method = o.getClass().getMethod(getter, new Class[] {});
            Object value = method.invoke(o, new Object[] {});
            // 如果是时间,指定时间格式yyyy-MM-dd HH:mm:ss
            if (value instanceof Date && null != value) {
                value = formatDate((Date)value, "yyyy-MM-dd HH:mm:ss");
            }
            return value;
        } catch (Exception e) {
            return null;
        }
    }

    public static String formatDate(Date date, String pattern) {
        if (null == date) {
            return null;
        }
        // 建立日期FORMAT的实例
        SimpleDateFormat df = new SimpleDateFormat(pattern);
        return df.format(date);
    }
   
    public List<XContentBuilder> constructBulkRequest(
            List<BossInsServiceLogDtl> bossLogDtlList, BulkRequest request, String indexName) throws IOException {
        List<XContentBuilder> builderList = new ArrayList<XContentBuilder>();
        for (BossInsServiceLogDtl bossDtl : bossLogDtlList) {
            IndexRequest indexRequest = new IndexRequest(indexName,"_doc");
            // 构造入参执行请求
            XContentBuilder sourceBuilder = XContentFactory.jsonBuilder().startObject();
            String[] fieldNames = getFiledName(bossDtl);
            for(int j=0 ; j<fieldNames.length ; j++){
                if (!fieldNames[j].equals("id")) {
                    sourceBuilder.field(EsLogFieldEnum.getEsFieldByFieldName(fieldNames[j]),
                            getFieldValueByName(fieldNames[j], bossDtl));
                }
            }
            sourceBuilder.endObject();
            indexRequest.source(sourceBuilder);
            builderList.add(sourceBuilder);
            request.add(indexRequest);
        }
        return builderList;
    }

    public void closeBtachSourceBuilder(List<XContentBuilder> tempBuilderList) {
        if (CollectionUtils.isNotEmpty(tempBuilderList)) {
            try {
                for (XContentBuilder sourceBuilder : tempBuilderList) {
                    if (null != sourceBuilder) {
                        sourceBuilder.close();
                    }
                }
            } catch (Exception e) {
                log.error("ElasticsearchTemplate.closeBtachSourceBuilder:批量释放资源完毕.error : {}",
                        ExceptionUtils.getFullStackTrace(e));
            }
        }
    }

    public void closeSourceBuilder(XContentBuilder builder) {
        try {
            if (null != builder) {
                builder.close();
            }
        } catch (Exception e) {
            log.error("ElasticsearchTemplate.closeSourceBuilder:释放资源完毕.error : {}",
                    ExceptionUtils.getFullStackTrace(e));
        }
    }
}

6、EsLogFieldEnum

import java.util.HashMap;
import java.util.Map;

/**
 * 项目名称:
 * 类名称:  EsLogFieldEnum.java
 * 类描述: 	日志表字段枚举
 */
public enum EsLogFieldEnum {
	USER_ID("userId","user_id"),
	ORDER_ID("orderId","order_id"),
	USER_ACCOUNT("userAccount","user_account"),
	REQUEST_MSG("requestMsg","request_msg"),
	RESPONSE_MSG("responseMsg","response_msg"),
	SERVICE_NAME("serviceName","service_name"),
	START_TIME("startTime","start_time"),
	END_TIME("endTime","end_time"),
	COST_TIME("costTime","cost_time"),
	METHOD_NAME("methodName","method_name"),
	METHOD_STATE("methodState","method_state"),
	ERR_DETAILS("errDetails","err_details"),
	PRODUCT_LINE("productLine","product_line"),
	PORTAL_TYPE("portalType","portal_type"),
	PLATFORM("platform","platform"),
	GMT_CREATE("gmtCreate","gmt_create"),
	GMT_MODIFIED("gmtModified","gmt_modified"),
	COMPANY_ID("companyId","company_id"),
	REPROCESS_FLAG("reprocessFlag","reprocess_flag"),
	MSG_KEY("msgKey","msg_key"),
	MESSAGE("message","message"),
	ERROR_CODE("errorCode","error_code"),
	ERROR_DETAIL_MESSAGE("errorDetailMessage","error_detail_message"),
	MESSAGE_SOURCE("messageSource","message_source"),
	MESSAGE_TYPE("messageType","message_type"),
	MESSAGE_TYPE_NAME("messageTypeName","message_type_name"),
	IS_PROCESSED("isProcessed","is_processed"),
	IS_NEED_FOLLOWED("isNeedFollowed","is_need_followed"),
	TOPIC_PARTITION("topicPartition","topic_partition"),
	TOPIC_OFFSET("topicOffset","topic_offset"),
	REMARKS("remarks","remarks"),
	TOPIC("topic","topic"),
	MSISDN("msisdn","msisdn"),
	OPER_CODE("operCode","oper_code"),
	OPER_TIME("operTime","oper_time"),
	SP_CODE("spCode","sp_code"),
	SERVICE_CODE("serviceCode","service_code"),
	TRANSACTION_ID("transactionId","transaction_id"),
	EXTENSION_1("extension1","extension_1"),
	EXTENSION_2("extension2","extension_2"),
	EXTENSION_3("extension3","extension_3"),
	EXTENSION_4("extension4","extension_4"),
	EXTENSION_5("extension5","extension_5"),
	USER_KEY("userKey","user_key"),
	PROCESS_CODE("processCode","process_code");

	private EsLogFieldEnum(String fieldName, String esField) {
		this.fieldName = fieldName;
		this.esField = esField;
	}

	private String fieldName;

	private String esField;

	public String getFieldName() {
		return this.fieldName;
	}

	public String getEsField() {
		return this.esField;
	}

	// 在加载初始化Map
  	private static final Map<String, EsLogFieldEnum> LOOKUP = new HashMap<>();
  	static{
  		for(EsLogFieldEnum esLogFieldEnum : EsLogFieldEnum.values()){
  			LOOKUP.put(esLogFieldEnum.getFieldName(), esLogFieldEnum);
  		}
  	}

    //反向查找的方法,供外部调用
   	public static String getEsFieldByFieldName(String fieldName) {
   		return null == LOOKUP.get(fieldName)
   				? fieldName : LOOKUP.get(fieldName).getEsField();
   	}

}

匿名

发表评论

匿名网友