Kafka的安装与配置学生操作手册

Kafka的安装与配置学生操作手册什么是Kafka，Kafka的作用ApacheKafka是一个开源的流处理平台，最初由LinkedIn公司开发，并于2011年成为Apache软件基金会的一个顶级项目。它被设计用于处理大规模的实时数据流，支持高吞吐量、持久性存储和水平扩展性。Kafka的核心概念包括：消息传递系统：Kafka是一个分布式消息传递系统，用于发布和订阅消息。它允许应用程序之间以异步方式发送和接收数据。发布-订阅模型：生产者（Producer）负责将消息发布到一个或多个主题（Topics），而消费者（Consumer）则订阅一个或多个主题以接收消息。分布式：Kafka是一个分布式系统，可以水平扩展，允许将数据分布在多个节点上，以实现高吞吐量和可靠性。持久性：Kafka通过将消息存储在磁盘上，实现了持久性存储。这意味着即使消费者断开连接，消息仍然可以被保留，并在稍后重新处理。日志：Kafka的存储机制被称为“日志（log）”，它将消息追加到可追加的文件中，从而实现了高效的写入和顺序读取。流处理：Kafka还支持流处理，使得可以在流数据上执行实时处理操作。Kafka的使用场景包括日志收集、事件处理、实时数据分析等。许多大型企业和组织都在其架构中使用Kafka来处理和传递实时数据。实时数据传输：Kafka作为消息传递系统，可以用于实时数据传输。生产者可以将实时生成的数据发布到Kafka的主题中，而消费者可以订阅这些主题以接收实时数据。日志集中存储：Kafka的日志存储机制使其成为一个优秀的日志集中存储系统。它能够持久性地存储大量的日志数据，并支持高效的写入和顺序读取。解耦数据系统：Kafka解耦了数据的生产和消费过程。生产者和消费者之间通过主题进行通信，这使得系统组件能够独立地演化和扩展，而不会对彼此产生直接的影响。消息队列：Kafka可以用作高可靠的消息队列系统。生产者将消息发送到Kafka，而消费者从Kafka中拉取消息。这种消息队列的模型对于处理异步任务、削峰填谷以及确保消息传递的可靠性都非常有用。流数据处理：KafkaStreamsAPI使得在Kafka上进行实时流数据处理变得更加容易。通过流处理，可以对数据进行实时分析、转换和聚合。事件溯源：Kafka可以用于实现事件溯源，即将所有关键的业务事件以事件流的形式存储和传递。这对于构建可追溯性、可回溯性的系统非常有帮助。日志和监控：通过存储事件日志，Kafka能够成为系统监控的有力工具，帮助了解系统的行为和性能。日志中的信息可以用于故障排查、性能分析等。总的来说，Kafka的作用在于提供一个高效、可扩展、持久性的平台，用于处理实时数据流，并在各种应用场景中实现可靠的数据传输和处理。Kafka的安装与配置1解压软件上传kafka_2.12-2.4.1.tgz文件包至/export/software,输入如下命令cd/export/softwaretar-xvfkafka_2.12-2.4.1.tgz-C/export/servers/将解压后的文件复制到slave1和salve2中，命令如下：scp-r/expot/servers/kafka_2.12-2.4.1slave1:/expot/servers/scp-r/expot/servers/kafka_2.12-2.4.1slave2:/expot/servers/2配置perties文件输入如下命令：cd/export/servers/kafka/configviperties在配置文件中找到对应的行并修改相关的值broker.id=0zookeeper.connect=master:2181,slave1:2181,slave2:2181Kafka负责存储基本单位是broker。Kafka需要传输的消息量较大时，这些消息数据被分为一个或多个分区，每个分区存储在一个broker上，在集群环境下，分区数据会有多个备份。这确保了消息的持久性和高可用性，即使某个broker发生故障，数据仍然可用。broker.id是用于标识Kafka集群中每个broker的唯一标识符。每个Kafkabroker都必须有一个唯一的broker.id。这个标识符对于Kafka集群的正常运行非常重要，因为它帮助Kafka集群中的各个节点（brokers）相互识别和通信。在这里要注意的是，master1中的broker.id设置为0，在slave1中的broker.id设置为1，slave2中的broker.id设置为2。3、启动Kafka服务输入如下命令：cd/export/servers/kafka/bin/kafka-server-start.shconfig/perties这里要注意的是，启动服务是需要perties支持的，当然用户也可以根据需求选择不同的perties作为启动配置项，设置不同的路径选择不同的文件即可。4.关闭Kafka服务相关的配置Kafka关闭服务中本身具有Bug，输入如下指令：vi/export/servers/kafka/bin/kafka-server-stop.sh找到这样一条语句：PIDS=$(psax|grep-i'kafka\.Kafka'|grepjava|grep-vgrep|awk'{print$1}')这条语句的含义就是得到所有包含kafka.Kafka的进程ID，但是由于kafka启动依赖比较多的jar，导致kafka进程的ps结果输出内容比较长，而ps输出结果受到PAGE_SIZE（其值通过命令getconfPAGE_SIZE可以得到）的限制，从而导致ps结果中看不到kafka.Kafka，所以不能kill掉kafkaserver。这样就导致了当使用bin/kafka-server-stop.sh脚本关闭服务时，可能得到的执行结果为：Nokafkaservertostop，提示找不到kafka服务。建议将这条语句删除掉或注释掉，添加如下脚本：PIDS=$(jps-lm|grep-i'kafka.Kafka'|awk'{print$1}')if[-z"$PIDS"];thenecho"Nokafkaservertostop"exit1elsekill-s$SIGNAL$PIDSfi找出所有与Kafka相关的进程全部关闭，这样就可以关闭所有进程kafka集群启动脚本的编写进入/bin目录，cd/bin新建脚本vika.sh添加shell脚本如下：#!/bin/bashcase$1in"start"){foriinhadoop01hadoop02hadoop03doecho"启动$iKafka"ssh$i"source/etc/profile;/export/servers/kafka/bin/kafka-server-start.sh-daemon/export/servers/kafka/config/perties"done};;"stop"){foriinhadoop01hadoop02hadoop03doecho"停止$iKafka"ssh$i"source/etc/profile;/export/servers/kafka/bin/kafka-server-stop.sh"done};;Esac添加脚本的执行权限：chomd777ka.sh执行脚本：ka.shstart启动kafka集群，一次开启三台服务器的zookeeper服务ka.shstop关闭kafka集群，一次关闭三台服务器的zookeeper服务5、创建Topic为order的主题bin/kafka-topics.sh--create--topicorder--partitions3--replication-factor3--zookeeperhadoop01:2181,hadoop02:2181,hadoop03:2181其中topic:要创建的主题的名称。partitions:主题的分区数，用于并行处理和提高可伸缩性。replication-factor:主题的副本数，用于提供数据冗余和容错性。zookeeper:用于指定zookeeper的地址。验证主题创建：使用kafka-topics.sh或其他Kafka管理工具来验证主题是否成功创建，以及查看主题的配置信息。这里介绍使用kafka-topics.sh验证的方法，具体命令主下kafka-topics.sh--describe--bootstrap-servermaster:9092--topicmy_topic这个命令将显示有关主题的详细信息，包括分区、副本、分配的broker等。6、生产者生产数据利用shell脚本生产数据一般应用在测试环境下，真实生产环境，一般是通过Flume进行采集数据再传递给KaFka。shell脚本生产数据的命令如下bin/kafka-console-producer.sh--broker-listmaster:9092--topicorderbroker-list:Kafka集群的引导服务器地址，用于指定连接到集群的入口点。topic:要发布消息的主题名称。命令执行时需要，在下方输入要传输的数据，按下回车键后会在消费者显示相同的内容7、消费者