下载

http://kafka.apache.org/downloads.html

http://mirror.bit.edu.cn/apache/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz

root@master:/usr/local/kafka_2.11-0.11.0.0/config# vim server.properties

broker.id=2 每个节点不同

log.retention.hours=168

message.max.byte=5242880

default.replication.factor=2

replica.fetch.max.bytes=5242880

zookeeper.connect=master:2181,slave1:2181,slave2:2181

复制到其它节点

注意要事先在zk中创建/kafka节点,否则会报出错误:java.lang.IllegalArgumentException: Path length must be > 0

root@master:/usr/local/zookeeper-3.4.9# bin/zkCli.sh -server master

[zk: master(CONNECTED) 7] create /kafka ''

Created /kafka

[zk: master(CONNECTED) 8] ls /

[cluster,controller,controller_epoch,brokers,zookeeper,kafka,admin,isr_change_notification,consumers,latest_producer_id_block,config]

[zk: master(CONNECTED) 9] ls /kafka

[]

以守护进程方式启动kafka

root@master:/usr/local/kafka_2.11-0.11.0.0# nohup bin/kafka-server-start.sh config/server.properties &

创建topic:

root@slave2:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh--create--zookeepermaster:2181--replication-factor 1--partitions1--topic test

Created topic "test".

列出所有topic:

root@slave2:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --list --zookeeper master:2181

test

发送消息

root@master:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-console-producer.sh --broker-list master:9092 --topic test

>this is a message

>this is ant^H message

消费消息

root@slave1:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-console-consumer.sh --zookeeper master:2181 --topic test --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].

this is a message

this is an message

查看集群状态信息

root@slave1:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --describe --zookeeper slave1:2181 --topic my-replicated-topic

Topic:my-replicated-topicPartitionCount:1ReplicationFactor:3Configs:

Topic: my-replicated-topicPartition: 0leader: 3Replicas: 1,3,2Isr: 3,2

安装kafka-manager

root@master:/usr/local/kafka_2.11-0.11.0.0# git clone https://github.com/yahoo/kafka-manager

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager# cd kafka-manager/

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager# ./sbt clean dist

[success] Total time: 3453 s,completed Aug 7,2017 8:48:15 PM

打包好的文件存在

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager/target/universal# ls

kafka-manager-1.3.3.12.ziptmp

修改kafka-manager配置文件

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# vim conf/application.conf

kafka-manager.zkhosts="192.168.117.243:2181,192.168.117.45:2181,192.168.117.242:2181"

启动kafka-manager

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# bin/kafka-manager -Dconfig.file=conf/application.conf

推荐启动方式:

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=7778 &

登录kafka manager:

http://192.168.117.243:7778/

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# netstat -antlup | grep 7778

tcp600 :::7778:::*LISTEN100620/java

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# bin/kafka-manager -Dconfig.file=conf/application.conf

This application is already running (Or delete /usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12/RUNNING_PID file).

停止kafka-manager

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# rm RUNNING_PID

root@master:/usr/local/kafka_2.11-0.11.0.0# cd kafka-manager-1.0-SNAPSHOT/

生产服务器配置

# Replication configurations

num.replica.fetchers=4

replica.fetch.max.bytes=1048576

replica.fetch.wait.max.ms=500

replica.high.watermark.checkpoint.interval.ms=5000

replica.socket.timeout.ms=30000

replica.socket.receive.buffer.bytes=65536

replica.lag.time.max.ms=10000

replica.lag.max.messages=4000

controller.socket.timeout.ms=30000

controller.message.queue.size=10

# Log configuration

num.partitions=8

message.max.bytes=1000000

auto.create.topics.enable=true

log.index.interval.bytes=4096

log.index.size.max.bytes=10485760

log.retention.hours=168

log.flush.interval.ms=10000

log.flush.interval.messages=20000

log.flush.scheduler.interval.ms=2000

log.roll.hours=168

log.retention.check.interval.ms=300000

log.segment.bytes=1073741824

# ZK configuration

zookeeper.connection.timeout.ms=6000

zookeeper.sync.time.ms=2000

# Socket server configuration

num.io.threads=8

num.network.threads=8

socket.request.max.bytes=104857600

socket.receive.buffer.bytes=1048576

socket.send.buffer.bytes=1048576

queued.max.requests=16

fetch.purgatory.purge.interval.requests=100

producer.purgatory.purge.interval.requests=100

