标签：ati   load   star   environ   pac   生产者   ...   JD   新建   

Kafka笔记

安装

安装依赖JDK

Java

Kafka的安装需要java环境

如没有Java 须先安装java

1、检查一下系统中的jdk版本

java –version

显示:

openjdk version "1.8.0_102"

OpenJDK Runtime Environment (build 1.8.0_102-b14)

OpenJDK 64-Bit Server VM (build 25.102-b14, mixed mode)

2、检测jdk安装包

rpm -qa | grep java

3. 卸载openjdk

yum remove *openjdk*

之后再次输入rpm -qa | grep java 查看卸载情况：

4. 安装新的jdk

首先到jdk官网上下载你想要的jdk版本，下载完成之后将需要安装的jdk安装包放到Linux系统指定的文件夹下，并且命令进入该文件夹下：

http://download.oracle.com/otn-pub/java/jdk/8u171-b11/512cd62ec5174c3487ac17c61aaa89e8/jdk-8u171-linux-i586.tar.gz

mkdir -p /usr/lib/jvm

tar -zxvf jdk-8u131-linux-x64.tar.gz -C /usr/lib/jvm

5.环境设置环境

vim /etc/profile

在最前面添加：

export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_131

export JRE_HOME=${JAVA_HOME}/jre

export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib

export PATH=${JAVA_HOME}/bin:$PATH

6、执行profile文件

source /etc/profile

7、检查新安装的jdk

java –version

安装并简单启动kafka

下载

下载: http://kafka.apache.org/downloads

选中https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.0/kafka_2.12-1.1.0.tgz

解压

下载好kafka安装包后，将其解压到/usr/local目录下，删除压缩包

tar –zxvf kafka_2.11-1.1.0.tgz –C /usr/local

进入目录

cd /usr/local/kafka_2.11-1.1.0

启动zookeeper

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

默认使用的2181端口，可在配置文件修改。

启动kafka server

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

(非本地生产者和消费者访问Kafka，记得修改 config/server.properties中的listeners, 例如

listeners=PLAINTEXT://192.168.1.253:9092)

创建topic

创建一个名为"test"的Topic，只有一个分区和一个备份

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

查看topic

bin/kafka-topics.sh --list --zookeeper localhost:2181

发送消息

bin/kafka-console-producer.sh --broker-list 192.168.1.253:9092 --topic test

输入消息:

Aaaaaa

消费消息

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

如果你有2台不同的终端上运行上述命令，那么当你在运行生产者时，消费者就能消费到生产者发送的消息。

单机设置多个集群

到目前，我们只是单一的运行一个broker，没什么意思。对于Kafka，一个broker仅仅只是一个集群的大小，所有让我们多设几个broker。

首先为每个broker创建一个配置文件:

> cp config/server.properties config/server-1.properties

> cp config/server.properties config/server-2.properties

现在编辑这些新建的文件，设置以下属性：

config/server-1.properties:

broker.id=1

listeners=PLAINTEXT:// 192.168.1.253:9093

log.dir=/tmp/kafka-logs-1

config/server-2.properties:

broker.id=2

listeners=PLAINTEXT://192.168.1.253:9094

log.dir=/tmp/kafka-logs-2

broker.id是集群中每个节点的唯一且永久的名称，我们修改端口和日志目录是因为我们现在在同一台机器上运行，我们要防止broker在同一端口上注册和覆盖对方的数据。

我们已经运行了zookeeper和刚才的一个kafka节点，所有我们只需要在启动2个新的kafka节点。

> bin/kafka-server-start.sh config/server-1.properties &

...

> bin/kafka-server-start.sh config/server-2.properties &

...

现在，我们创建一个新topic，把备份设置为：3

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

好了，现在我们已经有了一个集群了，我们怎么知道每个集群在做什么呢？运行命令"describe topics"

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic

输出:

Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:

Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0

输出解释：第一行是所有分区的摘要，其次，每一行提供一个分区信息，因为我们只有一个分区，所以只有一行。

"leader"：该节点负责该分区的所有的读和写，每个节点的leader都是随机选择的。

"replicas"：备份的节点列表，无论该节点是否是leader或者目前是否还活着，只是显示。

"isr"："同步备份"的节点列表，也就是活着的节点并且正在同步leader。

我们运行这个命令，看看一开始我们创建的那个节点：

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

Topic:test PartitionCount:1 ReplicationFactor:1 Configs:

Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0

这并不奇怪，刚才创建的主题没有Replicas，并且在服务器"0"上，我们创建它的时候，集群中只有一个服务器，所以是"0"。

让我们来发布一些信息在新的topic上：

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic

...

my test message 1

my test message 2

^C

Kafka笔记

标签：ati   load   star   environ   pac   生产者   ...   JD   新建   

原文地址：https://www.cnblogs.com/songanwei/p/9037883.html