【运维技术】从零开始搭建开发使用的Kafka环境

【原创】从零开始搭建开发使用的Kafka环境

入门资料

1. 百度百科：
   Kafka是一种高吞吐量的分布式发布订阅消息系统，这些数据一般是因为吞吐量的要求而经过处理日志和日志聚合来解决。 对于像Hadoop的同样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是经过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了经过集群来提供实时的消费。

2. 归属公司
   Apache Kafka
   软件语言：scala

3. 相关术语介绍

• Broker： Kafka集群包含一个或多个服务器，这种服务器被称为broker[
• Topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不一样Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic便可生产或消费数据而没必要关心数据存于何处）
• Partition：Partition是物理上的概念，每一个Topic包含一个或多个Partition.
• Producer：赋值发布消息到负责发布消息到Kafka broker（生产者）
• Consumer：消息消费者，向Kafka broker读取消息的客户端。
• Consumer Group:每一个Consumer属于一个特定的Consumer Group（可为每一个Consumer指定group name，若不指定group name则属于默认的group）。

目标以及流程：

1. 单机搭建kafka
2. 集群搭建kafka（下一步）
3. 搭建kafka控制台（下一步）

kafka控制台选型：

1. Kafka Web Console
2. Kafka Manager
3. KafkaOffsetMonitor
   比较：
   若只须要监控功能，推荐使用KafkaOffsetMonito，若偏重Kafka集群管理，推荐使用Kafka Manager。
   由于都是开源程序，稳定性欠缺。故需先了解清楚目前已存在哪些Bug，多测试一下，避免出现相似于Kafka Web Console的问题。

开始单机搭建kafka：

1. 官网：http://kafka.apache.org/intro
2. 学习官方网站的快速启动教程：http://kafka.apache.org/quickstart
3. 官网的教程比较有服务器上的测试

开始前的备注

# 查看防火墙状态
systemctl status firewalld
# 关闭防火墙
service firewalld stop  
# 启动防火墙
service firewalld start

首先要确认你已经安装了java环境

# 检查java的命令
java -version

1：下载kafka并解压

# 获取kafka最新安装包，这边使用的是镜像地址，能够去官方网站得到最新地址
wget http://mirrors.hust.edu.cn/apache/kafka/0.11.0.1/kafka_2.11-0.11.0.1.tgz
# 解压程序
tar -xzf kafka_2.11-0.11.0.1.tgz
# 进入目录
cd kafka_2.11-0.11.0.1

配置对应的配置文件，server.properties

# 配置服务器zk地址
zookeeper.connect=localhost:2181
# 配置内网绑定关系
listeners=PLAINTEXT://<your.ip>:9092
# 配置外网绑定关系
advertised.listeners=PLAINTEXT://your.host.name:9092
# 配置kafka使用内存kafka-server-start.sh
# 在start中加入jvm的启动参数，默认是1G
export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"

2：启动服务器

kafka须要使用zookeeper，因此你首先须要启动一个zookeeper的服务，若是你没有的话，就使用kafka内置的脚原本启动一个单节点的zookeeper的实例
加入& 使进程常驻在内存中
默认端口：9092
默认为localhost，若是不配置对应的服务器ip的话

#执行快速启动zookeeper，经过内置的zookeeper进行启动，若是要zookeeper服务器的话吗，须要再server.properties的配置文件里面加入zookeeper.connect = 你的服务器内网ip:2181
bin/zookeeper-server-start.sh config/zookeeper.properties &

[2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
...

而后启动kafka的服务器：经过配置文件启动kafka

# 启动kafka
# server.properties的配置文件中有一个项目： host.name须要配置成为你的内网服务器ip地址，访问的时候经过外网环境经过外网ip地址访问，内网环境经过内网地址访问
bin/kafka-server-start.sh config/server.properties &
[2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties)
[2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties)
...

3：建立一个topic名字叫作test

# 经过脚本命令建立一个主题为test的，而且使用的zookeeper的地址为localhost的
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

确认一下这个topic是否含有完毕

# 经过zookeeper的地址来访问对应的topics中的主题列表
bin/kafka-topics.sh --list --zookeeper localhost:2181
test

4：启动一个生产者发送一些消息过去

# 启动客户端推送对应的消息到服务器的kafka提供的端口
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

This is a message
This is another message

5：启动一个消费者获取对应主题的消息

# 启动客户端获取对应服务端信息的地址来消费消息，使用pull的方式，每间隔0.1s进行一次服务器获取
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

This is a message
This is another message

6：将kafka建立集群节点（暂时省略）

7：使用kafka链接进行导入导出数据（暂时省略）

8：使用kafka的流去处理数据（暂时省略）

使用场景切换：本地服务器，变成真实服务器，首先提供外部调用，应该使用的是服务器的地址

这个是在服务器本地测试场景：
切换成服务器场景的状况下，须要首先在将server.properties的配置文件中的
配置方式修正为服务器的内网ip地址，对外提供的外网ip地址会进行映射，映射到最终的内网地址中去
新版的只须要修改以下两个配置：参考文章
http://blog.csdn.net/chenxun_2010/article/details/72626618
zookeeper.connect = localhost:9092
listeners = PLAINTEXT://ip:9092

java项目进行场景测试：

服务器kafka版本：2.11- 0.11.1
客户端kafka版本：0.11.0.1

因此去maven中寻找对应的版本的jar包进行使用


org.apache.kafka
kafka-clients
0.11.0.1

查看具体的api调用：

1. 生产者：org.apache.kafka.clients.producer.KafkaProducer.class
2. 消费者：org.apache.kafka.clients.consumer.KafkaConsumer.class

包结构：
clients
admin
consumer：KafkaProducer（实现类）
producer：KafkaConsumer（实现类）
otherclass
common
server.policy（服务公用）
在具体实现类里面源码中有启动的示例代码再类的头部注释中

遇到了一个问题：
Failed to load class "org.slf4j.impl.StaticLoggerBinder".

kafka默认引用再java类中在程序中引入了org\slf4j\slf4j-api\1.7.25\slf4j-api-1.7.25.jar的jar包
引入对应的实现日志的框架使用logback框架

再pom的配置文件中加入对应的依赖
而且debug的日志等级很烦人，因此就加入了配置文件
参考文章：http://www.cnblogs.com/h--d/p/5668152.html

加入logback的库依赖引用
至少须要引用三个模块：
logback-classic
logback-core
logback-access
这三个模块的内容
其中参考了这篇文章以为很详细，因此就提供出来：
http://www.cnblogs.com/warking/p/5710303.html

使用一个新的工程进行测试，一个启动消费者，一个启动生产者，编写对应的代码

1. 编写单元测试读取配置创建链接，发送消息
2. 生产者消费者所须要的依赖jar包
3. 启动线程消费者拉取消息成功消费（实现消息队列的功能）

源码示例你们能够去github上拉取：

https://github.com/fly-piglet/kafkastudy