Kafka是一种高吞吐量的分布式发布订阅的消息队列系统,原本开发自LinkedIn,用作LinkedIn的活动流(ActivityStream)和运营数据处理管道(Pipeline)的基础。现在它已被多家不同类型的公司作为多种类型的数据管道和消息系统使用。

1 Kafka消息队列简介

1.1 基本术语

  • broker
    Kafka集群包含一个或多个服务器,这种服务器被称为broker[5]

  • Topic
    每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic。(物理上不同Topic的消息分开存储,逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处)

  • Partition
    Partition是物理上的概念,每个Topic包含一个或多个Partition.(一般为kafka节点数cpu的总核数)

  • Producer
    负责发布消息到Kafka broker

  • Consumer
    消息消费者,向Kafka broker读取消息的客户端。

  • Consumer Group
    每个Consumer属于一个特定的Consumer Group(可为每个Consumer指定group name,若不指定group name则属于默认的group)。

1.2 消息队列

1.2.1基本特性

  1. 可扩展

  2. 在不需要下线的情况下进行扩容

  3. 数据流分区(partition)存储在多个机器上

  4. 高性能

  5. 单个broker就能服务上千客户端

  6. 单个broker每秒种读/写可达每秒几百兆字节

  7. 多个brokers组成的集群将达到非常强的吞吐能力

  8. 性能稳定,无论数据多大

  9. Kafka在底层摒弃了Java堆缓存机制,采用了操作系统级别的页缓存,同时将随机写操作改为顺序写,再结合Zero-copy的特性极大地改善了IO性能。

  10. 持久存储

  11. 存储在磁盘上

  12. 冗余备份到其他服务器上以防止丢失

1.2.2消息格式

  1. 一个topic对应一种消息格式,因此消息用topic分类

  2. 一个topic代表的消息有1个或者多个patition(s)组成

  3. 一个partition中

  4. 一个partition应该存放在一到多个server上

  • 一个server为leader

  • 其他servers为followers

  • leader需要接受读写请求

  • followers仅作冗余备份

  • leader出现故障,会自动选举一个follower作为leader,保证服务不中断

  • 每个server都可能扮演一些partitions的leader和其它partitions的follower角色,这样整个集群就会达到负载均衡的效果

  • 如果只有一个server,就没有冗余备份,是单机而不是集群

  • 如果有多个server


消息按顺序存放消息顺序不可变只能追加消息,不能插入每个消息都有一个offset,用作消息ID,在一个partition中唯一offset有consumer保存和管理,因此读取顺序实际上是完全有consumer决定的,不一定是线性的消息有超时日期,过期则删除

1.2.3生产者 producer

  • producer将消息写入kafka

  • 写入要指定topic和partition

  • 消息如何分到不同的partition,算法由producer指定

1.2.4消费者 consumer

  • consumer读取消息并作处理

  • consumer group


    • 可以按照partition的数目进行并发处理

    • 每个partition都只有一个consumer读取,因而保证了消息被处理的顺序是按照partition的存放顺序进行,注意这个顺序受到producer存放消息的算法影响

    • 这个概念的引入为了支持两种场景:每条消息分发一个消费者,每条消息广播给所有消费者

    • 多个consumer group订阅一个topic,该topci的消息广播给所有consumer group

    • 一条消息发送到一个consumer group后,只能由该group的一个consumer接收和使用

    • 一个group中的每个consumer对应一个partition可以带来如下好处


  • 一个Consumer可以有多个线程进行消费,线程数应不多于topic的partition数,因为对于一个包含一或多消费线程的consumer group来说,一个partition只能分给其中的一个消费线程消费,且让尽可能多的线程能分配到partition(不过实际上真正去消费的线程及线程数还是由线程池的调度机制来决定)。这样如果线程数比partition数多,那么单射分配也会有多出的线程,它们就不会消费到任何一个partition的数据而空转耗资源 。

  • 如果consumer从多个partition读到数据,不保证数据间的顺序性,kafka只保证在一个partition上数据是有序的,但多个partition,根据你读的顺序会有不同

  • 增减consumer,broker,partition会导致rebalance,所以rebalance后consumer对应的partition会发生变化

