526互联

Kafka基础阶段与集群搭建详细教程

发布时间 2023-05-05 20:02:09作者: LycCj

Kafka第一天课堂笔记

一.Kafka简介

1.1消息队列

  • 消息队列——用于存放消息的组件
  • 程序员可以将消息放入到队列中，也可以从消息队列中获取消息
  • 很多时候消息队列不是一个永久性的存储，是作为临时存储存在的（设定一个期限：设置消息在MQ中保存10天）
  • 消息队列中间件：消息队列的组件，例如：Kafka、ActiveMQ、RabbitMQ、RocketMQ、ZeroMQ

注意：
说白了消息中间件就是用来存储消息的软件，举一个例子来说，为了分析网站的用户行为，我们需要记录用户的访问日志等一系列的信息，我们可以将他们保存在消息队列当中。等将来有一些应用程序需要处理这些日志的手，就可以随时将这些消息取出来进行处理。

1.2Kafka的应用场景

  • 异步处理

    • 可以将一些比较耗时的操作放在其他系统中，通过消息队列将需要进行处理的消息进行存储，其他系统可以消费消息队列中的数据
    • 比较常见的：发送短信验证码、发送邮件

    就是在电网网站当中，当新的用户注册的时候，需要将用户的信息保存到数据库中，同时还需要额外的发送注册的邮件通知信息，以及短信注册码给用户。但因为发送邮件，发送注册短信需要连接外部的服务器，需要额外的等待一段时间。因此，就可以使用消息队列来进行异步处理，从而实现快速的响应。

image-20230503193458049

  • 系统解耦

    • 原先一个微服务是通过接口（HTTP）调用另一个微服务，这时候耦合很严重，只要接口发生变化就会导致系统不可用
    • 使用消息队列可以将系统进行解耦合，现在第一个微服务可以将消息放入到消息队列中，另一个微服务可以从消息队列中把消息取出来进行处理。进行系统解耦

    例如：在SpringCloudAlibaba开发的微服务的项目当中，我们不同模块的微服务可能会将程序放在不同的server服务器当中,例如下面的订单系统可能会存放在北京的服务上，而库存系统可能会存在上海的服务当中，我们需要采取远程调用的方式去调取服务。但这个过程很受性能的影响。而且一个微服务的不可用可能会导致整个微服务的崩溃。这其中每一个微服务与微服务之间的耦合程度较高。因此我们可以选用消息队列来解耦。

image-20230503193510279

  • 流量削峰

    • 因为消息队列是低延迟、高可靠、高吞吐的，可以应对大量并发

    因此，我们可以给我们的消息队列做一个集群，一个集群式的消息队列来配合多台web服务器以及多个Mysql数据库来进行使用，可以一定程度上解决消息的大规模的访问数据库的问题。

    image-20230503193517953

  • 日志处理

    • 可以使用消息队列作为临时存储，或者一种通信管道

    大型电商网站（淘宝，京东，国美，苏宁），APP（抖音，美团，滴滴等等）需要分析用户的行为。需要根据用户的访问行为来发现用户的喜好以及活跃的情况，需要在页面上收集大量的用户的访问信息，这些信息大都可以使用日志来处理，例如：我们在抖音上频繁刷旅游的视频，大数据会根据视频标签将访问的请求归结到一条条的接口访问的日志，我们将这些日志可以存放到消息队列当中，然后，利用一些大数据的手段，来推荐给用户经常看的一些标签视频。例如大数据中的flume是高可用的，高可靠的分布式的海量的日志采集，聚合和传输的系统。另外我们还有可以在服务器Nginx中采用lua脚本利用OpenResty来进行编写（这一块在多级缓存的中使用的较多，一条多级缓存主线主要涉及Nginx,redis还有后台的Mysql数据库等）。

image-20230503193531738

