SLS日志服務(wù)的集成配置是怎樣的

SLS日志服務(wù)的集成配置是怎樣的，相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。

配置

pom.xml依賴

<!--        服務(wù)器-->
        <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>log-loghub-producer</artifactId>
            <version>0.1.4</version>
            <exclusions>
                <exclusion>
                    <groupId>com.alibaba</groupId>
                    <artifactId>fastjson</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>aliyun-log-producer</artifactId>
            <version>0.3.4</version>
        </dependency>
        <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>aliyun-log</artifactId>
            <version>0.6.33</version>
        </dependency>
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
            <version>2.5.0</version>
        </dependency>
        <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>loghub-client-lib</artifactId>
            <version>0.6.16</version>
        </dependency>

配置AliLogConfig

package com.yhzy.doudoubookserver.global.alilog;

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.producer.LogProducer;
import com.aliyun.openservices.log.producer.ProducerConfig;
import com.aliyun.openservices.log.producer.ProjectConfig;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

/**
 * @author zhangqinghe
 * @version 1.0.0
 * @email her550@dingtalk.com
 * @date 2020/12/23
 * @since 1.0.0
 */
@Configuration
@Scope("singleton")
public class AliLogConfig {
    public static String accessKeyId = "";
    public static String accessKeySecret = "";
    public static String endPoint = "";
    public static String projectName = SLSEnvironment.BOOK_PROJECT;

    @Bean
    @ConditionalOnClass(LogProducer.class)
    public LogProducer getLogProducer() {
        LogProducer producer = new LogProducer(producerConfig());
        producer.setProjectConfig(projectConfig());
        return producer;
    }

    @Bean
    @ConditionalOnClass(ProducerConfig.class)
    public ProducerConfig producerConfig() {
        ProducerConfig producerConfig = new ProducerConfig();
        //被緩存起來的日志的發(fā)送超時時間，如果緩存超時，則會被立即發(fā)送，單位是毫秒
        producerConfig.packageTimeoutInMS = 1000;
        //每個緩存的日志包的大小的上限，不能超過5MB，單位是字節(jié)
        producerConfig.logsBytesPerPackage = 5 * 1024 * 1024;
        //每個緩存的日志包中包含日志數(shù)量的最大值，不能超過4096
        producerConfig.logsCountPerPackage = 4096;
        //單個producer實例可以使用的內(nèi)存的上限，單位是字節(jié)
        producerConfig.memPoolSizeInByte = 1000 * 1024 * 1024;
        //IO線程池最大線程數(shù)量，主要用于發(fā)送數(shù)據(jù)到日志服務(wù)
        producerConfig.maxIOThreadSizeInPool = 50;
        //當使用指定shardhash的方式發(fā)送日志時，這個參數(shù)需要被設(shè)置，否則不需要關(guān)心。后端merge線程會將映射到同一個shard的數(shù)據(jù)merge在一起，而shard關(guān)聯(lián)的是一個hash區(qū)間，
        //producer在處理時會將用戶傳入的hash映射成shard關(guān)聯(lián)hash區(qū)間的最小值。每一個shard關(guān)聯(lián)的hash區(qū)間，producer會定時從loghub拉取，該參數(shù)的含義是每隔shardHashUpdateIntervalInMS毫秒，
        producerConfig.shardHashUpdateIntervalInMS = 10 * 60 * 1000;
        producerConfig.retryTimes = 3;
        return producerConfig;
    }

    @Bean
    @ConditionalOnClass(ProjectConfig.class)
    public ProjectConfig projectConfig() {
        return new ProjectConfig(projectName, endPoint, accessKeyId, accessKeySecret);
    }

    /**
     * 讀取sls對象 用于讀取數(shù)據(jù)
     * @return
     */
    @Bean
    public Client client(){
        String accessId = "";
        String accessKey = "";
        String host = "";
        return new Client(host, accessId, accessKey);
    }
}

配置AliLogUtil

package com.yhzy.doudoubookserver.common;

import com.aliyun.openservices.log.common.LogItem;
import com.aliyun.openservices.log.producer.LogProducer;
import com.yhzy.doudoubookserver.global.alilog.AliLogConfig;
import com.yhzy.doudoubookserver.global.alilog.CallbackLogInfo;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Vector;

/**
 * @author zhangqinghe
 * @version 1.0.0
 * @email her550@dingtalk.com
 * @date 2020/12/23
 * @since 1.0.0
 */
@Component
public class AliLogUtil {

    @Resource
    private AliLogConfig aliLogConfig;

    public void saveLog(String projectName,String logStore, Vector<LogItem> logGroup, String topic, String source,Long millis) throws InterruptedException {
        final LogProducer logProducer = aliLogConfig.getLogProducer();
        // 并發(fā)調(diào)用 send 發(fā)送日志
        logProducer.send(projectName, logStore, topic, source, logGroup,
                new CallbackLogInfo(projectName, logStore, topic,null, source, logGroup, logProducer));
        //主動刷新緩存起來的還沒有被發(fā)送的日志
        logProducer.flush();
        //等待發(fā)送線程退出
        Thread.sleep(millis);
        //關(guān)閉后臺io線程，close會將調(diào)用時刻內(nèi)存中緩存的數(shù)據(jù)發(fā)送出去
        logProducer.close();
    }
}