Ubuntu16.04安装Kafka集群的更多相关文章

  1. 深入云存储系统Swift核心组件:Ring实现原理剖析

    它的目的是用于托管Rackspace的CloudFilesservice,原始项目代号是swift,所以沿用至今。Ring是Swift中最重要的组件,用于记录存储对象与物理位置间映射关系。先来看一下Swift文档中关于Ring的描述:Ring用来确定数据驻留在集群中的位置。有单独对应于Account数据库、container数据库和单个object的ring。Ring使用zone的概念来保证数据的隔离。每个partition的replica都确保放在了不同的zone中。本文逐步深入探讨了Swift如何通过

  2. openstack swift ring文件解析

    一直以为对swift的ring文件充满好奇,这段时间重新研究swift,正好借此机会好好研究下。

  3. OpenStack对象存储――Swift

    Swift前身是RackspaceCloudFiles项目,随着Rackspace加入到OpenStack社区,于2010年7月贡献给OpenStack,作为该开源项目的一部分。Swift目前的最新版本是OpenStackEssex1.5.1。Swift特性在OpenStack官网中,列举了Swift的20多个特性,其中最引人关注的是以下几点。在OpenStack中还可以与镜像服务Glance结合,为其存储镜像文件。Auth认证服务目前已从Swift中剥离出来,使用OpenStack的认证服务Keysto

  4. swift中string操作

    在swift中得string操作和OC稍有不同。一些基本操作苹果文档已经有描述。但是关于index的操作则没有提到。如果想删除或者得到字串,首先需要得到String.Index这个类型和Int不同,不能直接转换,所以需要用到advance函数。比如上面的例子,就是获取最后一个字符,然后删除之。

  5. 对象存储系统Swift技术详解:综述与概念

    通过阅读Swift的技术文档,我们可以理解其中的设计的原理和实现的方法。本人于9月底开始接触swift,刚开始看文档的时候一知半解,有幸阅读了zzcase等人的博客,才得以入门。随着对swift设计原理的理解和源码的深入,文档经过数次反复的修改,希望对各位学习swift的童鞋有所帮助,水平有限,若各位发现有错误之处,恳请指出。

  6. 《转》OpenStack对象存储——Swift

    Swift前身是RackspaceCloudFiles项目,随着Rackspace加入到OpenStack社区,于2010年7月贡献给OpenStack,作为该开源项目的一部分。Swift目前的最新版本是OpenStackEssex1.5.1。Swift特性在OpenStack官网中,列举了Swift的20多个特性,其中最引人关注的是以下几点。在OpenStack中还可以与镜像服务Glance结合,为其存储镜像文件。Auth认证服务目前已从Swift中剥离出来,使用OpenStack的认证服务Keysto

  7. Openstack swift对象存储

    Swift提供的服务与AWSS3基本相同:作为IaaS的存储服务与OpenStackCompute对接,为其存储镜像文档存储存储需要长期保存的数据,例如log存储网站的图片,缩略图等Swift使用RESTfulAPI对外提供服务,目前1.4.6版本所提供的功能:Account的GET、HEADContainer的GET、PUT、HEAD、DELETEObject的GET、PUT、HEAD、DELETE、DELETEAccount、Container、Object的元数据支持大文件、访问控制、权限控制临时对

  8. android – 使用Camera2 API从服务录制视频

    我在一些我无法测试的手机上遇到问题(这是第一个问题)我正在尝试构建一个可以从camera2API录制的服务.但是,当我将应用程序发送到后台时,在某些手机上似乎存在我目前无法解决的问题.我只有日志,无法弄清楚应用程序无法正常工作的主要原因.由于该类太大而无法在此处显示为代码,因此您可以在此处找到它:https://github.com/Astyan42/RecordingService/blob/m

  9. android – 从AOSP master切换到froyo

    我刚用回购检查了主人.现在我想切换到froyo.如果不再下载一堆东西我怎么能这样做?我不想再次下载一堆东西,我只是希望能够在分支之间自由移动,就像在普通的gitclone中一样.解决方法由于您使用repo下载了所有内容,因此您可以轻松地使用repo在分支之间切换而无需再次下载所有内容:repoinit-bfroyo;回购同步这将只下载切换分支所需的文件,就像使用git在分支之间移动一样.

  10. MvvmCross:从android中的MvxListView获取所选项目的最佳方法是什么?

    我正在使用MvvmCrossv3.06,我在android中定义了一个绑定到列表的MvxListView.我可以看到列表,但无法确定获取单击它时所选项目的最佳方法.目前我在活动的OnCreate中执行以下操作,但它不是特别是MVVM,我想知道是否有更好的方法通过绑定?

