官网:http://kafka.apache.org/java
消息的生产者将消息推送到kafka集群,消息的消费者从kafka集群中拉取消息。node
说明:git
准备三台虚拟机,分别是node01,node02,node03,而且修改hosts文件以下:github
~~~shellvim /etc/hostsshell
192.168.40.133 node01192.168.40.134 node02192.168.40.135 node03apache
yum install -y rdaterdate -s time-b.nist.gov~~~bootstrap
因为Kafka 是用Scala 语言开发的,运行在JVM上,所以在安装Kafka 以前须要先安装JDK 。vim
安装过程略过,我这里使用的是jdk1.8。api
Kafka 依赖ZooKeeper ,经过ZooKeeper 来对服务节点、消费者上下线管理、集群、分区元数据管理等,所以ZooKeeper 也是Kafka 得以运行的基础环境之一。浏览器
#上传zookeeper-3.4.9.tar.gz到/export/software
cd /export/software
mkdir -p /export/servers/
tar -xvf zookeeper-3.4.9.tar.gz -C /export/servers/
#建立ZooKeeper的data目录
mkdir /export/data/zookeeper -p
cd /export/servers/zookeeper-3.4.9/conf/
#修改配置文件
mv zoo_sample.cfg zoo.cfg
vim zoo.cfg
#设置data目录
dataDir=/export/data/zookeeper
#启动ZooKeeper
./zkServer.sh start
#检查是否启动成功
jps复制代码
~~~shell
vim /export/data/zookeeper/myid
vim zoo.cfg
server.1=node01:2888:3888server.2=node02:2888:3888server.3=node03:2888:3888~~~
~~~shellvim /etc/profileexport ZK_HOME=/export/servers/zookeeper-3.4.9export PATH=${ZK_HOME}/bin:$PATH
source /etc/profile~~~
~~~shellscp /etc/profile node02:/etc/scp /etc/profile node03:/etc/
cd /export/serversscp -r zookeeper-3.4.9 node02:/export/servers/scp -r zookeeper-3.4.9 node03:/export/servers/~~~
~~~shellmkdir -p /export/servers/onekey/zkvim slave
node01node02node03
vim startzk.sh
cat /export/servers/onekey/zk/slave | while read linedo{echo "开始启动 --> "$linessh $line "source /etc/profile;nohup sh ${ZK_HOME}/bin/zkServer.sh start >/dev/null 2>&1 &"}&waitdoneecho "★★★启动完成★★★"
vim stopzk.sh
cat /export/servers/onekey/zk/slave | while read linedo{echo "开始中止 --> "$linessh $line "source /etc/profile;nohup sh ${ZK_HOME}/bin/zkServer.sh stop >/dev/null 2>&1 &"}&waitdoneecho "★★★中止完成★★★"
chmod +x startzk.sh stopzk.sh
export ZK_ONEKEY=/export/servers/onekeyexport PATH=${ZK_ONEKEY}/zk:$PATH~~~
发现三台机器都有“QuorumPeerMain”进程,说明机器已经启动成功了。
检查集群是否正常:
zkServer.sh status
发现,集群运行一切正常。
第一步:上传Kafka安装包而且解压
~~~shellrz 上传kafka_2.11-1.1.0.tgz到 /export/software/cd /export/software/tar -xvf kafka_2.11-1.1.0.tgz -C /export/servers/cd /export/serversmv kafka_2.11-1.1.0/ kafka~~~
第二步:配置环境变量
~~~shellvim /etc/profile
export KAFKA_HOME=/export/servers/kafkaexport PATH=${KAFKA_HOME}/bin:$PATH
source /etc/profile~~~
第三步:修改配置文件
~~~shellcd /export/servers/kafkacd configvim server.properties
broker.id=0
log.dirs=/export/data/kafka
zookeeper.connect=node01:2181
~~~
第四步:启动kafka服务
~~~shell
kafka-server-start.sh -daemon /export/servers/kafka/config/server.properties~~~
第五步:检测kafka是否启动
若是进程中有名为kafka的进程,就说明kafka已经启动了。
因为kafka是将元数据保存在ZooKeeper中的,因此,能够经过查看ZooKeeper中的信息进行验证kafka是否安装成功。
Kafka Manager 由 yahoo 公司开发,该工具能够方便查看集群 主题分布状况,同时支持对 多个集群的管理、分区平衡以及建立主题等操做。
源码托管于github:https://github.com/yahoo/kafka-manager
第一步:上传Kafka-manager安装包而且解压
~~~shellrz上传kafka-manager-1.3.3.17.tar.gz到 /export/software/cd /export/softwaretar -xvf kafka-manager-1.3.3.17.tar.gz -C /export/servers/cd /export/servers/kafka-manager-1.3.3.17/conf~~~
第二步:修改配置文件
~~~shell
vim application.conf
http.port=19000
kafka-manager.zkhosts="node01:2181"
~~~
第三步:启动服务
~~~shellcd /export/servers/kafka-manager-1.3.3.17/bin
./kafka-manager -Dconfig.file=../conf/application.conf
vim /etc/profileexport KAFKAMANAGEHOME=/export/servers/kafka-manager-1.3.3.17export PATH=${KAFKAMANAGEHOME}/bin:$PATH
source /etc/profile
cd /export/servers/onekey/mkdir kafka-managercd kafka-managervim start-kafka-manager.shnohup kafka-manager -Dconfig.file=${KAFKAMANAGEHOME}/conf/application.conf >/dev/null 2>&1 &chmod +x start-kafka-manager.shvim /etc/profileexport PATH=${ZK_ONEKEY}/kafka-manager:$PATHsource /etc/profile
~~~
第四步:检查是否启动成功
打开浏览器,输入地址:http://node01:19000/,便可看到kafka-manage管理界面。
进入管理界面,是没有显示Cluster信息的,须要添加后才能操做。
输入Cluster Name、ZooKeeper信息、以及Kafka的版本信息(这里最高只能选择1.0.0)。
点击Save按钮保存。
添加成功。
kafka集群的搭建是很是简单的,只须要将上面的单机版的kafka分发的其余机器,而且将ZooKeeper信息修改为集群的配置以及设置不一样的broker值便可。
第一步:将kafka分发到node0二、node03
~~~cd /export/servers/scp -r kafka node02:/export/servers/scp -r kafka node03:/export/servers/scp /etc/profile node02:/etc/scp /etc/profile node03:/etc/
source /etc/profile~~~
第二步:修改node0一、node0二、node03上的kafka配置文件
~~~shellcd /export/servers/kafka/configvim server.propertieszookeeper.connect=node01:2181,node02:2181,node03:2181~~~
~~~shellcd /export/servers/kafka/configvim server.propertiesbroker.id=1zookeeper.connect=node01:2181,node02:2181,node03:2181~~~
~~~shellcd /export/servers/kafka/configvim server.propertiesbroker.id=2zookeeper.connect=node01:2181,node02:2181,node03:2181~~~
第三步:编写一键启动、中止脚本。注意:该脚本依赖于环境变量中的KAFKA_HOME。
~~~shellmkdir -p /export/servers/onekey/kafkavim slave
node01node02node03
vim start-kafka.sh
cat /export/servers/onekey/kafka/slave | while read linedo{echo "开始启动 --> "$linessh $line "source /etc/profile;nohup sh ${KAFKAHOME}/bin/kafka-server-start.sh -daemon ${KAFKAHOME}/config/server.properties >/dev/null 2>&1 &"}&waitdoneecho "★★★启动完成★★★"
chmod +x start-kafka.sh
vim stop-kafka.sh
cat /export/servers/onekey/kafka/slave | while read linedo{echo "开始中止 --> "$linessh $line "source /etc/profile;nohup sh ${KAFKA_HOME}/bin/kafka-server-stop.sh >/dev/null 2>&1 &"}&waitdoneecho "★★★中止完成★★★"
chmod +x stop-kafka.sh
export PATH=${ZK_ONEKEY}/kafka:$PATHsource /etc/profile~~~
第四步:经过kafka-manager管理工具查看集群信息。
因而可知,kafka集群已经启动完成。
对kafka的操做有2种方式,一种是经过命令行方式,一种是经过API方式。
Kafka在bin目录下提供了shell脚本文件,能够对Kafka进行操做,分别是:经过命令行的方式,咱们将体验下kafka,以便咱们对kafka有进一步的认知。
~~~shellkafka-topics.sh --create --zookeeper node01:2181 --replication-factor 1 --partitions 1 --topic my-kafka-topic
Created topic "my-kafka-topic".~~~
参数说明:
~~~shellkafka-topics.sh --list --zookeeper node01:2181
__consumer_offsetsmy-kafka-topic~~~
能够查看列表。
若是须要查看topic的详细信息,须要使用describe命令。
~~~shellkafka-topics.sh --describe --zookeeper node01:2181 --topic test-topic
kafka-topics.sh --describe --zookeeper node01:2181~~~
经过kafka-topics.sh执行删除动做,须要在server.properties文件中配置 delete.topic.enable=true,该配置默认为 false。
不然执行该脚本并未真正删除主题 ,将该topic标记为删除状态 。
~~~shellkafka-topics.sh --delete --zookeeper node01:2181 --topic my-kafka-topic
[root@node01 config]# kafka-topics.sh --delete --zookeeper node01:2181 --topic my-kafka-topicTopic my-kafka-topic is marked for deletion.Note: This will have no impact if delete.topic.enable is not set to true.
[root@node01 config]# kafka-topics.sh --delete --zookeeper node01:2181 --topic my-kafka-topic2Topic my-kafka-topic2 is marked for deletion.Note: This will have no impact if delete.topic.enable is not set to true.
~~~
~~~shellkafka-console-producer.sh --broker-list node01:9092 --topic my-kafka-topic~~~
能够看到,已经向topic发送了消息。
~~~shellkafka-console-consumer.sh --bootstrap-server node01:9092 --topic my-kafka-topic
kafka-console-consumer.sh --bootstrap-server node01:9092 --from-beginning --topic my-kafka-topic~~~
除了经过命令行的方式操做kafka外,还能够经过Java api的方式操做,这种方式将更加的经常使用。
导入依赖:
~~~xml
<artifactId>itcast-bigdata-kafka</artifactId>复制代码
<dependencies>复制代码
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>1.1.0</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>1.1.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>复制代码
</dependencies>复制代码
<build>
<plugins>
<!-- java编译插件 -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.2</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
</plugins>
</build>
复制代码
~~~
因为主题的元数据信息是注册在 ZooKeeper 相 应节点之中,因此对主题的操做实质是对 ZooKeeper 中记录主题元数据信息相关路径的操做。 Kafka将对 ZooKeeper 的相关操做封装成一 个 ZkUtils 类 , 井封装了一个AdrninUtils 类调用 ZkClient 类的相关方法以实现对 Kafka 元数据 的操做,包括对主题、代理、消费者等相关元数据的操做。对主题操做的相关 API调用较简单, 相应操做都是经过调用 AdminUtils类的相应方法来完成的。
~~~javapackage cn.itcast.kafka;
import kafka.admin.AdminUtils;import kafka.utils.ZkUtils;import org.apache.kafka.common.security.JaasUtils;import org.junit.Test;
import java.util.Properties;
public class TestKafkaTopic {
@Test
public void testCreateTopic() {
ZkUtils zkUtils = null;
try {
//参数:zookeeper的地址,session超时时间,链接超时时间,是否启用zookeeper安全机制
zkUtils = ZkUtils.apply("node01:2181", 30000, 3000, JaasUtils.isZkSecurityEnabled());复制代码
String topicName = "my-kafka-topic-test1";
if (!AdminUtils.topicExists(zkUtils, topicName)) {
//参数:zkUtils,topic名称,partition数量,副本数量,参数,机架感知模式
AdminUtils.createTopic(zkUtils, topicName, 1, 1, new Properties(), AdminUtils.createTopic$default$6());
System.out.println(topicName + " 建立成功!");
} else {
System.out.println(topicName + " 已存在!");
}
} finally {
if (null != zkUtils) {
zkUtils.close();
}
}复制代码
}
}复制代码
~~~
测试结果:
~~~java@Testpublic void testDeleteTopic() {ZkUtils zkUtils = null;try {//参数:zookeeper的地址,session超时时间,链接超时时间,是否启用zookeeper安全机制zkUtils = ZkUtils.apply("node01:2181", 30000, 3000, JaasUtils.isZkSecurityEnabled());String topicName = "my-kafka-topic-test1";if (AdminUtils.topicExists(zkUtils, topicName)) {//参数:zkUtils,topic名称AdminUtils.deleteTopic(zkUtils, topicName);System.out.println(topicName + " 删除成功!");} else {System.out.println(topicName + " 不已存在!");}} finally {if (null != zkUtils) {zkUtils.close();}}
}
~~~复制代码
测试结果:
~~~javapackage cn.itcast.kafka;
import org.apache.kafka.clients.producer.*;import org.apache.kafka.common.serialization.StringSerializer;import org.junit.Test;
import java.util.Properties;
public class TestProducer {
@Test
public void testProducer() throws InterruptedException {
Properties config = new Properties();复制代码
// 设置kafka服务列表,多个用逗号分隔
config.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092");
// 设置序列化消息 Key 的类
config.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 设置序列化消息 value 的类
config.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());复制代码
// 初始化
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(config);
for (int i = 0; i < 100 ; i++) {
ProducerRecord record = new ProducerRecord("my-kafka-topic","data-" + i);
// 发送消息
kafkaProducer.send(record);
System.out.println("发送消息 --> " + i);复制代码
Thread.sleep(100);
}复制代码
kafkaProducer.close();复制代码
}复制代码
}
~~~
~~~javapackage cn.itcast.kafka;
import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.serialization.StringDeserializer;import org.apache.kafka.common.serialization.StringSerializer;import org.junit.Test;
import javax.sound.midi.Soundbank;import java.util.Arrays;import java.util.Properties;
public class TestConsumer {
@Test
public void testConsumer() {
Properties config = new Properties();
// 设置kafka服务列表,多个用逗号分隔
config.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092");
// 设置消费者分组id
config.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
// 设置序反列化消息 Key 的类
config.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 设置序反列化消息 value 的类
config.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
复制代码
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(config);
// 订阅topic
kafkaConsumer.subscribe(Arrays.asList("my-kafka-topic"));复制代码
while (true) { // 使用死循环不断的拉取数据
ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
String value = record.value();
long offset = record.offset();
System.out.println("value = " + value + ", offset = " + offset);
}
}复制代码
}
}
~~~
复制代码
什么是Kafka?Kafka监控工具汇总Kafka快速入门Kafka核心之ConsumerKafka核心之Producer
替代Flume——Kafka Connect简介最简单流处理引擎——Kafka Streams简介
更多实时计算,Flink,Kafka等相关技术博文,欢迎关注实时流式计算