1.3消息队列的两种模型

  • 生产者、消费者模型

    • 生产者负责将消息生产到MQ中
    • 消费者负责从MQ中获取消息
    • 生产者和消费者是解耦的，可能是生产者一个程序、消费者是另外一个程序

  • 消息队列的模式

    • 点对点：一个消费者消费一个消息

    点对点模式的特点：

    (1)每一个消息只有一个接收者（Consumer）(即一旦被消费，消息就不在消息队列当中了)
(2)发送者和接收者之间没有依赖性，发送者发送消息之后，不管有没有接收者在运行，都不会影响到发送者下次发送消息
(3)接收者在成功接收消息之后需要向队列应答成功，以便消息队列删除当前接收的消息。避免重复消费
    • 发布订阅：多个消费者可以消费一个消息：

    发布订阅模式的特点：

    (1)每一个消息可以有多个订阅者
(2)发布者和订阅者之间有时间上的依赖性。针对某一个主题（Topic）的订阅者，它必须创建一个订阅者模式后才能发布消息
(3)为了消费消息，订阅者需要提前订阅该角色主题，并保持在线运行

1.4Kafka简介：

image-20230505132203123

Kafka是由Apache软件基金会开发的一个开源流平台，由Scala和Java编写。Kafka的Apache官网是这样介绍Kafka的：

1.Apache kafka是一个分布式流平台。一个分布式的流平台应该包含3点关键的能力
（1）发布和订阅流数据，类似于消息队列或者是企业级消息传递系统
（2）以容错的持久化方式存储数据流
（3）处理数据流

我们关注的重点的三个关键部分：

(1)Publish and subscribe : 发布与订阅
(2)Store : 存储
(3)Process ：处理

二.Kafka集群搭建

  • Kafka集群是必须要有ZooKeeper的

注意：

  • 每一个Kafka的节点都需要修改broker.id（每个节点的标识，不能重复）
  • log.dir数据存储目录需要配置

详细的搭建流程：
1.首先将Kafka的安装包上传到虚拟机，并进行解压：准备虚拟的教程可以看：阿里云服务购买 - LycCj - 博客园 (cnblogs.com)（使用Xshell连接上远程的阿里云服务器）

（1）连接上虚拟机：

image-20230505134752770

（2）上传Kafka的文到/export/software文件目录下：查看Kafka官网就可以看见众多版本的Kafka，选取与视频版本一致的Kafka下载包,并执行以下命令：（export文件，server文件，software文件需要自己手动创建）

（1）cd /export/software/
（2）tar -xvzf kafka_2.12-2.4.1.tgz -C ../server/
（3）cd /export/server/kafka_2.12-2.4.1/

image-20230505140638453

2.修改配置文件：
（1）需要修改broker的id,因为将来为Kafka做集群的时候我们需要为不同的kafka配置不同的id

（2）指定Kafka的数据存放位置

配置如下：

(1)cd /export/server/kafka_2.12-2.4.1/config
(2)vim server.properties
# 指定broker的id
(3)broker.id=0
# 指定Kafka数据的位置
(4)log.dirs=/export/server/kafka_2.12-2.4.1/data

3.配置zookeeper集群：
注意在实际企业开发的过程当中，我们的Kafka通常会配置成为集群类型，而不是单一的Kafka节点，是为了保证高可用，而kafka是强依赖于Zookeeper的，因此我们需要首先搭建zookeeper的集群：
注意：我们这里可以使用一台服务器完成集群的搭建，如果资金充裕可以完全按照企业级的开发场景，其实本质是差不多的，例如：在一台机器上模拟一个zookeeper集群就是利用同一个IP开放不同的端口模拟不同的zookeeper,而如果是真实的企业级开发的要求就是，不同IP对应相同的端口，本质上配置并没有什么差别。

（1）首先安装zookeeper:image-20230505150435120

(2)对于zookeeper进行配置：

  • 进入到conf文件中将zoo_sample.cfg文件重新命名为zoo.cfg文件：

image-20230505151112840

  • 对于zoo.cfg文件进行配置：

1）配置数据的目录：

dataDir=/export/server/zookeeper-3.4.14/data

