英文淘宝网站建设,做网站快速排名,淄博信息港聊天室网址,唐山快速建站公司一、背景介绍
在分布式的项目中#xff0c;各功能模块产生的日志比较分散#xff0c;同时为满足性能要求#xff0c;同一个微服务会集群化部署#xff0c;当某一次业务报错后#xff0c;如果不能确定产生的节点#xff0c;那么只能逐个节点去查看日志文件#xff1b;lo…一、背景介绍
在分布式的项目中各功能模块产生的日志比较分散同时为满足性能要求同一个微服务会集群化部署当某一次业务报错后如果不能确定产生的节点那么只能逐个节点去查看日志文件logback中RollingFileAppenderConsoleAppender这类同步化记录器也降低系统性能综上一些问题可能考虑采用ELK elasticsearchlogstashkibana配合消息中间件去异步采集统一展示去解决。
这里之所以要加入kafka是因为
如果直接利用logstash同步日志则每个节点都需要部署logstash且logstash会严重消耗性能、浪费资源当访问量特别高时产生的日志速度也会特别快kafka可以削峰限流、降低logstash的压力当logstash故障时消息可以存储到kafka中不会丢失。
二、 整体流程图 三、搭建kafkazk环境
1、创建文件夹
mkdir /usr/elklog/kafka2、在创建好的文件夹下创建文件docker-compose.yml
version: 2services:zookeeper:image: docker.io/bitnami/zookeeper:3.8ports:- 2181:2181environment:- ALLOW_ANONYMOUS_LOGINyesnetworks:- es_defaultkafka:image: docker.io/bitnami/kafka:3.2user: rootports:- 9092:9092environment:- ALLOW_PLAINTEXT_LISTENERyes- KAFKA_CFG_ZOOKEEPER_CONNECTzookeeper:2181- KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://192.168.3.22:9092 #这里替换为你宿主机IP或host,在集群下各节点会把这个地址注册到集群并把主节点的暴露给客户端不要注册localhost
# - KAFKA_CFG_LISTENERSPLAINTEXT://0.0.0.0:9092depends_on:- zookeepernetworks:- es_default
networks:es_default:name: es_default
# external: true
volumes:zookeeper_data:driver: localkafka_data:driver: local
3、在docker-compose.yml同级目录中输入启动命令
docker-compose up -d 这里用的是docker-compose方式安装安装之前需要先安装好docker和docker-compose
docker安装方式https://blog.csdn.net/qq_38639813/article/details/129384923
docker-compose安装方式https://blog.csdn.net/qq_38639813/article/details/129751441
四、搭建elk环境
1、拉取elk所需镜像
docker pull elasticsearch:7.10.1
docker pull kibana:7.10.1
docker pull elastic/metricbeat:7.10.1
docker pull elastic/logstash:7.10.12、创建文件夹
mkdir /usr/elklog/elk
mkdir /usr/elklog/elk/logstash
mkdir /usr/elklog/elk/logstash/pipelinemkdir /usr/elklog/elk/es
mkdir /usr/elklog/elk/es/data3、给data文件夹授权
chmod 777 /usr/elklog/elk/es/data4、在/usr/elklog/elk/logstash/pipeline中创建logstash.conf
logstash.conf文件作用是将kafka中的日志消息获取出来 再推送给elasticsearch
input {kafka {bootstrap_servers 192.168.3.22:9092 #kafka的地址替换为你自己的 client_id logstash auto_offset_reset latest consumer_threads 5topics [demoCoreKafkaLog,webapiKafkaApp] #获取哪些topic在springboot项目的logback-spring.xml中指定type demo #自定义
# codec json}
}output {stdout { }elasticsearch {hosts [http://192.168.3.22:9200] #es的地址index demolog-%{YYYY.MM.dd} #这里将会是创建的索引名后续 kibana将会用不同索引区别#user elastic#password changeme}
}
5、在/usr/elklog/elk中创建docker-compose.yml
version: 2services:elasticsearch:image: elasticsearch:7.10.1restart: alwaysprivileged: trueports:- 9200:9200- 9300:9300volumes:- /usr/elklog/elk/es/data:/usr/share/elasticsearch/dataenvironment:- discovery.typesingle-nodenetworks:- es_defaultkibana:image: kibana:7.10.1restart: alwaysprivileged: trueports:- 5601:5601environment:- ELASTICSEARCH_URLhttp://192.168.3.22:9200depends_on:- elasticsearchnetworks:- es_defaultmetricbeat:image: elastic/metricbeat:7.10.1restart: alwaysuser: rootenvironment:- ELASTICSEARCH_HOSTShttp://192.168.3.22:9200depends_on:- elasticsearch- kibanacommand: -E setup.kibana.host192.168.3.22:5601 -E setup.dashboards.enabledtrue -E setup.template.overwritefalse -E output.elasticsearch.hosts[192.168.3.22:9200] -E setup.ilm.overwritetruenetworks:- es_defaultlogstash:image: elastic/logstash:7.10.1restart: alwaysuser: rootvolumes:- /usr/elklog/elk/logstash/pipeline:/usr/share/logstash/pipeline/ depends_on:- elasticsearch- kibananetworks:- es_default
networks:es_default:driver: bridgename: es_default
6、启动服务
docker-compose up -d检验es是否安装成功http://192.168.3.22:9200 检验kibana是否安装成功192.168.3.22:5601
7、kibana设置中文
从容器中复制出kibana.yml修改该文件再复制回去重启容器
docker cp elk-kibana-1:/usr/share/kibana/config/kibana.yml kibana.yml在这个文件最后加上 i18n.locale: zh-CNdocker cp kibana.yml elk-kibana-1:/usr/share/kibana/config/kibana.yml重启kibana容器便可五、springboot代码
1、引入依赖
!-- Kafka资源的引入 --
dependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactId
/dependency
dependencygroupIdcom.github.danielwegener/groupIdartifactIdlogback-kafka-appender/artifactIdversion0.2.0-RC1/version
/dependency
dependencygroupIdnet.logstash.logback/groupIdartifactIdlogstash-logback-encoder/artifactIdversion6.4/version
/dependency2、创建KafkaOutputStream
package com.elk.log;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.Charset;public class KafkaOutputStream extends OutputStream {Producer logProducer;String topic;public KafkaOutputStream(Producer producer, String topic) {this.logProducer producer;this.topic topic;}Overridepublic void write(int b) throws IOException {this.logProducer.send(new ProducerRecord(this.topic, b));}Overridepublic void write(byte[] b) throws IOException {this.logProducer.send(new ProducerRecordString, String(this.topic, new String(b, Charset.defaultCharset())));}Overridepublic void flush() throws IOException {this.logProducer.flush();}
}
3、创建KafkaAppender
package com.elk.log;import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.Layout;
import ch.qos.logback.core.OutputStreamAppender;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.util.StringUtils;import java.io.OutputStream;
import java.util.Properties;public class KafkaAppenderE extends OutputStreamAppenderE {private Producer logProducer;private String bootstrapServers;private LayoutE layout;private String topic;public void setLayout(LayoutE layout) {this.layout layout;}public void setBootstrapServers(String bootstrapServers) {this.bootstrapServers bootstrapServers;}public void setTopic(String topic) {this.topic topic;}Overrideprotected void append(E event) {if (event instanceof ILoggingEvent) {String msg layout.doLayout(event);ProducerRecordString, String producerRecord new ProducerRecord(topic, 0,((ILoggingEvent) event).getLevel().toString(), msg);logProducer.send(producerRecord);}}Overridepublic void start() {if (StringUtils.isEmpty(topic)) {topic Kafka-app-log;}if (StringUtils.isEmpty(bootstrapServers)) {bootstrapServers localhost:9092;}logProducer createProducer();OutputStream targetStream new KafkaOutputStream(logProducer, topic);super.setOutputStream(targetStream);super.start();}Overridepublic void stop() {super.stop();if (logProducer ! null) {logProducer.close();}}//创建生产者private Producer createProducer() {synchronized (this) {if (logProducer ! null) {return logProducer;}Properties props new Properties();props.put(bootstrap.servers, bootstrapServers);//判断是否成功我们指定了“all”将会阻塞消息 0.关闭 1.主broker确认 -1all.所在节点都确认props.put(acks, 0);//失败重试次数props.put(retries, 0);//延迟100ms100ms内数据会缓存进行发送props.put(linger.ms, 100);//超时关闭连接//props.put(connections.max.idle.ms, 10000);props.put(batch.size, 16384);props.put(buffer.memory, 33554432);//该属性对性能影响非常大如果吞吐量不够消息生产过快超过本地buffer.memory时将阻塞1000毫秒等待有空闲容量再继续props.put(max.block.ms,1000);props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer);props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer);return new KafkaProducerString, String(props);}}}
4、创建logback-spring.xml放到application.yml的同级目录
?xml version1.0 encodingUTF-8?
configuration scantrue scanPeriod60 seconds!-- include resourceorg/springframework/boot/logging/logback/base.xml/--logger namecom.elk levelinfo/!-- 定义日志文件 输入位置 --property namelogPath valuelogs /
!-- property namelogPath valueD:/logs/truckDispatch /--!-- 控制台输出日志 --appender nameSTDOUT classch.qos.logback.core.ConsoleAppenderencoderpattern%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger -%msg%n/patterncharset classjava.nio.charset.CharsetUTF-8/charset/encoder/appender!-- INFO日志文件 --appender nameinfoAppender classch.qos.logback.core.rolling.RollingFileAppenderfilter classch.qos.logback.classic.filter.LevelFilterlevelINFO/levelonMatchACCEPT/onMatchonMismatchDENY/onMismatch/filterrollingPolicy classch.qos.logback.core.rolling.TimeBasedRollingPolicy!-- 文件名称 --fileNamePattern${logPath}\%d{yyyyMMdd}\info.log/fileNamePattern!-- 文件最大保存历史天数 --MaxHistory30/MaxHistory/rollingPolicyencoderpattern%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n/patterncharset classjava.nio.charset.CharsetUTF-8/charset/encoder/appender!-- DEBUG日志文件 --appender namedebugAppender classch.qos.logback.core.rolling.RollingFileAppenderfilter classch.qos.logback.classic.filter.LevelFilterlevelDEBUG/levelonMatchACCEPT/onMatchonMismatchDENY/onMismatch/filterrollingPolicy classch.qos.logback.core.rolling.TimeBasedRollingPolicy!-- 文件名称 --fileNamePattern${logPath}\%d{yyyyMMdd}\debug.log/fileNamePattern!-- 文件最大保存历史天数 --MaxHistory30/MaxHistory/rollingPolicyencoderpattern%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n/patterncharset classjava.nio.charset.CharsetUTF-8/charset/encoder/appender!-- WARN日志文件 --appender namewarnAppender classch.qos.logback.core.rolling.RollingFileAppenderfilter classch.qos.logback.classic.filter.LevelFilterlevelWARN/levelonMatchACCEPT/onMatchonMismatchDENY/onMismatch/filterrollingPolicy classch.qos.logback.core.rolling.TimeBasedRollingPolicy!-- 文件名称 --fileNamePattern${logPath}\%d{yyyyMMdd}\warn.log/fileNamePattern!-- 文件最大保存历史天数 --MaxHistory30/MaxHistory/rollingPolicyencoderpattern%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n/patterncharset classjava.nio.charset.CharsetUTF-8/charset/encoder/appender!-- ERROR日志文件 --appender nameerrorAppender classch.qos.logback.core.rolling.RollingFileAppenderfilter classch.qos.logback.classic.filter.LevelFilterlevelERROR/levelonMatchACCEPT/onMatchonMismatchDENY/onMismatch/filterrollingPolicy classch.qos.logback.core.rolling.TimeBasedRollingPolicy!-- 文件名称 --fileNamePattern${logPath}\%d{yyyyMMdd}\error.log/fileNamePattern!-- 文件最大保存历史天数 --MaxHistory30/MaxHistory/rollingPolicyencoderpattern%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n/patterncharset classjava.nio.charset.CharsetUTF-8/charset/encoder/appender!-- 往kafka推送日志 --appender namekafkaAppender classcom.elk.log.KafkaAppender!-- kafka地址 --bootstrapServers192.168.3.22:9092/bootstrapServers!-- 配置topic --topicdemoCoreKafkaLog/topic!-- encoder负责两件事一是将一个event事件转换成一组byte数组二是将转换后的字节数据输出到文件中 --encoderpattern${HOSTNAME} %date [%thread] %level %logger{36} [%file : %line] %msg%n/patterncharsetutf8/charset/encoder!-- layout主要的功能就是将一个event事件转化为一个String字符串 --layout classch.qos.logback.classic.PatternLayoutpattern${HOSTNAME} %date [%thread] %level %logger{36} [%file : %line] %msg%n/pattern/layout/appender!-- 指定这个包的日志级别为error --logger nameorg.springframework additivityfalselevel valueERROR /!-- 控制台输出 --
!-- appender-ref refSTDOUT /--appender-ref referrorAppender //logger!-- 由于启动的时候以下两个包下打印debug级别日志很多 所以调到ERROR--!-- 指定这个包的日志级别为error --logger nameorg.apache.tomcat.util additivityfalselevel valueERROR/!-- 控制台输出 --
!-- appender-ref refSTDOUT/--appender-ref referrorAppender//logger!-- 默认spring boot导入hibernate很多的依赖包启动的时候会有hibernate相关的内容直接去除 --!-- 指定这个包的日志级别为error --logger nameorg.hibernate.validator additivityfalselevel valueERROR/!-- 控制台输出 --
!-- appender-ref refSTDOUT/--appender-ref referrorAppender//logger!-- 监控所有包日志输入到以下位置并设置日志级别 --root levelWARN!--INFO--!-- 控制台输出 --appender-ref refSTDOUT/!-- 这里因为已经通过kafka往es中导入日志所以就没必要再往日志文件中写入日志可以注释掉下面四个提高性能 --appender-ref refinfoAppender/appender-ref refdebugAppender/appender-ref refwarnAppender/appender-ref referrorAppender/appender-ref refkafkaAppender//root
/configuration
5、配置文件无需任何修改
server:tomcat:uri-encoding: UTF-8max-threads: 1000min-spare-threads: 30port: 8087connection-timeout: 5000msservlet:context-path: /6、编写测试类
package com.elk.log;import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;Slf4j
RestController
RequestMapping(/test)
public class TestController {GetMapping(/testLog)public String testLog() {log.warn(gotest);return ok;}GetMapping(/testLog1)public Integer testLog1() {int i 1/0;return i;}
}
六、利用kibana查看日志 注意这里的索引名字就是logstash.conf中创建的索引名出现这个也意味着整个流程成功 此时索引模式创建完毕我创建的索引模式名字是demo* 这时就可以看到日志了可以进一步调用测试接口去验证我这里不在展示至此全部完毕
