奖励软件下载网站,定制网站,网站精品案例,如何制作一个手机网站源码kafka入门
一种分布式的、基于发布/订阅的消息系统#xff0c;scala编写#xff0c;具备快速、可扩展、可持久化的特点。
基本概念
topic 主题
partition 分区#xff0c;一个topic下可以有多个partition#xff0c;消息是分散到多个partition里存储的#xff0c;part…kafka入门
一种分布式的、基于发布/订阅的消息系统scala编写具备快速、可扩展、可持久化的特点。
基本概念
topic 主题
partition 分区一个topic下可以有多个partition消息是分散到多个partition里存储的partition支持水平扩展。 一个partition内的消息是有序的partition间的消息则是无序的。 每个partition会有若干副本。
broker 一个kafka节点
consumer 消费者从topic里取得消息。 每个consumer维护自己的offset。 consumer数量小于分区数会有一个消费者处理多个分区反之会有空闲的消费者造成浪费。
producer 生产者负责将消息写入topic
特点
基于硬盘的消息保存避免在producer上累积消息或者消息丢失。 同一个消息可以由多个consumer消费。
可扩展性随着数据的增加可扩展为数十台上百台规模的大集群。扩展可以在集群正常运行的时候进行对于整个系统的运作没有影响这也就意味着对于很多台broker 的集群如果一台broker 有故障不影响为client 提供服务.集群如果要同时容忍更多的故障的话, 可以配置更高的replication factors。
高性能上面的这些特性使得Apache Kafka 成为一个能够在高负载的情况下表现出优越性能的发布-订阅消息系统。Producer, consumer 和broker 都能在大数据流的情况下轻松的扩展.
kafka的版本
比如kafka_2.10-0.10.2.0 这里2.10指的是编译kafka的scala版本真正的kafka版本号是后面的0.10.2.0
配置kafka
config/server.properties文件里的配置项说明 broker.id 每个kafka 的broker 都需要有一个整型的唯一标识这个标识通过broker.id 来设置。默认的情况下这个数字是0但是它可以设置成任何值。需要注意的 是需要保证集群中这个id 是唯一的。这个值是可以任意填写的并且可以在必要的时候从broker 集群中删除。比较好的做法是使用主机名相关的标识来做 为id比如你的主机名当中有数字相关的信息如hosts1.example.comhost2.example.com那么这个数字就可以用来作为broker.id 的值。
port 默认启动kafka 时监听的是TCP 的9092 端口端口号可以被任意修改。如果端口号设置为小于1024那么kafka 需要以root 身份启动。但是并不推荐以root 身份启动。
zookeeper.connect 这个参数指定了Zookeeper 所在的地址它存储了broker 的元信息。默认是运行在本机的2181 端口上因此这个值被设置成 localhost:2181。这个值可以通过分号设置多个值每个值的格式都是hostname:port/path
log.dirs 这个参数用于配置Kafka 保存数据的位置Kafka 中所有的消息都会存在这个目录下。可以通过逗号来指定多个目录kafka 会根据最少被使用的原则选择目录分配新的partition。注意kafka 在分配partition 的时候选择的规则不是按照磁盘的空间大小来定的而是分配的parition 的个数大小。
num.recovery.thread.per.data.dir kafka 可以配置一个线程池线程池的使用场景如下
当正常启动的时候开启每个parition 的文档块当失败后重启时检查parition 的文档块当关闭kafka 的时候清除关闭文档块 默认每个目录只有一个线程。最好是设置多个线程数这样在服务器启动或者关闭的时候都可以并行的进行操作。尤其是当非正常停机后重启时如果有大量的分区数那么启动broker 将会花费大量的时间。 【注意】这个参数是针对每个目录的。比如num.recovery.threads.per.data.dir 设置为8如果有3个log.dirs 路径那么一共会有24 个线程。
num.partitions 这个参数用于配置新创建的topic 有多少个分区默认是1 个。注意partition 的个数只可以被增加不能被减少。这就意味着如果想要减少主题的分区数那 么就需要重新创建topic。 在第一章中介绍过kafka 通过分区来对topic 进行扩展因此需要使用分区的个数来做负载均衡如果新增了broker那么就会引发重新负载分配。这并不意味着所有的主题的分区数都需要大于broker 的数量因为kafka 是支持多个主题的其他的主题会使用其余的broker。需要注意的是如果消息的吞吐量很高那么可以通过设置一个比较大的分区数来分摊压力。
log.retention.ms 这个参数用于配置kafka 中消息保存的时间也可以使用log.retention.hours默认这个参数是168 个小时即一周。另外还支持log.retention.minutes 和log.retention.ms。这三个参数都会控制删除过期数据的时间推荐还是使用log.retention.ms。如果多个同时设置那么会选择最小的那个。
log.retention.bytes 这个参数也是用来配置消息过期的它会应用到每个分区比如你有一个主题有8 个分区并且设置了log.retention.bytes 为1G那么这个主题总共可以保留8G 的数据。注意所有的过期配置都会应用到patition 粒度而不是主题粒度。这也意味着如果增加了主题的分区数那么主题所能保留的数据也就随之增加了。 如果设置了log.retention.bytes 和log.retention.ms或者其他过期时间的配置只要满足其中一个条件消息就会被删除。
log.segment.bytes 这个参数用来控制log 段文件的大小而不是消息的大小。在kafka 中所有的消息都会进入broker然后以追加的方式追加到分区当前最新的segment 段文件中。一旦这个段文件到达log.segment.bytes 设置的大小比如默认的1G这个段文件就会被关闭然后创建一个新的。一旦这个文件被关闭就可以理解成这个文件已经过期了。这个参数设置的越小那么关闭文件创建文件的操作就会越频繁这样也会造成大量的磁盘读写的开销。 通过生产者发送过来的消息的情况可以判断这个值的大小。比如主题每天接收100M 的消息并且log.segment.bytes 为默认设置那么10 天后这个段文件才会被填满。由于段文件在没有关闭的时候是不能删除的log.retention.ms 又是默认的设置那么这个消息将会在17 天后才过期删除。因为10 天后段文件才关闭。再过7 天这个文件才算真正过期才能被清除。
message.max.bytes 这个参数用于限制生产者消息的大小默认是1000000也就是1M。生产者在发送消息给broker 的时候如果出错会尝试重发但是如果是因为大小的原因那生产者是不会重发的。另外broker上的消息可以进行压缩这个参数是指压缩后的大小这样能多存储很多消息。 需要注意的是允许发送更大的消息会对性能有很大影响。更大的消息就意味着broker 在处理网络连接的时候需要更长的时间它也会增加磁盘的写操作压力影响IO 吞吐量。
启动kafka
进入kafka/bin目录。
启动zk
nohup ./zookeeper-server-start.sh ../config/zookeeper.properties 测试zk是否启动
telnet localhost 2181 输入srvr应该会返回
[2017-12-01 15:59:18,829] INFO Processing srvr command from /0:0:0:0:0:0:0:1:40194 (org.apache.zookeeper.server.NIOServerCnxn)
Zookeeper version: 3.4.6-1569965, built on 02/20/2014 09:09 GMT
Latency min/avg/max: 0/0/0
Received: 1
Sent: 0
Connections: 1
Outstanding: 0
Zxid: 0xdf
Mode: standalone
Node count: 127
[2017-12-01 15:59:18,833] INFO Closed socket connection for client /0:0:0:0:0:0:0:1:40194 (no session established for client) (org.apache.zookeeper.server.NIOServerCnxn)
Connection closed by foreign host.不建议zookeeper运行在多于7个的节点上因为集群性能会因一致性的特征而降低。
启动kafka
nohup ./kafka-server-start.sh -daemon ../config/server.properties以守护进程方式执行。
topic管理命令
创建具有指定数量分区或复制因子的topic ./kafka-topics.sh --create --zookeeper xx.xx.xx.xx:2181 --topic test1 --replication-factor 1 --partitions 6
查看topic的元信息 ./kafka-topics.sh --zookeeper localhost:2181 --describe --topic test1
查看所有topic ./kafka-topics.sh --zookeeper localhost:2181 --list
修改已有topic的分区
./kafka-topics.sh --alter --zookeeper localhost:2181 --topic test1 --partitions 4./kafka-topics.sh --zookeeper xx.xx.xx.xx:2181 --alter --topic test1 --config flush.ms5000./kafka-topics.sh --zookeeper xx.xx.xx.xx:2181 --alter --topic test1 --config flush.messages20000删除topic
./kafka-topics.sh --delete --zookeeper xx.xx.xx.xx:2181 --topic test1但删除topic并不是真的删除只是打个标记要真正删除需要同时修改server.properties
delete.topic.enabletrue并重启kafka才能生效。
向topic发布消息
./kafka-console-producer.sh --broker-list localhost:9092 --topic test1 输入 msg1 msg2 按^D停止发送。
若kafka的server.properties配置了host.name则localhost必须改成host.name的值例如 ./kafka-console-producer.sh --broker-list xx.xx.xx.xx:9092 --topic test1
查看topic里的消息
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1 --from-beginning 输出 msg1 msg2 按^C停止接收。
