✔️什么是分布式日志系统？
现在，很多应用都是集群部署的，一次请求会因为负载均衡而被路由到不同的服务器上面，这就导致一个应用的日志会分散在不同的服务器上面。
当我们要向通过日志做数据分析，问题排查的时候，就需要分别到每台机器上去查看日志，这样就太麻烦了。
于是就有了分布式日志系统，他可以做分布式系统中的日志的统一收集、存储及管理。并且提供好的可用性、扩展性。
比较主流的这类日志管理系统有ELK、Graylog、Apache Flume，还有很多类似的云产品，如阿里云的SLS。
一般来说，如果资金够就上SLS，不够就自建ELK。
实现分布式日志系统需要使用分布式系统的一些基本概念和技术，例如消息传递、数据复制和分布式一致性协议。在Java中实现分布式日志系统可以使用一些现有的框架和库，例如Apache Kafka或Logstash。
看一个Demo：
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.*;
/**
* 一个分布式日志系统的Demo、包括日志的收集、处理、存储和检索功能
*
*/
public class DistributedLogSystem {
public static void main(String[] args) {
// 创建日志收集器线程池
ExecutorService collectorPool = Executors.newFixedThreadPool(10);
// 创建Kafka生产者配置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 创建Kafka生产者
Producer<String, String> producer = new KafkaProducer<>(props);
// 启动日志收集器线程
for (int i = 0; i < 10; i++) {
final int threadId = i;
collectorPool.submit(() -> {
try {
// 模拟日志收集逻辑
for (int j = 0; j < 1000; j++) {
String topic = "logs";
String message = "Log message " + threadId + "_" + j;
ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
producer.send(record);
}
} catch (Exception e) {
e.printStackTrace();
}
});
}
// 关闭日志收集器线程池和Kafka生产者
collectorPool.shutdown();
try {
if (!collectorPool.awaitTermination(60, TimeUnit.SECONDS)) {
collectorPool.shutdownNow();
}
producer.close();
} catch (InterruptedException e) {
collectorPool.shutdownNow();
producer.close();
Thread.currentThread().interrupt();
}
}
}
上面这个Demo中，首先创建了一个线程池用于日志收集器线程，然后创建了一个Kafka生产者对象。接着，使用线程池启动了10个日志收集器线程，每个线程模拟日志收集逻辑，并将收集到的日志消息发送到Kafka中。最后，关闭了日志收集器线程池和Kafka生产者对象。