配置CallbackLogInfo

package com.yhzy.doudoubookserver.global.alilog;

import com.aliyun.openservices.log.common.LogItem;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.producer.ILogCallback;
import com.aliyun.openservices.log.producer.LogProducer;
import com.aliyun.openservices.log.response.PutLogsResponse;

import java.util.Vector;

/**
 * @author zhangqinghe
 * @version 1.0.0
 * @email her550@dingtalk.com
 * @date 2020/12/23
 * @since 1.0.0
 */
public class CallbackLogInfo extends ILogCallback {
    // 保存要發(fā)送的數(shù)據(jù)，當時發(fā)生異常時，進行重試
    public String project;
    public String logstore;
    public String topic;
    public String shardHash;
    public String source;
    public Vector<LogItem> items;
    public LogProducer producer;
    public int retryTimes = 0;

    public CallbackLogInfo(String project, String logstore, String topic, String shardHash, String source,
                           Vector<LogItem> items, LogProducer producer) {
        super();
        this.project = project;
        this.logstore = logstore;
        this.topic = topic;
        this.shardHash = shardHash;
        this.source = source;
        this.items = items;
        this.producer = producer;
    }

    public void onCompletion(PutLogsResponse response, LogException e) {
        if (e != null) {
            // 打印異常
            System.out.println(e.GetErrorCode() + ", " + e.GetErrorMessage() + ", " + e.GetRequestId());
            // 最多重試三次
            if (retryTimes++ < 3) {
                producer.send(project, logstore, topic, source, shardHash, items, this);
            }
        } else {
            //請求id
            System.out.println("send success, request id: " + response.GetRequestId());
        }
    }
}

配置SLSEnvironment

記錄sls日志相關(guān) project & logStore 參數(shù)名配置

package com.yhzy.doudoubookserver.global.alilog;


import java.time.LocalDateTime;
import java.time.ZoneOffset;

/**
 * sls日志相關(guān) project & logStore 參數(shù)名配置
 *
 * @author zhangqinghe
 * @version 1.0.0
 * @email her550@dingtalk.com
 * @date 2020/12/25
 * @since 1.0.0
 */
public interface SLSEnvironment {
    /** 測試project*/
    String TEST_PROJECT = "test_project";

    /** 測試logStore*/
    String TEST_LOGSTORE = "test_logstore"; 
    
    /** 開始時間*/
    public static Integer FROM() {
        return Math.toIntExact(LocalDateTime.now().plusDays(-3).withHour(0).withMinute(0).withSecond(0).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli() / 1000);
    }
    /** 結(jié)束時間*/
    public static Integer TO() {
        return Math.toIntExact(LocalDateTime.now().plusDays(1).withHour(23).withMinute(59).withSecond(59).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli() / 1000);
    }
    
    /** 最小開始時間    該時間為2020年12月1日的unix時間戳*/
    public static Integer MINFROM(){
        return 1606752000;
    }
    /** 最大結(jié)束時間    該時間為當前時間的unix時間戳*/
    public static Integer MAXTO() {
        return Math.toIntExact((System.currentTimeMillis()+86400000) / 1000);
    }

}

讀取寫入測試AliLogTest

讀取es中的數(shù)據(jù)寫入到sls中

package com.yhzy;

/**
 * @author zhangqinghe
 * @version 1.0.0
 * @email her550@dingtalk.com
 * @date 2020/12/24
 * @since 1.0.0
 */
import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.*;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.response.*;
import com.yhzy.doudoubookserver.common.AliLogUtil;
import com.yhzy.doudoubookserver.global.alilog.SLSEnvironment;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;

