1. 什么是消息系统?
(1)消息系统是负责将数据从一个程序传输到另一个应用程序,因此应用程序可以专注于数据,不用担心如何共享它 分布式消息传递给基于可靠消息队列的概念. 消息在客户端应用程序和消息传递系统之间异步排队(2)有两种类型的消息模式可用 a.点对点的方式 b.发布-订阅消息系统(pub-sub)
2.消息系统问题?
(1)消息系统怎么知道某一个主题被哪些订阅者订阅? 需要进行维护,消息需要推送给订阅者(2)采用订阅方式,会存在订阅者越多,消息系统维护的元数据很多,name发送的推送的时候就曾大 增加消息系统的处理单元(加主机/CPU)(3)如果在推送的时候,订阅者挂掉? 那么之后发布的消息该如何处理? 通过消费者拉的方式去获取消息
3.消息系统的主题介绍
RSS :中指是频道 消息系统当中:主题(消息分类) 主题设计的原则? 如果没有主题的话,订阅者接受到的消息将会是消息系统中的所有信息 有主题的话,只需要关注自己想要的内容.
4.系统的演变过程?
一个普通的j2EE项目(所有的东西全在一个系统里面)====>集群扩展(1变N) ---->分布式(某一些服务器只处理特定的业务)
业务与业务之间就回存在数据传递!
5.业务员场景
1.应用解耦
(订单信息的队列)
2.流量肖锋
可控制活动人数
可以缓解短时间内的高流量压垮应用
3.日志处理
解决大量日志传输问题,负责日志的采集,定时写入kafka队列,日志数据的接受存储和换发
4.消息通讯,聊天室,实时对话...
6.什么是kafka?
(1)使用Scala语言开发,kafka是一个分布式,分区的,多副本的,多订阅者的日志系统,可用于web/ngnix日志,搜索日志,监控日志,访问日志.(2)目前支持多种客户端语言: java python c++ php …(3)apache kafka 是一个分布式 发布-订阅消息系统和一个强大的队列,可以处理大量的数据,并且使能够将消息从一个断电传递到另一个断电. kafka适合离线和在线消息消费.kafka消息保留在磁盘上,兵仔集群内复制以防止数据丢失,kafka构建在zookeeper同步服务至上 它与Apache storm 和 spark非常好的集成,用于实施流式数据分析.
7.kafka的好处?
可靠性 – 分布式 ,分区 复制 和容错 可扩展性 – kafka消息传递系统轻松缩放 耐用性 --分布式提交日志,尽快保留在磁盘上 性能—对发布和订阅消息都具有高吞吐量,即使存储量TB的消息,也可以保持稳定的性能 kafka非常的快
8.kafka的应用?
用于操作实时监控数据,分布式应用程序信息的统计 日志聚合解决方案,收集日志 流处理-流行的框架从主体中读取数据,兵对其进行处理,将处理后的数据写入新主题.
9.kafka术语?
broker: kafka集群包含一个或多个服务实例,建议一个broker包含一个实例 topic: 每发布到kafka集群的消息都有一个类别—主题 partition: 物理上的概念,每个topic包含一个或多个partition(分区) producer: 负责发布消息到kafka broker consumer: 消息消费者, 向kafka读取消息的客户端 consumer group : 每个consumer属于一个特定的consumer group
10.kafka的安装应用
(1)kafka中需要用到zk ,先启动zk,如果在伪分布式下,kafka已经集成了zk (后天自启动) nohup zookeeper-server-start.sh /kafka-2.11-2.0.0/config/zookeeper.properties &
(2)启动broker nohup kafka-server-start.sh /kafka-2.33-2.0.0/config/server.properties &
(3)测试 模拟一个生产者,向另一个消费者发送数据 3.1创建一个主题
kafka-topics.sh --create --zookeeper localhost:2181 --topic mytopic --partitions 1 --replication-factor 1
--topic : 主题的名称
--partitions: 主题下有几个分区
--replication-factor: 副本数
3.2创建一个生产者
kafka-console-producer.sh --topic mytopic --broker-list localhost:9092
3.3消费者的创建
kafka-console-consumer.sh --topic mytopic --zookeeper localhost:2181
11 kafka主题(topic)
消息在进入kafka集群的时候,事宜主题进行数据归纳[一条消息必须属于某个主题]
在kafka集群中,可以有无数的主题
从生产者角度: 它所操作的单元,一般情况下以主题为单位
从消费者角度: 它锁操作的单元,一般情况下是以主题为单位
生产者 消费者可以以主题更细的单位操作-->分区
从kafka角度: 没有限制生产者也没有限制消费者
生产者与消费这咋处理数据的时候,必须要知道要操作的主题是哪一个.
12.如何创建主题?
(1)一个broker下是否可以创建多个分区? 可以,broker数与分区属无关 kafka-topics-sh --create --topic mytopic2 --zookeeper localhost:2181 --partitions 3–replication-factor 1(2)一个broker下是否可以创建多个副本? 不可以,创建主题时,副本应为应该小鱼等于可用的broker数 (3) 配置属性 --config eg: --config max.message.bytes=6400 --config flush.messages=1 最大消息字节, flush一次
13.查看有哪些主题?
(1)主题详细信息 kafka-topics.sh --describe --zookeeper localhost:2181 在kafka中每一个分区会有一个编号:从0开始 Leader : broker.id=0 应为在kafka中有多个副本的话,就回存在leader和follower的关系 表示当前这个副本为leader所在的broker是哪一个 server.properties文件中,如果有多个broker ,broker.id 不能重复 Replicas: 所有的副本列表 0,1,2 Isr:可用列表数 0,1(2)查看主题名称: kafka-topics.sh --list --zookeeper localhost:2181(3)某一个主题查询 kafka-topics.sh --describe --zookeeper localhost:2181 --topic mytopic(4) 某一个是佛UCUN在 kafka-topics.sh --describe --zookeeper localhost:2181 | grep mytopic(5)修改主题 kafka-topics.sh --zookeeper localhost:2181 --alter --topic mytopic --config flush.messages=1(6)删除配置参数 kafka-topics.sh --zookeeper localhost:2181 --alter --topic mytopic --delete-config flush.messages(7)删除 kafka-topics.sh --zookeeper localhost:2181 --delete --topic mytopic 当执行命令之后,topic不是物理删除,而是一个标记删除的操作 mytopic - marked for deletion 标记删除后的主题还可以继续生产数据注意? 当服务器重启,就回删除已经标记的topic ,这个和版本有关系 真正的物理删除,需要在server.properties文件中加入参数 delete .topic.enable=true (一般放在zookeeper代码注释之上)
14 分区 partitions ?
分区是针对某一个主题下数据的拆分 让消费者在拿数据的时候,速度会更快 一个broker服务下可以有多个分区 分区中数据是有序的,且不可修改,但不同分区中的数据可以是无序的
生产方的生产数据顺序与拿到的顺序不一致? 可能出现先退货,然后再发货 让同一类的数据进入到同一个分区里!同一个分区下的数据是有序的 如何保证一个主题下的数据,一定有序的? 生产什么样的顺序,那么消费的时候也是什么样的顺序? 让主题只有一个分区
新进来的消息会追加到某一个分区的尾部
(2) 分区与消费组之间的关系?
消费组: 由一个或者多个消费者组成,同一个组中的消费者对于同一条消息只消费一次
当某一个主题中的分区数对于消费组来说应该小于等于该主题下的分区数
eg : 某一个主题有4个分区,那么消费组中的消费者应该小于4,而且最好分区与分区数成整倍
同一个分区下的数据,在同一时刻,不能被同一个消费组的不同消费者消费
15.offset数据相应的编号
偏移量: 记录当前有多少记录数,而且让消费者可以知道自己消费到什么地方 可以让消费者自定义选择消费某一条消息对于同一条消息,消费者可以消费多次 确定一条唯一的数据: 消息可以通过 offset partition topic 跟踪记录
16.分区下的副本因子
创建副本因子的时候,副本因子应该小于等于可用的broker数
副本应为操作的单位是分区为单位
当有多个副本数时,kafka并不是将多个副本同时对外读写提供服务
作用?
让kafka读取和写入数据时的高可靠
副本因子数时包含本身的 同一个副本因子不能放在同一个broker中
在有多个副本的情况下,kafka会为同一个分区下的分区,设定角色关系,一个leader和多个follower
leader负责和外面(生产者/消费者)的角色,进行读写处理
生产 ----> 副本关系?
生产方: 可以采用同步或者异步的方式
同步: 发送一批数据给kafka之后,等待kafka返回结果
异步: 发送一批数据给kafka之后,只提供一个回调函数
生产者获取返回结果是什么时候? (确认机制)
1)需要所有的副本确认才表示该条消息写入成功,返回给生产者.
可能出现,在返回给生产者时,未及时返回, 出现重复发送
2)不许要确认.数据丢失
再次发送
3)只需要leader确认
4)部分确认
如果某一个分区有三个副本因子就算其中一个挂掉,那么只会在剩下的2个钟选择一个leader.
不会再其他的broker钟,另启动一个副本,[因为另外启动一个副本的话,存在数据传递,只要主机之间有数据传递,就会长时间暂用网络IO]
但是kafka是一个高吞吐量的消息系统,这种情况不允许发送,所以不会再另外一个broker中启动.
如果所有的副本都挂了? 生产者如果生产的数据时指定分区的话,将写入不成功.
将重新启动,最先重新启动好的作为leader
ISR(可用副本) : 选择副本做为leader
ack机制: 确认机制
在用户获取数据的时候,可以采取kafka的两种API
高级API和低级API
高级API:隐藏consumer和broker的相关细节,相关信息保存在zookeeper中 [推荐]
让用户使用时很方便,大部分操作都是封装好的. 比如: 当前消费到哪个位置下. 但是不够灵活
低级API: 没有进行包装,所有操作由用户决定,需要维护leader broker
如: 自己保存某一个分区下的记录,你当前消费的位置
leader broker是之前的一种说法,现在其它上不用
kafka集群中 ---> 包含了很多的broker,但是这么多的broker中也会有一个老大存在
是在kafka节点中的一个临时节点,去创建相应的数据. controller broker (最先创建的broker)
controller broker作用: 管理所有的broker
controller_epoch 选举
brokers : broker 服务
consumers: 消费
config : kafka 配置信息
admin -->delete-topic 标记删除的主题
17.kafka shell
#bin/kafka-topics.sh
create,delete,describe,or change a topic
--alter 更改或添加
--create 创建
--delete 删除(标记删除,若要物理删除需要在server.properties中设置)
--describe 主题信息
--list 主题列表
--partitions 分区
--replication-factor 副本
创建: topic 名称 zookeeper地址 partitions 个数 replication-factor 副本数 是必须的参数
删除: topic zookeeper 是必要参数,在服务端配置delete.topic.enable=true
修改: topic zookeeper 必须参数, 副本因子一般只能添加
list: topic zookeeper 必须参数
describe: zookeeper必须参数
18. 集群搭建
1.zookeeper集群搭建
2.kafka集群搭建
jps
注意:
每个broker的broker.id 必须不一样
log.dirs 修改地址,之前的在/tmp
zookeeper.connect=ziji de zk集群地址 (多个,分割)
启动kafka,每一台都要启动
19.kafka生产者代码:
1. 映入包
一般安装包的版本,maven引用的版本一致
创建主题:
bin/kafka-topics.sh --create --zookeeper master,salve01,salve02 --topic mytopic --partitions 3 --replication-factor 3
查看详情
bin/kafka-topics.sh --describe --zookeeper master,slave01,salve02 --topic mytopic
每个broker 上对于分区上都有一个副本leader
// ProduceRecord:
/**
2个参数: 主题,内容
3个参数: 主题,key,内容发送
4个参数: 主题,分区,key,内容发送
5个参数: 主题,分区,时间戳,key,内容发送
主题: 自定义
分区: 0 1 2 ...
当没有key的时候,消费者拿到的数据时什么形式?
采用内容分区规则,数据均衡的原则
采用自己的key,key值决定了进入哪个分区
指定了key的时候,key 相同,消息会进入同一个分区
如果按照key值进行处理(采用的是kafka内部的)
假设一个主题下有10个分区:按省份处理
四川 : 0
广州 : 1
若省份大于了分区数...怎么办?
采用自定义分区来解决: 得到key ,分区数
*/
20.kafka 消费者代码
// 手动提交,偏移量
props.put("enable.auto.commit", "false");
// 设置从哪些主题下获取数据
//consumer.subscribe(Arrays.asList("mytopic"));
// 通过分区获取
// (主题 分区)
TopicPartition tp = new TopicPartition("mytopic","0");
// 主题下分区获取数据
consumer.assign(Arrays.asList(tp));
// 同步方式提交
consumer.commitAsync();
21.提交过程
已经消费的数据对于kafka来说,会将该消费组里面的offset值进行修改, 什么时候去修改? 自动提交: 提交过程是通过kafka可以将offset进行移动到下一个信息
比如数据拿到之后,存入到hbase中
如果hbase在这个时候,连接不上,如果在处理的时候已经进行了提交,那么kafka上的offset值已经修改了, 但是hbase中没有数据
这样会导致数据丢失,怎么处理?
手动提交: 将数据处理完之后,再来进行offset进行修改提交,
默认情况下是自动提交,修改手动提交
propes.put("enable.auto.commit","false");
// 同步方式提交
consumer.commitAsync();
若数据写入完成, 在提交修改offset请求是,中断产生了, 那么修改该次的offset是失败的
那么下一次读取同一个分区中的数据时,会从已经处理的offset值再次进行处理一次,造成数据重复
怎么解决?
22. producer自定义分区
/**
partitioner.class=com.zhuihui.Mypartition 分区接口指定为自己的, 在properties文件中,或者在生产者用的时候
*/
class MyPartition implements Partitioner
{
/**
topic : 主题
key :
keyBytes : key的字节数组
value
valueBytes :
cluster 集群信息
模拟:
135开头 0
138 1
130 2
*/
public int partition(String topic,Object key,byte[] keyBytes, Object value,byte[] valueBytes, Cluster cluster){
// 获取主题下的分区数\
// 如果分区数不为3,那么可以进入到0
Integer count = cluster.partitionCountForTopic(topic);
String keyString = key.toString();
if (keyString !=null && count==3 )
{
if (keyString.startWith("135")
{
return 0;
}
if (keyString.startWith("138")
{
return 1;
}
if (keyString.startWith("130")
{
return 2;
}
}
}
}
