分布式日志ELK+Kafka的配置与基本使用&Exception thrown when sending a message with key= and payload=‘1‘ to topic te

一、 分布式日志介绍

二、ELK+Kafka日志组成

1.Elasticsearch 2.Kibana 3.Logstash 4.Kafka：消息中间件的一种 （消息中间件参考我的另一篇博文https://blog.csdn.net/m0_45899013/article/details/106994409：） 高吞吐：处理消息的能力强，比RabbitMQ和ActiveMQ处理消息的能力强。

三、ELK+Kafka日志收集的原理

即：logstash从kafka拿到日志发给es，kibana来展示。

四、运行虚拟机Ubuntu，有镜像的前提下，创建ELK+Kafka的四个容器

docker run -d --name kgc_kafka -p 9092:9092 --network kgc_net --network-alias kafka kgc/kafka docker run -d --name kgc_elasticsearch -p 9200:9200 --network kgc_net --network-alias elasticsearch kgc/elasticsearch docker run -d -it --name kgc_kibana -p 5601:5601 --network kgc_net --network-alias kibana kgc/kibana docker run -d -it --name kgc_logstash --network kgc_net --network-alias logstash kgc/logstash

登录portainer容器检查：

五、修改配置文件

1.kafka 配置文件 server.properties 修改 打开本地文件dockerfiles-master文件，src，kafka，查看配置文件server.properties，定位需要改动的地方。查看Dockerfile，定位配置文件所在位置，到虚拟机上进行修改

vi /usr/local/kafka_2.10-0.10.2.1/config/server.properties

2. elasticsearch.yml 不用修改 3. kibana.yml 不用修改 4. logstash.conf 和 logstash.yml 修改

vi /usr/local/logstash-6.3.0/bin/logstash.conf

vi /usr/local/logstash-6.3.0/config/logstash.yml

5.修改完毕，重启4个容器。restart。 6.访问kibana

六、父级maven项目shop-common编写Kafka配置类、测试类等

参考语法： 1.父级maven项目shop-common编写KafkaConfig、KafkaUtil类。

utils模块pom引入依赖：

<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> @Component @ConfigurationProperties(prefix = "kafkaConfig") public class KafkaConfig { private String moduleName; public static final class LogType{ public static String INFO="info"; public static String ERROR="error"; } public static String qgKey="qg"; public void setModuleName(String moduleName){ this.moduleName = moduleName; } public String getModuleName() { return moduleName; } } @Component public class KafkaUtil { @Autowired private KafkaTemplate kafkaTemplate; @Autowired private KafkaConfig kafkaConfig; //发送日志的正常方法 public String sendInfoMessage(String message) { try { kafkaTemplate.send(kafkaConfig.getModuleName() + "-" + KafkaConfig.LogType.INFO, KafkaConfig.qgKey, message); } catch (Exception e) { e.printStackTrace(); return "fail"; } return "success"; } //发送日志的异常方法 public String sendErrorMessage(Exception em) { try { Writer writer = new StringWriter(); PrintWriter pw = new PrintWriter(writer); em.printStackTrace(pw); kafkaTemplate.send(kafkaConfig.getModuleName() + "-" + KafkaConfig.LogType.ERROR, KafkaConfig.qgKey, writer.toString()); } catch (Exception e) { e.printStackTrace(); return "fail"; } return "success"; } }

编写完毕，刷新maven，install。 2.qg-user-provider配置文件添加配置

kafkaConfig.moduleName=user

3.qg-user-consumer配置文件添加配置

kafkaConfig.moduleName=user #kafka配置文件 spring.kafka.producer.bootstrap-servers=10.211.55.3:9092 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

4.qg-user-consumer编写测试类TestUserController

//用于测试kafka日志 @RestController public class TestUserController { @Autowired private KafkaUtil kafkaUtil; @RequestMapping(value = "/test") public String test(){ kafkaUtil.sendInfoMessage("this is a test function"); System.out.println("================================="); return "发送成功"; } }

七、在浏览器访问kibana并添加数据进行测试

1.添加进一条数据，执行。 2.添加索引，保存。 3.启动qg-user-provider，启动qg-user-consumer，访问http://localhost:8082/test，出现“测试成功”，成功。刷新kibana： 4.添加新索引后索引，选择@timestamp模式，create index pattern进行应用。到discover查看： 添加完毕怎么请求呢？持续更新。。。

八、异常解决 Exception thrown when sending a message with key= and payload=‘1’ to topic test

停掉服务，重启zookeeper。