2）注意zookeeper对外开放的端口是2181，因为如果使用的是阿里云，需要开放阿里云的安全组的规则：
image-20230505151713348

3）启动，查看zookeeper是否安装成功：显示有进程信息则zookeeper安装成功
image-20230505152122034

（3）搭建zookeeper集群：

为每一台zookeeper配置对应的数据与端口号：

阿里云1服务器：

image-20230505182918077

阿里云2服务器：

image-20230505183023460

阿里云3服务器：

image-20230505183023460

(4)在每一个zookeeper的data目录下创建一个myid文件，内容分别是0，1，2.这个文件就是记录每一个服务器的(注意此时的data文件只有启动过一次zookeeper才会出现)

image-20230505154106967

（5）我们需要配置每一个zookeeper的zoo.cfg配置集群服务器的IP列表：

解释：server.服务ID = 服务器IP地址 ：服务器之间的通信端口：服务器之间投票选举的端口

server.0=47.93.97.93:2888:3888
server.1=123.57.85.120:2888:3888
server.2=60.205.169.227:2888:3888

（6）因为zookeeper是通过Java开发的需要Java的环境，因此配置Java的环境变量：

解压缩到export/server目录下：
image-20230505172412078

配置环境变量：
在/etc/profile文件中配置如下变量：

export JAVA_HOME=自己解压的jdk目录
export CLASSPATH=$JAVA_HOME/lib
export PATH=$PATH:$JAVA_HOME/bin

刷新一下文件：

source /etc/profile

查看自己的jdk是否安装好：

java -version

（7）对于云ECS服务器需要添加以下的配置：

quorumListenOnAllIPs=true

（8）测试自己的zookeeper集群是否搭建完成：

image-20230505184446837

可以看出两主一从的集群已经搭建完毕，并且自动的选择一个主节点

4.利用已经搭建好的zookeeper集群来搭建kafka集群：

更改kafka文件下conf中的server.properties中的zookeeper.connect：

zookeeper.connect=47.93.97.93:2181,123.57.85.120:2181,60.205.169.227:2181

另外集群还需更改broker.id与log.dir还有zookeeper.connect.

5.配置Kafka的HOME环境变量：

(1)vim /etc/profile
(2)KAfka的环境变量：
export KAFKA_HOME=/export/server/kafka_2.12-2.4.1
export PATH=:$PATH:${KAFKA_HOME}
(3)source /etc/profile

6.启动Kafka:

linux命令如下：

image-20230505191904350

nohup bin/kafka-server-start.sh config/server.properties &

7.修改server.properties中的：

#advertised.listeners=PLAINTEXT://your.host.name:9092 
advertised.listeners=PLAINTEXT://服务器IP:9092

8.判断Kafka集群搭建是否成功：

注意第7步是为第8步做铺垫，不然很有可能报如下错误：

服务名称不对错误：

2023-05-05 19:37:53,101] WARN [AdminClient clientId=adminclient-1] Error connecting to node iZ2ze2cznowrv7erlrvph2Z:9092 (id: 2 rack: null) (s.NetworkClient)
java.net.UnknownHostException: iZ2ze2cznowrv7erlrvph2Z: Name or service not known
	at java.base/java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method)
	at java.base/java.net.InetAddress$PlatformNameService.lookupAllHostAddr(InetAddress.java:929)
	at java.base/java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1538)
	at java.base/java.net.InetAddress$NameServiceAddresses.get(InetAddress.java:848)
	at java.base/java.net.InetAddress.getAllByName0(InetAddress.java:1527)
	at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1380)
	at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1302)
	at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:104)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:403)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:363)
	at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:151)
	at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:949)
	at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:291)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.sendEligibleCalls(KafkaAdminClient.java:969)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1184)
	at java.base/java.lang.Thread.run(Thread.java:834)


bin/kafka-topics.sh --bootstrap-server IP名:9092 --list

如果什么错都没报就说明Kafka集群搭建成功。