本项目主要采用技术软件：centos7（三台）、nginx 、kafka、zookeeper 、mysql、celery、filebeat 、python。

该项目主要用于使用filebeat收集前端作为生产者nginx用户访问集群nginx web产生页面access日志，然后统一存储收集到的日志信息kafka然后编写平台python消费者程序获得的消费者程序nginx清洁日志，分析日志ip省、运营商、带宽等，获取流量信息后存储mysql在数据库中；并设置阈值，每分钟监控报警。

主要是为了防止运营商在某个省份的流量突然增加，导致服务器异常，起到监控和预警的作用。

分为几个部分：用户访问——>nginx负载均衡——>应用集群（nginx web服务）——>kafka集群（基于zookeeper）——>python消费者程序——>存入mysql数据库

总体详细流程说明(根据个人理解，如有误解，请帮我指出):
1、用户访问：通过域名或代理集群测试用户ip地址来访问web集群中web服务。

2、代理集群：在用户输入www.sc.com之后，dns它将给用户一个负载平衡，允许用户随机访问这两个负载平衡器上的任何一个。这两个负载平衡器将帮助代理转发（按某种算法转发）并转发给这三个nginx web任何一个服务器。

3、应用集群：这个时候web集群中的服务器上会产生日志，然后我们会安装在服务器上filebeat，filebeat它相当于生产者的角色，用于收集nginx web生成在服务器上的日志。

4、kafka集群：filebeta收集日志后，将数据发送到kakfa集群中，filebeat会去请求kafka集群中的任何一个broker（创建了topic）。因为如果互动的话broker里面不是partition的leader，那么follower将返回当前请求副本leader信息，然后filebeat再跟leader交互。这里所有的partition都加了同一个名字nginxlog的topic中。filebeat随机发送数据给一个partition中的leader之后，（partition中的ISR列表中的会将leader数据同步到其他数据follower上），然后leaer会返回一个ack信息，ack默认为1，leader收到消息后，向生产者发送响应，然后filebeat然后发送下一个数据，确保发送数据的一致性。

5、zookeeper：是用来管理kafka用于保存kafka的元信息、topic、partition、副本信息，通过抢占选举kafka的controller，这个controller用来管理kakfa副本partition中的leader和follower的选举。而zookeeper本身也是有leader和follower是的，它的选举方法是一致性算法（zab），按照少数服从多数、票数超过一半的原则选举leader。所以在zk集群中，机器存活数必须过半，集群才能正常使用，所以我们通常也会将zk为了方便选举，集群节点数设为奇数。跟filebeat给Kafka传输数据是一样的，Kafka连接任何一个zookeeper可以操作，但必须操作新的数据修改和其他事务leader上运行。

扩展：如果连接到客户端follower上进行事务操作，follower会返回给leader的IP，最终客户端还在leader上操作，但可直接连接follower进行查询操作。