/**
 * @author zhangqinghe
 * @version 1.0.0
 * @email her550@dingtalk.com
 * @date 2020/12/23
 * @since 1.0.0
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = DoudouSLSApplication.class)
public class AliLogTest {
    @Resource
    private AliLogUtil aliLogUtil;
    @Resource
    private RestHighLevelClient restHighLevelClient;

    /**
     * 讀取分析測試
     * @throws LogException
     */
    @Test
    public void read() throws LogException {
        String accessId = "";
        String accessKey = "";
        String host = "";
        String project = "zqhtest";
        String logStore = "test_logstore";

        Client client = new Client(host, accessId, accessKey);
        long time = new Date().getTime()/1000;
        /*
         * project : Project名稱。
         * logStore : LogStore名稱。
         * from : 查詢開始時間點。Unix時間戳格式，表示從1970-1-1 00:00:00 UTC計算起的秒數(shù)。
         * to : 查詢結(jié)束時間點。Unix時間戳格式，表示從1970-1-1 00:00:00 UTC計算起的秒數(shù)。
         * topic : 日志主題。
         * query : 查詢分析語句。
               具體查看 https://help.aliyun.com/document_detail/53608.html?spm=a2c4g.11186623.2.17.56322e4acEkU4X#concept-nyf-cjq-zdb
         * line : 請求返回的最大日志條數(shù)。最小值為0，最大值為100，默認值為100。  沒用明白呢...這個和query中的limit是兩個東西？？？？
         * offset : 查詢開始行。默認值為0。
         * reverse : 是否按日志時間戳逆序返回日志，精確到分鐘級別。默認值為false。
               true：按照逆序返回日志。
               false：按照順序返回日志。
         */
        GetLogsResponse getLogsResponse = client.GetLogs(
                project,
                logStore,
                (int) (time - 36000000),
                (int) time,
                "",
                "* | select app_version,channel_id,content_id,COUNT(content_id) contentCount,COUNT(content_id) chapterCount, sum(time) timeSum GROUP BY app_version,channel_id,content_id LIMIT 100010",
                25,
                0,
                false);

        for (QueriedLog getLog : getLogsResponse.GetLogs()) {
            System.out.println(getLog.GetLogItem().ToJsonString());
        }
    }

    /**
     * 采集數(shù)據(jù)測試
     *  讀取es中的數(shù)據(jù),寫入到sls中
     * @throws IOException
     */
    @Test
    public void test1() throws IOException {

        Vector<LogItem> logItems = new Vector<>();
        //讀取es中的書籍閱讀數(shù)據(jù)
        SearchRequest request = new SearchRequest("bookanalysis");

        long start = LocalDateTime.now().plusDays(-1).withHour(11).withMinute(0).withSecond(0).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli();
        long end = LocalDateTime.now().plusDays(-1).withHour(12).withMinute(0).withSecond(0).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli();

        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        List<QueryBuilder> filter = boolQuery.filter();
        //時間區(qū)間條件
        filter.add(rangeQuery(start, end, "create_time"));
        //這里拼接動態(tài)條件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //添加條件
        searchSourceBuilder.query(boolQuery);
        request.source(searchSourceBuilder.trackTotalHits(true).size(15000));
        SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
        for (SearchHit hit : response.getHits().getHits()) {
            Map<String, Object> map = hit.getSourceAsMap();
            LogItem logItem = new LogItem();
            logItem.PushBack("app_version",map.get("app_version")==null?"1":map.get("app_version").toString());
            logItem.PushBack("book_id",map.get("book_id")==null?"1":map.get("book_id").toString());
            logItem.PushBack("book_name",map.get("book_name")==null?"1":map.get("book_name").toString());
            logItem.PushBack("category_id_1",map.get("category_id_1")==null?"1":map.get("category_id_1").toString());
            logItem.PushBack("category_id_2",map.get("category_id_2")==null?"1":map.get("category_id_2").toString());
            logItem.PushBack("channel_id",map.get("channel_id")==null?"1":map.get("channel_id").toString());
            logItem.PushBack("chapter_name",map.get("chapter_name")==null?"1":map.get("chapter_name").toString());
            logItem.PushBack("content_id",map.get("content_id")==null?"1":map.get("content_id").toString());
            logItem.PushBack("create_time",map.get("create_time")==null?"1":map.get("create_time").toString());
            logItem.PushBack("device_id",map.get("device_id")==null?"1":map.get("device_id").toString());
            logItem.PushBack("is_new",map.get("is_new")==null?"1":map.get("is_new").toString());
            logItem.PushBack("is_official",map.get("is_official")==null?"1":map.get("is_official").toString());
            logItem.PushBack("is_vip",map.get("is_vip")==null?"1":map.get("is_vip").toString());
            logItem.PushBack("log_type",map.get("log_type")==null?"1":map.get("log_type").toString());
            logItem.PushBack("position",map.get("position")==null?"1":map.get("position").toString());
            logItem.PushBack("sex",map.get("site")==null?"1":map.get("site").toString());
            logItem.PushBack("time",map.get("time")==null?"1":map.get("time").toString());
            logItem.PushBack("user_id",map.get("user_id")==null?"1":map.get("user_id").toString());
            logItems.add(logItem);
        }

        try {
            aliLogUtil.saveLog(SLSEnvironment.BOOK_PROJECT,SLSEnvironment.BOOK_LOGSTORE,logItems,"read","source ip",1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static RangeQueryBuilder rangeQuery(Long startTime, Long endTime, String name){
        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(name);
        if (startTime != null) {
            rangeQueryBuilder.gte(startTime);
        }
        if (endTime != null) {
            rangeQueryBuilder.lte(endTime);
        }
        return rangeQueryBuilder;
    }
}

看完上述內(nèi)容，你們掌握SLS日志服務(wù)的集成配置是怎樣的的方法了嗎？如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注億速云行業(yè)資訊頻道，感謝各位的閱讀！