标签：pac 主题 修改 速度 base 添加 就是 密码 全局
一、概述Kafka最初是由LinkedIn公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统。Apache Kafka也是一个开源消息系统，由Scala写成；Kafka是一个分布式消息队列。Kafka对消息保存是根据Topic进行归类，发送消息者称为Producer，消息接受者称为Consumer，此外kafka集群有多个kafka实例组成，每个实例(server)称为broker。无论是kafka集群，还是consumer都依赖于zookeeper集群保存一些meta（生产者，消费者，broker，topic等等）信息，来保证系统可用性。kafka主要用做存储系统与消息系统。
Kafka主要设计目标如下：
? 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
? 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。
? 支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
? 同时支持离线数据处理和实时数据处理。
? Scale out:支持在线水平扩展
（引自：https://baike.baidu.com/item/Kafka/17930165?fr=aladdin）
消息队列的两种模式
1）点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除）
点对点模型通常是一个基于拉取或者轮询的消息传送模型，这种模型从队列中请求信息，而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理，即使有多个消息监听者也是如此。
2）发布/订阅模式（一对多）
发布订阅模型则是另一个消息传送模型。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即使当前订阅者不可用，处于离线状态。
消息队列的优势
1）解耦
允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。
2）冗余
消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。
3）扩展性
因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。
4）灵活性 & 峰值处理能力
在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。
5）可恢复性
系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。
6）顺序保证
在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。（Kafka保证一个Partition内的消息的有序性）
7）缓冲
有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
8）异步通信
很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。
kafka属于消息队列的发布/订阅模式
架构图
详细架构图
Producer：生产者，允许一个应用程序发布一串流式的数据到一个或者多个Kafka topic
Consumer：消费者，允许一个应用程序订阅一个或多个 topic ，并且对发布给他们的流式数据进行处理
Consumer Group：每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group，Consumer Group是kafka用来实现一个topic消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic可以有多个CG。topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个partion只会把消息发给该CG中的一个consumer。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic；
Broker ：一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic；
Topic ：主题，可以理解为一个队列
Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列，每个partition中的数据使用多个segment文件存储。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。分区是对Topic的分布式存储。
offset：0.9版本以前，partition的offset是由zookeeper管理，0.9版本之后，offset存在brooker本地。
Leader：每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。
Follower：Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。如果Leader失效，则从Follower中选举出一个新的Leader。当Follower与Leader挂掉、卡住或者同步太慢，leader会把这个follower从“in sync replicas”（ISR）列表中删除，重新创建一个Follower。Follower是分区的备份，备份数小于等于Broker数。
Ps：更多详细见kafka中文文档：http://kafka.apachecn.org/intro.html
前提：
①集群要求半数以上节点存活，所以集群数量最好是奇数个，最少是3个
②集群每台服务器必须安装JDK
③各节点时间/时区同步
④集群内部免密
⑤修改主机名并作本地解析
⑥已部署zk集群
集群规划：在node1，node2，node3三个节点部署kafka
JDK版本 1.8
zk版本 3.4.9
kafka版本 2.11
tar -zxvf kafka_2.11-1.0.0.tgz -C /home/bigdata/mkdir kafka_2.11-1.0.0/{kafka-logs,logs}需作修改的配置文件有两个
①config/log4j.properties
将kafka日志的目录更新为/home/bigdata/kafka_2.11-1.0.0/logs
②config/server.properties
修改以下内容
#broker的全局唯一编号，不能重复
broker.id=0
#删除topic功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=10485760
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=10485760
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志存放的路径
log.dirs=/home/bigdata/kafka_2.11-1.0.0/kafka-logs/
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#日志大小（默认1G）
log.segment.bytes=1073741824
#配置连接Zookeeper集群地址
zookeeper.connect=192.168.143.140:2181,192.168.143.141:2181,192.168.143.142:2181/kafka
#连接zk集群的超时时间
zookeeper.connection.timeout.ms=30000
#向生产者和消费者宣告kafka监听的IP和端口
listeners=PLAINTEXT://192.168.143.140:9092
advertised.listeners=PLAINTEXT://192.168.143.140:9092
#设置kafka端口
port=9092
#不允许自建主题
auto.create.topics.enable=false
#启用日志定期删除策略
log.cleanup.policy=delete
#消息正文的最大大小，单位字节。
message.max.bytes=10485760
#副本每次获取的最大数据大小
replica.fetch.max.bytes=20485760
在/etc/profile中添加环境变量
#kafka
export KAFKA_HOME=/home/bigdata/kafka_2.11-1.0.0
export PATH=$KAFKA_HOME/bin:$PATH
重读环境变量
source /etc/profile
JMX_PORT=9191 nohup /home/bigdata/kafka_2.11-1.0.0/bin/kafka-server-start.sh /home/bigdata/kafka_2.11-1.0.0/config/server.properties > /dev/null 2>&1 &Ps：开启JMX轮询
/home/bigdata/kafka_2.11-1.0.0/bin/kafka-server-stop.sh stopkafka-topics.sh ‘--create‘ ‘--zookeeper‘ ‘‘192.168.143.140‘:2181/kafka‘ ‘--replication-factor‘ 1 ‘--partitions‘ 3 ‘--topic‘ test1
选项说明：
--topic 定义topic名
--replication-factor 定义副本数
--partitions 定义分区数[root@node1 bin]# kafka-topics.sh --zookeeper 192.168.143.140:2181/kafka --list
__consumer_offsets
test1需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启
kafka-topics.sh --zookeeper 192.168.143.140:2181/kafka --delete --topic test1[root@node1 kafka_2.11-1.0.0]# kafka-topics.sh --zookeeper 192.168.143.140:2181/kafka --describe --topic test1
Topic:test1 PartitionCount:3 ReplicationFactor:1 Configs:
Topic: test1 Partition: 0 Leader: 2 Replicas: 2 Isr: 2
Topic: test1 Partition: 1 Leader: 0 Replicas: 0 Isr: 0
Topic: test1 Partition: 2 Leader: 1 Replicas: 1 Isr: 1[root@node1 kafka_2.11-1.0.0]# kafka-console-producer.sh --broker-list node1:9092 --topic test1
>hello kafka
>i am eric jia.第一种：offset存在zk集群
[root@node1 kafka_2.11-1.0.0]# kafka-console-consumer.sh --zookeeper node1:2181/kafka --from-beginning --topic test1
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].
hello kafka
i am eric jia.第二种：offset存在本地
[root@node1 kafka_2.11-1.0.0]# kafka-console-consumer.sh --bootstrap-server node1:9092 --from-beginning --topic test1选项说明：
--from-beginning：会把主题中以往所有的数据都读取出来
参考：
《尚硅谷大数据技术之Kafka》
http://kafka.apachecn.org/intro.html
https://www.cnblogs.com/qingyunzong/p/9004509.html
Kafka-manager是目前最主流的kafka管理工具。该工具可以方便查看集群 主题分布情况，同时支持对 多个集群的管理、分区平衡以及创建主题等操作。
特性：
? 管理多个集群
? 轻松检查群集状态（主题，消费者，偏移，代理，副本分发，分区分发）
? 运行首选副本选举
? 使用选项生成分区分配以选择要使用的代理
? 运行分区重新分配（基于生成的分配）
? 使用可选主题配置创建主题（0.8.1.1具有与0.8.2+不同的配置）
? 删除主题（仅支持0.8.2+并记住在代理配??置中设置delete.topic.enable = true）
? 主题列表现在指示标记为删除的主题（仅支持0.8.2+）
? 批量生成多个主题的分区分配，并可选择要使用的代理
? 批量运行重新分配多个主题的分区
? 将分区添加到现有主题
? 更新现有主题的配置
tar -zxvf kafka-manager-1.3.3.18 -C /home/bigdata/kafka-manager.zkhosts="kafka-manager-zookeeper:2181"
修改为：
kafka-manager.zkhosts="192.168.143.140:2181"nohup /home/bigdata/kafka-manager-1.3.3.18/bin/kafka-manager -Dconfig.file=$base_dir/kafka-manager-1.3.3.18/conf/application.conf -Dhttp.port=9998 > /dev/null 2>&1 &选项说明：
-Dconfig.file：指明配置文件
-Dhttp.port：指明服务监听的端口（也可在配置文件中指定）
Cluster Name：集群名
Cluster Zookeeper Hosts：zk集群地址
Kafka Version：kafka版本
Enable JMX Polling：是否开启 JMX 轮训，该部分直接影响部分 kafka broker 和 topic 监控指标指标的获取，生效的前提是 kafka 启动时开启了 JMX_PORT。
JMX Auth Username：JMX认证用户名
JMX Auth Password：JMX认证密码
JMX with SSL ：JMX使用ssl认证
Poll consumer information ：是否开启获取消费信息，直接影响能够在消费者页面和 topic 页面查看消费信息
Filter out inactive consumers ：过滤掉不活跃的消费者
Enable Logkafka ：开始日志
Enable Active OffsetCache：是否开启 offset 缓存，决定 kafka-manager 是否缓存住 topic 的相关偏移量。
Display Broker and Topic Size：展示broker和topic的大小。
Broker监控指标
Topic监控指标
指标说明
Messages in /sec：每秒流入的消息数
Bytes in /sec：每秒流入的字节数
Bytes out /sec：每秒流出的字节数
Bytes rejected /sec：每秒拒绝流入的字节数
Failed fetch request /sec：每秒失败的抓取数据请求数
Failed produce request /sec：每秒失败的生产数据请求数
Ps：更多参见：https://www.jianshu.com/p/6a592d558812
标签：pac 主题 修改 速度 base 添加 就是 密码 全局
原文地址：https://blog.51cto.com/jiayimeng/2487923
