分布式日志处理：ELK+Kafka实现日志收集

sudo sysctl -w vm.max_map_count=262144      	#设置内存！每次启动都要给个运行内存！
tar -zxvf elasticsearch-6.2.4.tar.gz		 	#指定目录下 解压
#解压完成,设置elasticsearch-6.2.4/config/目录下：elasticsearch.yml文件保存这一些配置信息,端口9200...

#linunx 创建一个组 Elasticsearch 不能使用Root用户，需要由一个独立用户！
groupadd elsearch	#创建一个组
useradd elsearch -g elsearch -p elasticsearch #设置密码

#解压elasticsearch中文安装包！
#unzip unzip 解压命令! zip压缩包解压,需要下载应用！ install 在线安装(需要网络！)！
yum install unzip	#下载unzip应用;
#解压目录下文件！
unzip /usr/local/elasticsearch-analysis-ik-6.2.4.zip

#授权用户操作elsearch  
chown -R elsearch:elsearch  elasticsearch-6.2.4
#创建一个目录, 把中午解压包放进去。 就可以显示中文了！ 
mkdir -p /usr/local/elasticsearch-6.2.4/plugins/analysis-ik/	#elasticsearch-6.2.4/plugins下创建一个文件analysis-ik
#中文解压文件,直接copy进去
cp -r /usr/local/elasticsearch/* /usr/local/elasticsearch-6.2.4/plugins/analysis-ik/

#Contos还需进行配置：修改 soft    nproc     4096
vi /etc/security/limits.d/20-nproc.conf		#修改 soft    nproc     4096 对应...

vi /etc/security/limits.conf  				#文件添加：↓↓
*               soft    nofile          65536
*               hard    nofile          65536
*               soft    nproc           4096
*               hard    nproc           4096
#授权：
chown -R elsearch:elsearch /usr/local/elasticsearch/

# 切换用户
su elsearch
# 刷新
source /etc/profile
# elasticsearch  /bin目录下启动 
./elasticsearch				#启动Elasticsearch

#解压
tar -zxvf logstash-6.3.0.tar.gz

#修改配置,logstash-6.3.0/bin/logstash.conf
#修改配置,logstash-6.3.0/config/logstash.yml
#启动...比较慢！ 
#/usr/local 是我的路径,注意要根据自己的来！
/usr/local/logstash-6.3.0/bin/logstash -f /usr/local/logstash-6.3.0/bin/logstash.conf

#解压
tar -zxvf kibana-6.2.4-linux-x86_64
#配置：Elasticsearch 的ip端口引用依赖！ /usr/local/kibana-6.2.4-linux-x86_64/config/kibana.yml 文件！
#更改ip

#bin目录启动：
#/usr/local/kibana-6.2.4-linux-x86_64/bin# 
./kibana	#启动

#root 用户下解压：
tar -zxvf kafka_2.10-0.10.2.1.tgz
#配置:kafka_2.10-0.10.2.1/config/server.properties文件,
#内置配置：监听器,将内部的默认ip端口更改为,linux 本机的ip 	0.0.0.0 指所有的ip都可以访问，但不建议更改！ 只需要把看到的IP都改了即可！

#zookeeper
/usr/local/kafka_2.10-0.10.2.1/bin/zookeeper-server-start.sh -daemon /usr/local/kafka_2.10-0.10.2.1/config/zookeeper.properties
#kafka
/usr/local/kafka_2.10-0.10.2.1/bin/kafka-server-start.sh  -daemon /usr/local/kafka_2.10-0.10.2.1/config/server.properties

#创建 producer(生产者): 测试生产消息，产生消息, 主题：topic
bin/kafka-console-producer.sh --broker-list localhost:9092 -topic test

#创建 consumer(消费者): 测试消费, 接受消息的 主题：topic
bin/kafka-console-consumer.sh --zookeeper localhost:2181 -topic test --from-beginning

<!-- 父依赖 -->
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.6.RELEASE</version>
    </parent>
<!-- web 和 kafka依赖！ -->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- 程序只需要向kafka写入日志,内部就会通过 Logstash——Elasticsearch——Kibana展示  -->
<dependency>
	<groupId>org.springframework.kafka</groupId>
	<artifactId>spring-kafka</artifactId>
</dependency>

spring:
  kafka:
    producer:
      bootstrap-servers: 192.168.1.110:9092                                     #kafka的服务地址!
      key-serializer: org.apache.kafka.common.serialization.StringSerializer    #消息编码方式Serializer 网络传输序列化操作！
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class TestRun {
    public static void main(String[] args) {
        SpringApplication.run(TestRun.class, args);
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class TestController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;      //kafkaTemplate模板对象！

    @GetMapping("/tolog")
    public String tolog() {
        //循环写日志!
        for (int i = 0; i < 10; i++) {
            //向kafka中写入日志...日志不一定是log 只是一种记录...别死板！学习最怕死板了...
            //                         消息主题         key                 日志数据value
            kafkaTemplate.send("dmservice", "dmservice", "kafa send message to elk--------------" + i);
            //消息主题与,logstash-6.3.0/bin/logstash.conf 中设置的 topics 主题匹配对应！！通过它进行匹配通信！
        }
        return "ok";
    }
}