搭建kafka

一. 概述

1. 介绍

Kafka 是个分布式的、持分区的（partition）、多副本的 （replica），基于 zookeeper 协调的分布式消息系统，它最大的特性就是可以实时处理大量数据以满足各类需求场景：

• 日志收集：使用 Kafka 收集各种服务的日志，并通过 kafka 以统一接口服务的方式开放给各种 consumer，例如 hadoop、Hbase、Solr 等

• 消息系统：解耦和生产者和消费者、缓存消息等

• 用户活动跟踪：Kafka 经常被用来记录 web 用户或者 app 用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到 kafka 的 topic 中，然后订阅者通过订阅这些 topic 来做实时的监控分析，或者装载到 hadoop、数据仓库中做离线分析和挖掘

• 运营指标：Kafka 也经常用来记录运营监控数据，包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告

   

2. 概念

1. kafka作为一个集群运行在一个或多个服务器上。

2. kafka集群存储的消息是以topic为类别记录的。

3. 每个消息（也叫记录record，我习惯叫消息）是由一个key，一个value和时间戳构成。

3. 四个核心API

应用程序使用 Producer API 发布消息到1个或多个topic（主题）中。

应用程序使用 Consumer API 来订阅一个或多个topic，并处理产生的消息。

应用程序使用 Streams API 充当一个流处理器，从1个或多个topic消费输入流，并生产一个输出流到1个或多个输出topic，有效地将输入流转换到输出流。

Connector API 可构建或运行可重用的生产者或消费者，将topic连接到现有的应用程序或数据系统。例如，连接到关系数据库的连接器可以捕获表的每个变更

4. 消息系统介绍

一个消息系统负责将数据从一个应用传递到另外一个应用，应用只需关注于数据，无需关注数据在两个或多个应用间是如何传递的。分布式消息传递基于可靠的消息队列，在客户端应用和消息系统之间异步传递消息。有两种主要的消息传递模式：点对点传递模式、发布-订阅模式。大部分的消息系统选用发布-订阅模式。Kafka就是一种发布-订阅模式。

点对点消息传递模式

在点对点消息系统中，消息持久化到一个队列中。此时，将有一个或多个消费者消费队列中的数据。但是一条消息只能被消费一次。当一个消费者消费了队列中的某条数据之后，该条数据则从消息队列中删除。该模式即使有多个消费者同时消费数据，也能保证数据处理的顺序。 生产者发送一条消息到queue，只有一个消费者能收到。

发布-订阅消息传递模式

在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者。 发布者发送到topic的消息，只有订阅了topic的订阅者才会收到消息。

参考来源：https://www.orchome.com/5

https://www.cnblogs.com/Yee-Q/p/15866729.html

https://www.cnblogs.com/cxy2020/p/14999545.html

二. 配置

1. 下载

选择适合的版本进行下载

下载地址：https://kafka.apache.org/downloads

2. 配置文件

• server.profperties

   vim server.profperties

   #全局唯一编号，不能重复
   broker.id=0 （在hadoop01,02,03上，分别改为0,1,2）
   #kafa运行日志存放路径(vi 中，在非insert模式 按/ 后面加上要搜索的内容)
   log.dirs=/opt/export/data/kafka/logs/
   ​
   #制定zookeeper的位置，主要用来保存meta数据
   zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181
   ​
   #允许删除主题 （此属性没有，要自己添加）
   delete.topic.enable=true
   ​
   #设置本机IP  每台机器都要改成自己的IP（注意）
   host.name=hadoop01

• 日志

   # Kafka设置系统运行日志输出路径
   vim /opt/export/servers/kafka_2.12-2.4.1/bin/kafka-run-class.sh
   LOG_DIR=/opt/export/data/kafka/logs

   

3. 环境变量

 sudo vim /etc/profile
 export KAFKA_HOME=/opt/export/servers/kafka_2.12-2.4.1
 export PATH=$PATH:$KAFKA_HOME/bin
 ​
 source /etc/profile

4. 分发文件

 scp -r kafka_2.12-2.4.1/ hadoop02:/opt/export/servers/
 scp -r kafka_2.12-2.4.1/ hadoop03:/opt/export/servers/
 ​
 scp /etc/profile hadoop02:/etc/profile
 scp /etc/profile hadoop03:/etc/profile
 # 在hadoop02、hadoop03 刷新环境配置
 source /etc/profile

 

## 4. 启动

需要分别在hadoop01,hadoop02,hadoop03启动kafka

 # 后台启动  nohup &
 # >/opt/export/data/kafka/logs/start-kafka.log 2>&1  为启动日志存放路径
 nohup kafka-server-start.sh /opt/export/servers/kafka_2.12-2.4.1/config/server.properties >/opt/export/data/kafka/logs/start-kafka.log 2>&1 &

三、测试使用

 #查询当前kafka的所有主题
 kafka-topics.sh --list --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181
 ​
 #创建主题(名字叫test)
 kafka-topics.sh --create --topic test --partitions 3 --replication-factor 3 --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181
 ​
 #删除主题() 正在使用的主题无法删除
 kafka-topics.sh --delete --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --topic test
 ​
 # 创建一个生产者，向某个(test)主题中写入数据,连接上
 kafka-console-producer.sh --broker-list hadoop01:9092,hadoop02:9092,hadoop03:9092 --topic test
 ​
 [hadoop@hadoop01 ~]$ kafka-console-producer.sh --broker-list hadoop01:9092,hadoop02:9092,hadoop03:9092 --topic test
 >hello world
 >
 ​
 # 创建一个消费者，订阅某个主题(test)
 kafka-console-consumer.sh --from-beginning --topic test --bootstrap-server hadoop01:9092,hadoop02:9092,hadoop03:9092
 ​
 [hadoop@hadoop02 ~]$ kafka-console-consumer.sh --from-beginning --topic test --bootstrap-server hadoop01:9092,hadoop02:9092,hadoop03:9092
 hello world
 ​
 # ctrl + c 退出
 # 关闭kafka
 kafka-server-stop.sh

 

标签：12,--,hadoop02,hadoop01,9092,kafka,topic,2.4
来源： https://www.cnblogs.com/sirlateautumns/p/16201711.html