5、消费者：然后我们需要创建一个python消费者程序(使用pymysql消费数据(所以没有消费组。如果有消费群体，同一消费群体中的消费者只能同时消费一个partition的数量。每个消费者可以同时消费一个消费群体topic不同分区的数据。），获取到kahka里面的数据之后，对数据进行清洗，清洗成我们需要的IP地址、时间、带宽等字段，然后通过淘宝接口IP地址分析为省和运营商。

6、**mysql数据块：**最后，通过淘宝的接口和pymysql存储在数据库中的模块。

用户直接访问测试时nginx web也可以，绑定域名访问。www.sc.com这个域名的分析放在里面Windows的host客户机需要在文件中使用。

dns(域名)也可以平衡负载，即通过域名(如www.sc.com）也可以分析成多个ip通过轮询分析地址ip。但是如果服务器挂了，dns这个地址不会马上删除，会继续分析成挂断ip，访问可能会失败。虽然客服端有重试，是还是会影响用户体验。

1、在应用集群 nginx web服务前面加反向代理。安全性也会高一点。负载均衡控制也会容易很多。
2、反向代理机中使用keepalived双vip互为主备做高可用，提高资源利用率。

代理集群：www.baidu.com 解析成两个虚拟ip：一个域名可以解析成两个虚拟ip。
假如：当上面一台一号代理机1.5master挂掉之后，下面二号代理机的1.5backup能运行，下一台二号代理机的1.6master也能正常运行；如果二号代理机1.6master挂掉了，也能使用一号代理机上1.6backup进行访问。

反向代理：代理的是服务器。
正向代理：代理的是客户机（VPN）。
反向代理是充当Web服务器网关的代理服务器。当用户将请求发送到使用反向代理的Web服务器时，他们将先转到反向代理，由该代理将确定是将其路由到Web服务器还是将其阻止。这意味着有了反向代理，我们永远不会与使用它的Web服务器进行直接通信。可以将它们看作web服务器或服务器集群的某种包装器。通过负载平衡和缓存，它们可以保护web免遭攻击，并提供更好的web性能。

反向代理完全是服务器端工程师的工作。反向代理它充当服务器的网关，除了通过它们，没有请求能够到达使用它们的服务器。就像用户需要隐私保护，而服务器也是如此的。但是，除了可以防御攻击之外，有些还可以利用反向代理通过负载平衡和缓存来提高其Web性能。

对于使用这种类型代理的服务器或一组服务器。我们将不知道有关实际服务器的任何信息。但是可以知道反向代理的ip地址。当用户使用反向代理向web服务发送请求的时候，用户的请求将会直接发送到这个反向代理服务器上面。

然后，代理会决定如何处理该请求，是将其中继到适当的服务器还是发回禁止的错误响应。如果它授权用户的请求，则它将保存用户的请求副本（包括IP地址），并将用户的请求中继到适当的服务器。

当响应被发送回时，它被发送到反向代理，反向代理将其转发给用户。所有这些都是在短时间内发生的，用户甚至都不会注意到。某些反向代理实际上可以缓存资源，这意味着，如果用户正在寻找的信息已经在缓存中，它不必浪费资源，试图再次获取它，它会直接发送给用户，而不打扰服务器。

反向代理服务器可以保护 nginx web服务免受各种类型的连接的侵害，或用作过滤、防火墙或其他安全性。

反向代理的3种必要应用程序之一——负载均衡，人们使用反向代理的最重要原因是负载平衡。

高流量的网站通常每分钟处理大量请求，这可能会减慢其系统性能并破坏响应时间。为了确保更好的用户体验和更快的响应，于是就使用了反向代理。它们将具有一组服务器，它们全部执行相同的功能。它没有使一台服务器响应所有请求，而是在服务器之间分配了请求，从而缩短了响应时间，并且不会使任何服务器工作过度。

代理集群中可用dns或者nginx反向代理做负载均衡。
在负载均衡中，当我们向某个服务器发送请求的时候，服务端可能会对请求做一个负载，将流量分发到不同的服务器。

负载均衡是高可用网络基础架构的关键组件，通常用于将工作负载分布到多个服务器来提高网站、应用、数据库或其他服务的性能和可靠性。

①一个没有负载均衡的 web 架构类似于下图：

在流程图中，我们可以看到用户是直接连接到web服务器，如果这个服务器宕机了，那么用户将无法访问到web服务器。此外，如果同时有很多用户试图访问web服务器，超过了其能处理的极限，就会出现加载速度缓慢或根本无法连接的情况。

②有负载均衡的web框架图如下所示:
通过在后端引入一个负载均衡器和至少一个额外的 web 服务器，就可以缓解这个故障。通常情况下，所有的后端服务器会保证提供相同的内容，以便用户无论哪个服务器响应，都能收到一致的内容。
为解决负载均衡器的单点故障问题，可以将第二个负载均衡器连接到第一个上，从而形成一个集群。这样的话当主负载均衡器发生了故障，就需要将用户请求转到第二个负载均衡器。因为 DNS 更改通常会较长的时间才能生效，因此需要能灵活解决 IP 地址重新映射的方法，比如浮动 IP（floating IP）。这样域名可以保持和相同的 IP 相关联，而 IP 本身则能在服务器之间移动。

负载均衡器可以处理HTTP、HTTPS、TCP、UDP四种请求。

应用集群：nginx web服务：仅仅做静态页面展示。在应用web前面加反向代理。安全性也会高一点。负载均衡控制也会容易很多。

用户通过域名或者代理集群中的某个ip地址来访问web集群中web服务的时候会产生nginx访问日志，日志存放在/var/log/nginx/sc_access，然后我们会在服务器上安装filebeat，filebeat相当于一个生产者的角色，用来收集nginx web服务器上的日志。

Filebeat是一个本地文件的日志数据采集器。在服务器上安装客户端后，filebeat会监控日志目录或者指定的日志文件，tail file，追踪读取这些文件（追踪文件的变化，不停的读），并且转发这些信息到MQ中间件，或者直接到elasticsearch或者logstarsh中存放，转发到kafka中进行索引 。

当开启Filebeat程序的时候，它会启动一个或多个探测器去检测指定的日志目录或文件。对于探测器找出的每一个日志文件，Filebeat会启动收集进程，每一个收集进程读取一个日志文件的内容，然后将这些日志数据发送到后台处理程序，后台处理程序会集合这些事件，最后发送集合的数据到output指定的目的地。

根据filebeat的yaml格式文件的配置：数据和输入和输出来获取信息和发送信息：

#ymal格式
{
“filebeat.inputs”: [ #输入
{ “type”:“log”,
“enabled”:true,
“paths”:[“/var/log/nginx/sc/access.log” #这里在下面做了修改
},
],
}

对filebeat的/etc/filebeat/filebeat.yml的配置文件进行修改。

其中输入（input）为：nginx日志的存放路径(也就是在获取nginx上的日志信息)：/var/log/nginx/sc/access.log；
输出（output）为：kafka的同一个topic中（三台安装了kafka的机器都需要配置好各自的ip地址）。
output只支持一个输出。
nginx 的web中的日志只需要吐一次，将数据吐到kafka中，不需要去filebeat里面，如果需要，直接去kafka里面去拿，而且filebeat中只支持一个输出（output）。如果在/etc/filebeat下文件写入多个output会导致服务重启不了。

日志采集器有很多，比如Logstash，虽然Logstash的功能强大，但是它依赖java并且在数据量大的时候进程会消耗过多的系统资源，会严重影响业务系统的性能。

而filebeat就是一个完美的替代者，它基于Go语言没有任何依赖，配置文件简单。同时，filebeat比logstash更加轻量级，所以占用系统资源极少，非常适合安装在生产机器上。

Filebeat可以直接将数据发送到Elasticsearch、Kafka或者Redis，在那里可以进一步处理和增强数据，然后在Kibana中将其可视化，目前来说Filebeat是 ELK 日志系统在Agent上的第一选择。

Kafka是是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做MQ系统），常见可以用于web/nginx日志、访问日志，消息服务等等。
kafka是一个消息中间件。
kafka通常用在：日志收集、业务解耦、流量削峰。

kafka的主要应用场景是：日志收集系统和消息系统。

面向消息的系统（消息中间件）是在分布式系统中完成消息的发送和接收的基础软件。消息中间件也可以称消息队列，是指用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息队列模型，可以在分布式环境下扩展进程的通信。

消息中间件：是两种通信模式：点对点、发布订阅（kafka）。
kafka是消息中间件中的一种。kafka支持的通信方式：发布—订阅。

1、故障发生时方便定位问题
2、日志集中管理，后续需要日志的程序需要直接懂kafka获取日志即可。
3、尽可能的减少日志处理对nginx的影响。

kafka的日志在/opt/kafka_2.12-2.8.1/logs下，主要看server.log

从上图中，producer就是生产者，是数据的入口。图中的producer在写入数据的时候永远在找leader，不会直接将数据写入follower。
消息写入leader之后，follower是主动的去leader进行同步的！producer采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的！

Producer：生产者，消息的产生者，是消息的入口。
Broker:broker是kafka实例，每个服务器上有一个或者多个kafka的实例，我们暂时认为每个broker对应一台服务器，每个kafka集群内的broker都有一个不重复的编号，如：broker1、broker2、broker3。
Topic：消息的主题，可以理解为消息的分类，kafka的数据保存在topic。在每个broker上都可以创建多个topic。
Partition：topic的分区，每个topic可以有一个或者多个分区（partition），partition是比较具体的东西，partition在服务器上的表现形式就是一个一个的文件夹，每个partition的文件夹下面会有多组segment文件，每组segment文件又包含.index文件、log文件、.timeindex文件三个文件，log文件就是实际是存储message的地方，而index和timeindex文件为索引文件，用于检索消息。分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同分区的数据是不重复的，partition的表现形式就是一个一个的文件夹。

图中，这个的partition有三组segment文件，每个log文件的大小是一样的，但是存储的message数量是不一定相等的，（每条的message大小不一致)。文件的命名是以该segment最小offset来命令的，如000.index存储offset为0~368795的消息,kafka就是利用分段+索引的方式来解决查找效率的问题。

Segment：partition物理上由多个segment组成，Partition的结构：Partition在服务器上的表现形式就是一个一个的文件夹，每个Partition的文件夹下面会有多组segment文件，每组segment文件又包含.index文件、.log文件、.timeindex文件这三个文件，log文件就实际是存储message的地方，而index和timeindex文件为索引文件，用于检索消息。

Replication：每个分区都有多个副本，副本的作用就是做备胎。当主分区（leader）故障的时候会选择一个备胎（follower)上位,成为leader。在kafka中默认副本的最大数量是10个，并且副本的数量不能大于一个副本（包括自己）。

kafka的副本机制每个主题上会划分成若干个分区。副本就是在分区层级下定义的，每个分区都会配置若干个副本。在同一个分区下的所有副本保存有相同的消息序列，这些副本保存在不同的broker上，能够对抗部分broker宕机带来的数据不可用。

Message:每一条发送的消息主体。
Consumer：消费者，即消息的消费方，是消息的出口。
Consumer Group:我们可以将多个消费组组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量。
Zookeeper：kafka集群依赖zookeeper来保存集群的元信息，来保证系统的可用性。

从上图中，producer就是生产者，是数据的入口。图中的producer在写入数据的时候永远在找leader，不会直接将数据写入follower。
消息写入leader之后，follower是主动的去leader进行同步的！producer采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的。

数据会写入到不同的分区，那kafka为什么要做分区：
分区的主要目的是：
① 方便扩展。因为一个topic可以有多个partition，所以我们可以通过扩展机器去轻松应对日益增长的数据量。

② 提高并发。以partition为读写单位，可以多个消费者同时消费数据，提高了消息的处理效率。

producer将数据写入kafka后，集群就需要对数据进行保存了！kafka将数据保存在磁盘，kafka初始会单独开辟一块磁盘空间，顺序写入数据（效率比随机写入高）。

多个broker +多个partition +多个 replica

如何保证消费者消费数据一致性：消费者可以连接到任何一台机器，机器会告诉它leader在哪里，如何通过leader进行外界交互。
同一个消费组，里面的消费者同一时刻只能消费一个partition的数量。
一个消费组里面的每一个消费者可以同时去消费一个topic里面的不同分区里的数据。

针对topic来说。只要能连接上这个kafka，消费者都可以进行消费，只要有权限获取到topic，消费者之间的消费互不影响。

1、producer可以通过request.required.acks设置。
① 生产者一致性：
保证消息不丢失是一个消息队列中间件的基本保证，那producer在向kafka写入消息的时候，为了保证消息不丢失，就可以通过ACK回答机制。在生产者向队列写入数据的时候可以设置参数来确定是否确认kafka接收到数据，ack参数可以设置为1、1、-1
acks =0 ,生产者不会等待任何来自服务器的响应。生产者不需要接收响应，发完就发下一条，不确保消息是否发送成功。安全性最低但是效率最高。
ack1=1 （默认值）:leader收到就会发送响应给生产者，只需要确保leader发送成功，再发送下一条。
acks=-1 (ISR)，等待ISR列表中的每一个副本都接收到，才给生产者响应。生产者才会接收到来自服务器的成功响应，这个模式是最安全的，但是效率很低。
acks=2，除了一个leader收到，还需要一个flower收到。
ack= n，除了一个leader收到，还需要等到n-1个follower收到。
ack= all，代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下一条，确保leader发送成功和所有的副本都完成备份。安全性最高，但是效率最低。

2、消费者消费数据时，引入了High Water Mark机制、木桶效应，只能消费ISR列表里偏移量最小的副本的消息数量。

1、使用offset偏移量：记录消费者消费偏移量。
本来offset保存在本地，但是现在保存在kafka的__consumer_offsets中主题中，每次消费每隔几秒就会写入，需要的时候，就会自动去kafka的主题里面拿。
也阔以自己写一个文档，里面的写入你上次消费的记录。
消费者消费的时候，会记录自己的消费偏移量，消费偏移量可以自己保存在本地，也可以提交到kafka的_consumer_offsets主题里的保存。

bin/kafka-console-consumer.sh --bootstrap-server 192.168.2.152:9092 --topic nginxlog --from-beginning
执行kafka-console-consumer.sh的时候，就会自动保存到kafka主题中。

[root@nginx-kafka01 opt]# cd /data
[root@nginx-kafka01 data]# ls
cleaner-offset-checkpoint  __consumer_offsets-30  log-start-offset-checkpoint
__consumer_offsets-0       __consumer_offsets-33  meta.properties
__consumer_offsets-12      __consumer_offsets-36  nginxlog-0
__consumer_offsets-15      __consumer_offsets-39  recovery-point-offset-checkpoint
__consumer_offsets-18      __consumer_offsets-42  replication-offset-checkpoint
__consumer_offsets-21      __consumer_offsets-45  replication-offset-checkpoint.tmp
__consumer_offsets-24      __consumer_offsets-48  sc-0
__consumer_offsets-27      __consumer_offsets-6   sc-1
__consumer_offsets-3       __consumer_offsets-9   sc-2
[root@nginx-kafka01 data]# cd nginxlog-0/
[root@nginx-kafka01 nginxlog-0]# ls
00000000000000000000.index  00000000000000000000.timeindex
00000000000000000000.log    leader-epoch-checkpoint
index记录每个offset所对应的的每个位置，是一个二进制文件，真正的是存放在多个log文件中。
发送过来的数据都存放在log文件中。
/data里面存放kafka数据的位置：

每一个partition的数据都是由很多个segment存储，每一个segment（段）由一个index（索引）和log文件组成。分出多个segment：段便于做数据清理。

存放在多个log文件里面的原因：
方便kahka清除数据，按时间或者大小存储，不会永久保存，定期保存，默认保存七天，可以进行修改。修改配置文件为：/opt/kafka_2.12-2.8.1/config

kafka可以按照两个维度清理数据：
1、按照大小
2、按时间
任意一个条件满足，都可以触发日志清理。

zookeeper是分布式应用协调管理服务、配置管理、域名管理、分布式数据管理、集群管理

zookeeper架构图：

zk集群中，节点存活数必须过半，集群才能正常使用。一般集群的节点数都设置为奇数，方便选举，而且容错率一样。
如果不过半容易脑裂。就比如一个大集群，里面有5个节点，zk1-5，里面存活的数必须过半，进行选举，所以节点里面一般以奇数。如果zk4和zk5坏掉了，至少也得有前三台都存活。

写的话是通过leader。读的话leader和follower都可以。
可以直接连接follower进行查询操作。
follower：查询、选举。选举是少数服从多数。

zookeeper是用来管理kafka的，用来保存kafka的元信息、topic、partition、副本信息，通过抢占的方式选举出kafka的controller，这个controller用来管理kakfa副本partition中的leader和follower的选举。

而zookeeper本身也是有leader和follower的，它的选举方式是采用一致性算法（zab），根据少数服从多数，票数过半的原则来选举leader。

所以在zk集群中，机器存活数必须过半，集群才能正常使用，所以我们通常也会将zk集群的节点数设为奇数个，这是为了方便选举。

跟filebeat给Kafka传递数据一样，Kafka连接任意一台zookeeper都可以操作，但是数据新增修改等事务操作必须在leader上运行。

客户端连接任意一台zk都可以操作：但是数据新增修改等事物操作必须在leader上运行。客户端如果连接到follower上进行事务操作，follower会返回给leader的IP，最终客户端还是在leader上操作，但是可以直接连接follower进行查询操作。

follower：查询、选举。选举是少数服从多数。
zookeeper中数据的同步，只要过半节点同步完成，就表示数据已经提交。（commit）
zookeeper不是强一致性的。它属于最终一致性。
zookeeper里面存放的数据很少，
每个节点存储的数据量默认不会超过1M。存储速度快。
zookeeper相当于一个文件树。

编写的python程序代码存放在gitee中：

python_consumer.py

数据存入mysql中的，需要先创建好mysql数据库：

1、需求分析
需要nginx日志的ip，时间，带宽字段
将ip字段解析成相应的省份、运营商
存入数据库的字段： id， 时间， 省份， 运营商， 带宽
带宽：通过nginx.log去获取。
淘宝接口：
使用pykafka模块。

#步骤
1、创建数据表
2、编写python脚本， 从kafka获取nginx日志
3、获取好的nginx日志，提取出ip，时间，带宽字段
4、提取出的ip字段通过淘宝的一个接口解析出省份和运营商
url = “https://ip.taobao.com/outGetIpInfo?accessKey=alibaba-inc&ip=114.114.114.114”
5、格式化时间字段 “2021-10-12 12:00:00”
6、存入数据库

#创建表
create table nginxlog (
id  int primary key auto_increment,
dt  datetime not null,
prov varchar(256) ,
isp  varchar(256),
bd  float
) CHARSET=utf8;

create table prov_index(
id  int primary key auto_increment,
prov_name  varchar(256)
) charset=utf8;

create table isp_index(
id int primary key auto_increment,
isp_name varchar(256)
) charset=utf8;

到这里项目的详解就结束了，如有理解错误的地方，请大家进行批评指正。