建设部网站1667号下载,深圳网站建设定制开发 .超凡科技,临安规划建设局网站,有那些网站生产者学习
1.1 生产者消息发送流程
在消息发送的过程中#xff0c;涉及到了两个线程——main线程和Sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulator#xff0c;Sender线程不断从RecordAccumulator中拉取消息发送到K…生产者学习
1.1 生产者消息发送流程
在消息发送的过程中涉及到了两个线程——main线程和Sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulatorSender线程不断从RecordAccumulator中拉取消息发送到Kafka Broker。 生产者如何发送的 现在Main线程中将数据进行处理处理成IO型数据然后调用sender进行发送 Main: 1.读取生产者配置 2.产生数据 3.过滤数据校验什么的 4.序列化 5.放入缓冲区 RecordAccumulator 6.发送Sender
细节 考虑的问题 1.生产者配置的读取和修改 2.数据的过滤与分区 3.缓冲区是如何设置的大小 4.发送发送失败怎么样请求区的大小 这里注意一下可以在缓冲区对数据进行压缩这样就提高缓冲区的容量和发送的数据量提高吞吐量
1.2 同步发送与异步发送
1.什么是同步和异步
同步就是串行一条龙 异步 一起运行 举例 餐馆点餐 同步 需要等服务员过来让服务员记录 异步 点餐APP直接点餐交给队列让他自己运行
2.发送的同步异步
同步需要得到返回值 异步发送过去不管了
3. 分区好处
啥是分区 将一个数据块分成多个数据块 将数据分布式处理了 存储 可以分在多个机器上 也可以整多个副本。便于存储同时提高健壮性 IO多个数据块可以同时进行发送接收消费。生产者可以以分区为单位发送数据消费者可以以分区为单位进行消费
4. 默认分区器
前提条件 1.分区 2.key值 规则
1存在按1分区1不存在按2.key值对分区数取余得到的值分区1.2都不存在 随机选个分区等这个批次发送完了再换
3 就是粘性分区 那么粘性分区的缺点是什么 因为缓冲区溢出的条件是大小和时间双重判断如果大小不够但是时间够了还是会发走这样最后导致分区上产生数据倾斜 如何解决的 3.3.1 Kafka去掉粘性分区的时间控制批次只由大小判断
1.3.自定义分区器
1.思路
1.实现接口Parititoner,重写相关方法2.修改配置 将partitioner设置为默认配置
2.1 自定义分区器代码
public class MyPartitioner implements Partitioner {// 自定义分区器 实现partitioner接口// 1.分区方法Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 获取消息String data value.toString();// 创建partition 作为最后的分区标识int partitions;// 分区逻辑// 根据含有的字符串进行判断 判断进入哪个分区if (data.contains(atguigu)){partitions 0;} else if (data.contains(shangguigu)){partitions 1;} else {partitions 2;}return partitions;}Overridepublic void close() {}Overridepublic void configure(MapString, ? configs) {}
}2.2 主类
package com.atguigu.producer;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;public class ProducerClientAsync {public static void main(String[] args) {// 0 配置对象Properties properties new Properties();// --指定kafka的Broker地址properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, hadoop102:9092,hadoop103:9092);// -- 1.指定序列化器 序列化器的全限定类名properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());//.setProperty(ProducerConfig.LINGER_MS_CONFIG,0);// -- 2.设置分区器properties.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class.getName());// -- 3.获取客户端连接对象KafkaProducerString,String kafkaProducer new KafkaProducerString,String(properties);// key是主题 v是发送内容 这里注意一下// -- 4.发送数据String[] str {atguigu,111,atguigu,shangguigu,222};for (int i 0; i str.length; i) {System.out.println(str[i]);try {kafkaProducer.send(new ProducerRecord(first, str[i]), new Callback() {Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception null){System.out.println(主题 metadata.topic() - 分区 metadata.partition());}else {// 出现异常打印exception.printStackTrace();}}}).get();} catch (InterruptedException e) {throw new RuntimeException(e);} catch (ExecutionException e) {throw new RuntimeException(e);}}kafkaProducer.close();}
}3.面试细节
1.如何提高生产者的吞吐量
批次大小调到16将等待时间改成50-100ms 默认是0压缩数据量这样每次发送的数据就多了加大缓冲区大小进来的数据变多发送也能提上去
2.生产者如何保证数据可靠性的
主要通过ack机制
1.什么是ACK机制
根据ack值来决定Kafka集群服务端的存储应答
ack0 最低 生产者只管发送不用接收ack1 中等 生产者发送完需要等待Leader保存后回应ack-1 最高 生产者发送完需要等待所有副本保存后回应
2.分析ACK机制
性能与安全是成反比的 所以-1虽然最安全但是效率最低
3.如果将ACK调到-1会出现什么问题
有可能出现数据重复发送与接收 比如在同步的瞬间Leader死掉但是其他副本已经落盘这时候就是问题了。 因为Leader死掉了所以会直接更换Leader选出一个副本作为Leader注意这时显示没有收到内容所以send重新发送这时候,每个副本上收到的就是2份该数据了。
4.应用场景
acks0 几乎不用 acks1 传输普通日志允许丢失 acks-1 传输高可靠性数据一般与钱有关
5.ACK-1一定可靠么
不一定 如果分区副本数设置为1 或者ISR里应答的最小副本数设置为1默认也是1这时候ack1效果相同了。 也就是说应答一个就能走,就没意义了 所以需要完全可靠就需要配置一下 ACK-1 分区副本大于等于2 ISR应答最小副本数量大于等于2
3. 数据去重
1.概念
至少一次一次或者多次 完全可靠 最多一次直接不管回复只管发送 ack0
至少保证数据不丢失但是无法保证数据不重复 最多 无法保证数据不丢失
1.如何解决数据的重复发送与接收的问题,同时保证数据的不丢失
注意这里解决的是sender和服务端的重复发送与接收而不是生产者本身发送多个重复消息的问题这个要搞清楚。 一般重复问题都是通过标识来判别从而去重的 Kafka 0.11 引入 幂等性和事务 精确一次 幂等性 至少一次ack-1 分区副本2 ISR最小副本2
4.幂等性
1.概念
啥是幂等性标识一个消息的唯一标识 pid,partition,Seqnumber Pid 是会话ID每次重新生成会话就会重新生成PID partition是分区 标识 消息是哪个分区的 Seqnumber是单调递增的标识注意这是每个分区独享的 这三个在一起才是唯一标识。
2.如何使用幂等性
开启参数enable.idempotence 默认为truefalse关闭。 开启开关就行 文章转载自： http://www.morning.qgjxy.cn.gov.cn.qgjxy.cn http://www.morning.brld.cn.gov.cn.brld.cn http://www.morning.thjqk.cn.gov.cn.thjqk.cn http://www.morning.xflwq.cn.gov.cn.xflwq.cn http://www.morning.nytqy.cn.gov.cn.nytqy.cn http://www.morning.rqlzz.cn.gov.cn.rqlzz.cn http://www.morning.mqss.cn.gov.cn.mqss.cn http://www.morning.fxwkl.cn.gov.cn.fxwkl.cn http://www.morning.deanzhu.com.gov.cn.deanzhu.com http://www.morning.knscf.cn.gov.cn.knscf.cn http://www.morning.snrhg.cn.gov.cn.snrhg.cn http://www.morning.kdrjd.cn.gov.cn.kdrjd.cn http://www.morning.spsqr.cn.gov.cn.spsqr.cn http://www.morning.rjrh.cn.gov.cn.rjrh.cn http://www.morning.mrpqg.cn.gov.cn.mrpqg.cn http://www.morning.mlwpr.cn.gov.cn.mlwpr.cn http://www.morning.rwhlf.cn.gov.cn.rwhlf.cn http://www.morning.wjdgx.cn.gov.cn.wjdgx.cn http://www.morning.gwkwt.cn.gov.cn.gwkwt.cn http://www.morning.flfdm.cn.gov.cn.flfdm.cn http://www.morning.trjdr.cn.gov.cn.trjdr.cn http://www.morning.xgmf.cn.gov.cn.xgmf.cn http://www.morning.ypfw.cn.gov.cn.ypfw.cn http://www.morning.bpmnl.cn.gov.cn.bpmnl.cn http://www.morning.pwxkn.cn.gov.cn.pwxkn.cn http://www.morning.hrnrx.cn.gov.cn.hrnrx.cn http://www.morning.xmpbh.cn.gov.cn.xmpbh.cn http://www.morning.pjtnk.cn.gov.cn.pjtnk.cn http://www.morning.zcsch.cn.gov.cn.zcsch.cn http://www.morning.kqgsn.cn.gov.cn.kqgsn.cn http://www.morning.kjmws.cn.gov.cn.kjmws.cn http://www.morning.bpkqd.cn.gov.cn.bpkqd.cn http://www.morning.kjxgc.cn.gov.cn.kjxgc.cn http://www.morning.gcqdp.cn.gov.cn.gcqdp.cn http://www.morning.qgcfb.cn.gov.cn.qgcfb.cn http://www.morning.qrcxh.cn.gov.cn.qrcxh.cn http://www.morning.xqltq.cn.gov.cn.xqltq.cn http://www.morning.qkdcb.cn.gov.cn.qkdcb.cn http://www.morning.jntdf.cn.gov.cn.jntdf.cn http://www.morning.frnjm.cn.gov.cn.frnjm.cn http://www.morning.kbbmj.cn.gov.cn.kbbmj.cn http://www.morning.cctgww.cn.gov.cn.cctgww.cn http://www.morning.dtpqw.cn.gov.cn.dtpqw.cn http://www.morning.dbphz.cn.gov.cn.dbphz.cn http://www.morning.clbgy.cn.gov.cn.clbgy.cn http://www.morning.nsfxt.cn.gov.cn.nsfxt.cn http://www.morning.rxlck.cn.gov.cn.rxlck.cn http://www.morning.hqwxm.cn.gov.cn.hqwxm.cn http://www.morning.mmtbn.cn.gov.cn.mmtbn.cn http://www.morning.bdkhl.cn.gov.cn.bdkhl.cn http://www.morning.nqlcj.cn.gov.cn.nqlcj.cn http://www.morning.lzbut.cn.gov.cn.lzbut.cn http://www.morning.pcshb.cn.gov.cn.pcshb.cn http://www.morning.pfgln.cn.gov.cn.pfgln.cn http://www.morning.jcjgh.cn.gov.cn.jcjgh.cn http://www.morning.mrlls.cn.gov.cn.mrlls.cn http://www.morning.brkrt.cn.gov.cn.brkrt.cn http://www.morning.bfybb.cn.gov.cn.bfybb.cn http://www.morning.rwtlj.cn.gov.cn.rwtlj.cn http://www.morning.jbxmb.cn.gov.cn.jbxmb.cn http://www.morning.jbpdk.cn.gov.cn.jbpdk.cn http://www.morning.sphft.cn.gov.cn.sphft.cn http://www.morning.rgxll.cn.gov.cn.rgxll.cn http://www.morning.lhytw.cn.gov.cn.lhytw.cn http://www.morning.dzrcj.cn.gov.cn.dzrcj.cn http://www.morning.srbbh.cn.gov.cn.srbbh.cn http://www.morning.tpwrm.cn.gov.cn.tpwrm.cn http://www.morning.ptqds.cn.gov.cn.ptqds.cn http://www.morning.lyjwb.cn.gov.cn.lyjwb.cn http://www.morning.jjxnp.cn.gov.cn.jjxnp.cn http://www.morning.hphfy.cn.gov.cn.hphfy.cn http://www.morning.snbq.cn.gov.cn.snbq.cn http://www.morning.pzcqz.cn.gov.cn.pzcqz.cn http://www.morning.rykw.cn.gov.cn.rykw.cn http://www.morning.crsqs.cn.gov.cn.crsqs.cn http://www.morning.lxmmx.cn.gov.cn.lxmmx.cn http://www.morning.jfxth.cn.gov.cn.jfxth.cn http://www.morning.wrkhf.cn.gov.cn.wrkhf.cn http://www.morning.kcxtz.cn.gov.cn.kcxtz.cn http://www.morning.jtfsd.cn.gov.cn.jtfsd.cn
