搭建高吞吐量 Kafka 分布式发布订阅消息 集群

2017/08/29 Kafka

搭建高吞吐量 Kafka 分布式发布订阅消息 集群

简介

Kafka 是一种高吞吐的分布式发布订阅消息系统,能够替代传统的消息队列用于解耦合数据处理,缓存未处理消息等,同时具有更高的吞吐率,支持分区、多副本、冗余,因此被广泛用于大规模消息数据处理应用。Kafka 支持Java 及多种其它语言客户端,可与Hadoop、Storm、Spark等其它大数据工具结合使用。

特性

Kafka是一种高吞吐量的分布式发布订阅消息系统,有如下特性:

  • 通过O(1)的磁盘数据结构提供消息的持久化,这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。
  • 高吞吐量:即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。
  • 支持通过Kafka服务器和消费机集群来分区消息。
  • 支持Hadoop并行数据加载。

主要功能

  • 发布和订阅消息流,这个功能类似于消息队列,这也是kafka归类为消息队列框架的原因

  • 以容错的方式记录消息流,kafka以文件的方式来存储消息流

  • 可以再消息发布的时候进行处理

使用场景

在系统或应用程序之间构建可靠的用于传输实时数据的管道,消息队列功能

构建实时的流数据处理程序来变换或处理数据流,数据处理功能

相关术语介绍

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)

环境

Zookeeper集群: 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181

kafka 集群: 192.168.252.124 , 192.168.252.125 , 192.168.252.126

kafka-manager: 192.168.252.127

主机名修改

CentOs7.3 修改主机名

ssh 免密登录

CentOs7.3 ssh 免密登录

安装 JDK1.8

CentOs7.3 安装 JDK1.8

搭建 Zookeeper 集群

CentOs7.3 搭建 ZooKeeper-3.4.9 Cluster 集群服务

Zookeeper集群: 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181

主机名依次被我修改成: node1,node2,node3

搭建 kafka 集群

kafka 集群: 192.168.252.124 , 192.168.252.125 , 192.168.252.126

主机名依次被我修改成: node4,node5,node6

1.下载代码

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

下载最新版本的kafka ,我在北京我就选择,清华镜像比较快

清华镜像:https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/

阿里镜像:https://mirrors.aliyun.com/apache/kafka/

$ cd /opt
$ wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/0.11.0.0/kafka_2.12-0.11.0.0.tgz
$ tar -zxvf kafka_2.12-0.11.0.0.tgz
$ cd kafka_2.12-0.11.0.0

2.修改配置

在 node4 操作

$ vi /opt/kafka_2.12-0.11.0.0/config/server.properties
broker.id=0  每台服务器不能重复

#设置zookeeper的集群地址
zookeeper.connect=192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181

把配置复制到 node5,node6 集群

$ for a in {5..6} ; do scp -r /opt/kafka_2.12-0.11.0.0/ node$a:/opt/kafka_2.12-0.11.0.0 ; done

修改 node5,node6 集群的broker.id

$ vi /opt/kafka_2.12-0.11.0.0/config/server.properties

3.启动kafka

在 node1,启动 Kafka使用的 ZooKeeper,所以先启动ZooKeeper服务器

$ for a in {1..3} ; do ssh node$a "source /etc/profile;  /opt/zookeeper-3.4.9/bin/zkServer.sh start" ; done

现在 node4 启动Kafka服务器

$ for a in {4..6} ; do ssh node$a "source /etc/profile;  /opt/kafka_2.12-0.11.0.0/bin/kafka-server-start.sh /opt/kafka_2.12-0.11.0.0/config/server.properties" ; done

或者后台启动运行,日志查看去Kafka解压目录有个log 文件夹查看

$ for a in {4..6} ; do ssh node$a "source /etc/profile; nohup  /opt/kafka_2.12-0.11.0.0/bin/kafka-server-start.sh /opt/kafka_2.12-0.11.0.0/config/server.properties > /dev/null 2>&1 &" ; done

查看进程,Kafka 是否启动成功

$ jps
3825 Kafka
6360 Jps

如果报错删除

kafka.common.KafkaException: Failed to acquire lock on file .lock in /tmp/kafka-logs. A Kafka instance in another process or thread is using this directory.

$ rm -rf /tmp/kafka-logs

4.创建主题

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-topics.sh --create --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181 --replication-factor 2 --partitions 1 --topic ymq

–replication-factor 2 #复制两份
–partitions 1 #创建1个分区
–topic #主题为ymq

运行list topic命令,可以看到该主题:

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-topics.sh --list --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181

5.生产消息

Kafka附带一个命令行客户端,它将从文件或标准输入中输入,并将其作为消息发送到Kafka群集。默认情况下,每行将作为单独的消息发送。