随机推荐

  1. crontab发送一个月份的电子邮件

    ubuntu14.04邮件服务器:Postfixroot收到来自crontab的十几封电子邮件.这些邮件包含PHP警告.>我已经解决了这些警告的原因.>我已修复每个cronjobs不发送电子邮件(输出发送到>/dev/null2>&1)>我删除了之前的所有电子邮件/var/mail/root/var/spool/mail/root但我仍然每小时收到十几封电子邮件.这些电子邮件来自cronjobs,

  2. 模拟两个ubuntu服务器计算机之间的慢速连接

    我想模拟以下场景:假设我有4台ubuntu服务器机器A,B,C和D.我想在机器A和机器C之间减少20%的网络带宽,在A和B之间减少10%.使用网络模拟/限制工具来做到这一点?

  3. ubuntu-12.04 – 如何在ubuntu 12.04中卸载从源安装的redis?

    我从源代码在Ubuntu12.04上安装了redis-server.但在某些时候它无法完全安装,最后一次makeinstallcmd失败.然后我刚刚通过apt包安装.现在我很困惑哪个安装正在运行哪个conf文件?实际上我想卸载/删除通过源安装的所有内容,只是想安装一个包.转到源代码树并尝试以下命令:如果这不起作用,您可以列出软件自行安装所需的步骤:

  4. ubuntu – “apt-get source”无法找到包但“apt-get install”和“apt-get cache”可以找到它

    我正在尝试下载软件包的源代码,但是当我运行时它无法找到.但是当我运行apt-cache搜索squid3时,它会找到它.它也适用于apt-getinstallsquid3.我使用的是Ubuntu11.04服务器,这是我的/etc/apt/sources.list我已经多次更新了.我尝试了很多不同的debs,并没有发现任何其他地方的错误.这里的问题是你的二进制包(deb)与你的源包(deb-src)不

  5. ubuntu – 有没有办法检测nginx何时完成正常关闭?

    &&touchrestarted),因为即使Nginx没有完成其关闭,touch命令也会立即执行.有没有好办法呢?这样的事情怎么样?因此,pgrep将查找任何Nginx进程,而while循环将让它坐在那里直到它们全部消失.你可以改变一些有用的东西,比如睡1;/etc/init.d/Nginx停止,以便它会休眠一秒钟,然后尝试使用init.d脚本停止Nginx.你也可以在某处放置一个计数器,这样你就可以在需要太长时间时发出轰击信号.

  6. ubuntu – 如何将所有外发电子邮件从postfix重定向到单个地址进行测试

    我正在为基于Web的应用程序设置测试服务器,该应用程序发送一些电子邮件通知.有时候测试是使用真实的客户数据进行的,因此我需要保证服务器在我们测试时无法向真实客户发送电子邮件.我想要的是配置postfix,以便它接收任何外发电子邮件并将其重定向到一个电子邮件地址,而不是传递到真正的目的地.我正在运行ubuntu服务器9.10.先感谢您设置本地用户以接收所有被困邮件:你需要在main.cf中添加:然后

  7. ubuntu – vagrant无法连接到虚拟框

    当我使用基本的Vagrantfile,只配置了两条线:我看到我的虚拟框打开,但是我的流氓日志多次显示此行直到超时:然后,超时后的一段时间,虚拟框框终于要求我登录,但是太久了!所以我用流氓/流氓记录.然后在我的物理机器上,如果我“流氓ssh”.没有事情发生,直到:怎么了?

  8. ubuntu – Nginx – 转发HTTP AUTH – 用户?

    我和Nginx和Jenkins有些麻烦.我尝试使用Nginx作为Jenkins实例的反向代理,使用HTTP基本身份验证.它到目前为止工作,但我不知道如何传递带有AUTH用户名的标头?}尝试将此指令添加到您的位置块

  9. Debian / Ubuntu – 删除后如何恢复/ var / cache / apt结构?

    我在ubuntu服务器上的空间不足,所以我做了这个命令以节省空间但是现在在尝试使用apt时,我会收到以下错误:等等显然我删除了一些目录结构.有没有办法做apt-getrebuild-var-tree或类似的?

  10. 检查ubuntu上安装的rubygems版本?

    如何查看我的ubuntu盒子上安装的rubygems版本?只是一个想法,列出已安装的软件包和grep为ruby或宝石或其他:)dpkg–get-selections

返回
顶部