1. kafka 简介

kafka 是一个分布式的消息队列系统；官网：https://kafka.apache.org/

kafka 集群有多个broker 服务器组成，每个类型的消息被定义为topic;(主题)

（消息队列 里面 一般分为消费者 和生产者 如其他的消息队列，activeMQ..）

消息生产者producer 和消费者 consumer 可以在多个broker 上生产/消费 topic

概念理解：

producer :生产者，生产topic 消息

consumer: 消费者；消费对应的topic 消息

topic: 主题； （相当于map中的key生产者put一个key-value , 消费者拿个key 获取value）

partition ：分区；每个topic 包含多个partition,partition数量可以在创建topic 的时候指定

broker: kafka集群中的一台服务器

kafka 只保证每个分区中的消息有序，不能保证同一个主题中的不同分区之间的消息有序，如果你想要保证所有的消息都绝对有序可以只为一个主题分配一个分区。

分区会给每个消息记录分配一个顺序ID号（偏移量）， 能够唯一地标识该分区中的每个记录。Kafka集群保留所有发布的记录，不管这个记录有没有被消费过，Kafka提供相应策略通过配置从而对旧数据处理。

每个消费者可以根据偏移量，读取任何位置的消息；

2. kafka 集群部署

1. 准备zookeeper ,安装zooker 省略 。。。 太多了

我的zookeeper 的配置信息 3台，

2. 安装kafka下载压缩包（官网地址：http://kafka.apache.org/downloads.html

上传解压：

tar zxvf kafka_2.10-0.9.0.1.tgz

修改配置文件：进入到kafka 的配置文件目录conf 中

修改连接地址

将kafka 及配置信息 分发到其他设备上

修改其他设备的配置文件中的 broker.id 使用sed 命令 也可以直接vi那个文件

sed -i -e 's/broker.id=.*/broker.id=1/' /var/mytools/kafka_2.10-0.9.0.1/config/server.properties

启动kafka 集群：

1 分别启动zookeeper集群

分别在三台服务器上执行以下命令启动：启动kafka 集群

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

启动成功：是一个阻塞状态

3. 测试

（kafka-topics.sh --help查看帮助手册）

1. 创建topic

bin/kafka-topics.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --create --replication-factor 2 --partitions 3 --topic test

（参数说明：

--replication-factor：指定每个分区的复制因子个数，默认1个

--partitions：指定当前创建的kafka分区数量，默认为1个

--topic：指定新建topic的名称）

查看topic 列表

bin/kafka-topics.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --list

查看topic--test 的描述

bin/kafka-topics.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --describe --topic test

创建生产者：

bin/kafka-console-producer.sh --broker-list hadoop01:9092,hadoop02:9092,hadoop03:9092 --topic test

你可以输入你的消息

生产消息：123456

创建消费者：

bin/kafka-console-consumer.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --from-beginning --topic test
读取到信息

至此，简单安装使用完成！！！！！！！！

注：

查看帮助手册：

bin/kafka-console-consumer.sh help