node5 运行生产者,然后在控制台中输入一些消息以发送到服务器。

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic ymq
>www.ymq.io

6.消费消息

node6 运行消费者,将把消息转储到标准输出。

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-console-consumer.sh --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181  --topic ymq --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper]. 
www.ymq.io

7.topic详情

用describe 查看集群中topic每个节点情况

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-topics.sh --describe --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181 --topic ymq
Topic:ymq	PartitionCount:1	ReplicationFactor:2	Configs:
	Topic: ymq	Partition: 0	Leader: 1	Replicas: 1,3	Isr: 3,1

以下是输出的说明。第一行给出了所有分区的摘要,每个附加行提供有关一个分区的信息。由于我们这个主题只有一个分区,只有一行。

leader负责给定分区的读取和写入分配节点编号,每个分区的部分数据会随机指定不同的节点 replicas是复制此分区的日志的节点列表 isr一组正在同步的副本列表

8.删除topic

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-topics.sh --delete --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181 --topic ymq
Topic ymq is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

9.停止kafka

$ for a in {4..6} ; do ssh node$a "source /etc/profile;  /opt/kafka_2.12-0.11.0.0/bin/kafka-server-stop.sh /opt/kafka_2.12-0.11.0.0/config/server.properties " ; done

部署 Kafka Manager

Yahoo开源Kafka集群管理器Kafka Manager

作为一个分布式的消息发布-订阅系统,Apache Kafka在Yahoo内部已经被很多团队所使用,例如媒体分析团队就将其应用到了实时分析流水线中,同时,Yahoo整个Kafka集群处理的峰值带宽超过了20Gbps(压缩数据)。为了让开发者和服务工程师能够更加简单地维护Kafka集群,Yahoo构建了一个基于Web的管理工具,称为Kafka Manager,日前该项目已经在GitHub上开源。

通过Kafka Manager用户能够更容易地发现集群中哪些主题或者分区分布不均匀,同时能够管理多个集群,能够更容易地检查集群的状态,能够创建主题,执行首选的副本选择,能够基于集群当前的状态生成分区分配,并基于生成的分配执行分区的重分配,此外,Kafka Manager还是一个非常好的可以快速查看集群状态的工具。

Kafka Manager使用Scala语言编写,其Web控制台基于Play Framework实现,除此之外,Yahoo还迁移了一些Apache Kafka的帮助程序以便能够与Apache Curator框架一起工作。

一、它支持以下内容:

  • 管理多个群集
  • 容易检查集群状态(主题,消费者,偏移量,经纪人,副本分发,分区分配)
  • 运行首选副本选举
  • 使用选项生成分区分配,以选择要使用的代理
  • 运行分区的重新分配(基于生成的分配)
  • 创建可选主题配置的主题(0.8.1.1具有不同于0.8.2+的配置)
  • 删除主题(仅支持0.8.2+,并记住在代理配​​置中设置delete.topic.enable = true)
  • 主题列表现在表示标记为删除的主题(仅支持0.8.2+)
  • 批量生成多个主题的分区分配,并选择要使用的代理
  • 批量运行多个主题的分区重新分配
  • 将分区添加到现有主题
  • 更新现有主题的配置
  • 可选地,启用JMX轮询代理级和主题级度量。
  • 可选地筛选出在zookeeper中没有ids / owner /&offset /目录的消费者。

源码,并编译打包

在 kafka-manager: 192.168.252.127 node7 部署

编译超级慢

$ yum install git
$ cd /opt/
$ git clone https://github.com/yahoo/kafka-manager
$ cd kafka-manager/
$ ./sbt clean dist

下载编译好的包

反正我是没编译成功,从网上找了一个编译好的

链接: 百度网盘下载 密码: kha2

$ yum install unzip
$ unzip kafka-manager-1.3.2.1.zip
$ vi /opt/kafka-manager-1.3.2.1/conf/application.conf

修改这个 zk 地址

kafka-manager.zkhosts="192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181"

启动 kafka-manager

默认端口 NettyServer - Listening for HTTP on /0:0:0:0:0:0:0:0:9000

$ /opt/kafka-manager-1.3.2.1/bin/kafka-manager -Dconfig.file=conf/application.conf

或者后台运行 并且配置端口

$ nohup bin/kafka-manager  -Dconfig.file=/home/hadoop/app/kafka-manager-1.3.2.1/conf/application.conf -Dhttp.port=9000 &

访问:http://ip:9000/

Contact

  • 作者:鹏磊
  • 出处:http://www.ymq.io
  • Email:admin@souyunku.com
  • 版权归作者所有,转载请注明出处
  • Wechat:关注公众号,搜云库,专注于开发技术的研究与知识分享

关注公众号-搜云库

Show Disqus Comments

Search

    Table of Contents