wordpress网站转app,淘宝美工需要学什么软件,网站 伪静态,安徽网站建设SEO优化制作设计公司#x1f44f;作者简介#xff1a;大家好#xff0c;我是爱吃芝士的土豆倪#xff0c;24届校招生Java选手#xff0c;很高兴认识大家#x1f4d5;系列专栏#xff1a;Spring源码、JUC源码、Kafka原理#x1f525;如果感觉博主的文章还不错的话#xff0c;请#x1f44… 作者简介大家好我是爱吃芝士的土豆倪24届校招生Java选手很高兴认识大家系列专栏Spring源码、JUC源码、Kafka原理如果感觉博主的文章还不错的话请三连支持一下博主哦博主正在努力完成2023计划中源码溯源一探究竟联系方式nhs19990716加我进群大家一起学习一起进步一起对抗互联网寒冬 文章目录 练习生产者Ack 应答机制消费者重要参数 API 开发topic 管理列出主题查看主题信息创建主题删除主题其他管理 练习消费者消费者 练习
需求
写一个生产者不断地去生成“用户行为事件” 并写入kafka
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789}
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789}
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789}
…
写一个消费者不断地从kafka种取消费如上 “ 用户行为事件” 数据并做统计计算每五分钟输出一次截至到当时的数据中出现过的用户总数。
生产者
public class KafkaTest{public static void main(String[] args){MyDataGen myDataGen new MyDataGen();myDataGen.genData();}
}/*
业务数据生成器
*/
class MyDataGen{ProducerString, String produce;public MyDataGen(){Properties props new Properties();//设置 kafka 集群的地址props.put(bootstrap.servers, doitedu01:9092,doitedu02:9092,doitedu03:9092);//序列化器props.put(key.serializer,
org.apache.kafka.common.serialization.StringSerializer);props.put(value.serializer,
org.apache.kafka.common.serialization.StringSerializer);//ack 模式取值有 01-1all all 是最慢但最安全的
props.put(acks, all)producer new KafkaProducer(props);}public void genData(){UserEvent userEvent new UserEvent();while(true){// 造一条随机用户行为事件数据对象userEvent.setGuid(RandomUtils.nextInt(1,10000));userEvent.setEventId(RandomStringUtils.randomAlphabetic(5,8));userEvent.setTimeStamp(System.currentTimeMills());// 转成json串String json JSON.toJSONString(userEvent);// 讲业务数据封装成ProducerRecord对象ProducerRecordString,String record new ProducerRecordString,String(test,json);// 用producer写入kafkaproducer.send(record);Thread.sleep(RandomUtils.nextInt(500,2000));}}
}NoArgsConStructor
AllArgConstructor
Getter
Setter
class UserEvent{private long guid;private String eventId;private long timeStamp;
}Ack 应答机制
Ack 应答机制参数配置 0 生产者发出消息后不等待服务端的确认1 生产者发出消息后要求服务端的分区 leader 确保数据存储成功后发送一个确认信息-1 生产者发出消息后要求服务端的分区的 ISR 副本全部同步成功后发送一个确认信息
生产者的 ackall也不能完全保证数据发送的 100%可靠性
为什么因为如果服务端目标 partition 的同步副本只有 leader 自己了此时它收到数据就会给生产者反馈成功
可以修改服务端的一个参数分区最小 ISR 数[min.insync.replicas]2来避免此问题
消费者重要参数
fetch.min.bytes1B 一次拉取的最小字节数
fetch.max.bytes50M 一次拉取的最大数据量
fetch.max.wait.ms500ms 拉取时的最大等待时长
max.partition.fetch.bytes 1MB 每个分区一次拉取的最大数据量
max.poll.records500 一次拉取的最大条数
connections.max.idle.ms540000ms 网络连接的最大闲置时长
request.timeout.ms30000ms 一次请求等待响应的最大超时时间 consumer 等待请求响应的最长时间
metadata.max.age.ms300000 元数据在限定时间内没有进行更新则会被强制更新
reconnect.backoff.ms50ms 尝试重新连接指定主机之前的退避时间
retry.backoff.ms100ms 尝试重新拉取数据的重试间隔
isolation.levelread_uncommitted 隔离级别 决定消费者能读到什么样的数据 read_uncommitted可以消费到 LSOLastStableOffset位置 read_committed可以消费到 HWHigh Watermark位置
max.poll.interval.ms 超过时限没有发起 poll 操作则消费组认为该消费者已离开消费组
enable.auto.committrue 开启消费位移的自动提交
auto.commit.interval.ms5000 自动提交消费位移的时间间隔
API 开发topic 管理
如果希望将管理类的功能集成到公司内部的系统中打造集管理、监控、运维、告警为一体的生态平台那么就需要以程序调用 API 方式去实现。如下所示 为什么在网页上或者用命令都能做到topic的创建、删除、列出、查看详情其底层逻辑是什么
其本质上是通过不同方式调用了kafka提供的创建topic api 底层逻辑web的后端java程序中去调用相关的api
工具类 KafkaAdminClient 可以用来管理 broker、配置和 ACL Access Control List管理 topic 构造一个 KafkaAdminClient
AdminClient adminClient KafkaAdminClient.create(props);列出主题
ListTopicsResult listTopicsResult adminClient.listTopics();
SetString topics listTopicsResult.names().get();
System.out.println(topics);查看主题信息
DescribeTopicsResult describeTopicsResult
adminClient.describeTopics(Arrays.asList(tpc_4, tpc_3));
MapString, TopicDescription res describeTopicsResult.all().get();
SetString ksets res.keySet();
for (String k : ksets) {System.out.println(res.get(k));
}创建主题
// 参数配置
Properties props new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,doit01:9092,doit02:9092,doit03:
9092);
props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);// 创建 admin client 对象
AdminClient adminClient KafkaAdminClient.create(props);
// 由服务端 controller 自行分配分区及副本所在 broker
NewTopic tpc_3 new NewTopic(tpc_3, 2, (short) 1);
// 手动指定分区及副本的 broker 分配
HashMapInteger, ListInteger replicaAssignments new HashMap();
// 分区 0分配到 broker0broker1
replicaAssignments.put(0,Arrays.asList(0,1));
// 分区 1分配到 broker0,broker2
replicaAssignments.put(1,Arrays.asList(0,1));NewTopic tpc_4 new NewTopic(tpc_4, replicaAssignments);
CreateTopicsResult result adminClient.createTopics(Arrays.asList(tpc_3,tpc_4));// 从 future 中等待服务端返回
try {result.all().get();
} catch (Exception e) {e.printStackTrace();
}
adminClient.close();删除主题
DeleteTopicsResult deleteTopicsResult adminClient.deleteTopics(Arrays.asList(tpc_1,
tpc_1));
MapString, KafkaFutureVoid values deleteTopicsResult.values();
System.out.println(values);其他管理
除了进行 topic 管理外KafkaAdminClient 也可进行诸如动态参数管理分区管理等各类管理操作
练习
需求
写一个生产者不断地去生成“用户行为事件” 并写入kafka
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789}
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789}
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789}
…
写一个消费者不断地从kafka种取消费如上 “ 用户行为事件” 数据并做统计计算每五分钟输出一次截至到当时的数据中出现过的用户总数。 本质 去重guid数
消费者
首先分析思路可以使用hashmap来进行统计思路如下所示 如果存在线程安全的问题可以使用ConcurrentHashMap。
还有一个业务需求就是 每五分钟去输出一个结果这个时机、节奏如何去把控。
实际上就是一个线程源源不断的拉数据一个线程定期的输出结果。
public class consumer{public static void main(String[] args){ConcurrentHashMapLong, String guidMap new ConcurrentHashMap();// 启动数据消费线程new Thread(new ConsumeRunnable(guidMap)).start();// 启动统计及消费输出结果的线程每5s输出一次// 优雅一点实现定时调度可以用各种定时调度器有第三方的也可以用JDK自己的TimerTimer timer new Timer();timer.scheduleAtFixedRate(new StatisticTask(guidMap),0,1000);}
}class ConsumeRunnable implements Runnable{ConcurrentHashMapLong, String guidMap;public ConsumeRunnable(ConcurrentHashMapLong, String guidMap){this.guidMap guidMap;}public void run(){Properties props new Properties();// 定义 kakfa 服务的地址不需要将所有 broker 指定上props.put(bootstrap.servers, doitedu01);// key 的反序列化类props.put(key.deserializer,org.apache.kafka.common.serialization.StringDeserializer);// value 的反序列化类props.put(value.deserializer,org.apache.kafka.common.serialization.StringDeserializer);// 如果没有消费偏移量记录则自动重设为起始 offsetlatest, earliest, noneprops.put(auto.offset.reset,latest);// 制定 consumer groupprops.put(group.id, g1);while (true) {// 读取数据读取超时时间为 5000msConsumerRecordsString, String records consumer.poll(5000);for (ConsumerRecordString, String record : records)String eventJson record.value();UserEvent userEvent JSON.parseObject(eventJson,UserEvent.class);guidMap.put(userEvent.getGuid(), );}}
}class StatisticTask extends TimerTask{ConcurrentHashMapLong, String guidMap;public StatisticTask(ConcurrentHashMapLong, String guidMap){this.guidMap guidMap;}public void run(){System.out.println(截止到当前的用户总数为guidMap.size());}
}ConcurrentHashMap 中无论 key 还是 Value都不能填 null值具体在源码中有体现
其 put流程如下 public V put(K key, V value) {return putVal(key, value, false);// onlyIfAbsent如果是false那么每次都会用新值替换掉旧值}final V putVal(K key, V value, boolean onlyIfAbsent) {if (key null || value null) throw new NullPointerException();// 其中 spread 方法会综合高位低位, 具有更好的 hash 性int hash spread(key.hashCode());int binCount 0;// 死循环for (NodeK,V[] tab table;;) {// f 是链表头节点// fh 是链表头结点的 hash// i 是链表在 table 中的下标NodeK,V f; int n, i, fh;// 要创建 tableif (tab null || (n tab.length) 0)// 初始化 table 使用了 cas, 无需 synchronized 创建成功, 进入下一轮循环// 因为是懒惰初始化的所以直到现在才开始创建 初始化使用cas 创建其它失败得再次进入循环没有用syn 我们得线程并没有被阻塞住tab initTable();// 要创建链表头节点else if ((f tabAt(tab, i (n - 1) hash)) null) {// 添加链表头使用了 cas, 无需 synchronized// 用cas将头节点加进去如果加入失败了继续循环if (casTabAt(tab, i, null,new NodeK,V(hash, key, value, null)))break;}// 帮忙扩容// 其实就是看你的头结点是不是 ForwardingNode其对应得MOVED是一个负数else if ((fh f.hash) MOVED)// 帮忙之后, 进入下一轮循环// 锁住当前的链表帮助去扩容tab helpTransfer(tab, f);// 能进入这个else说明 table既不处于扩容中也不是处于table的初始化过程中而且这时肯定发生了锁下标的冲突else {V oldVal null;// 锁住链表头节点// 并没有锁住整个tab而是锁住这个桶链表的头节点synchronized (f) {// 再次确认链表头节点没有被移动if (tabAt(tab, i) f) {// 链表// 链表的头节点hash码大于等于 0 if (fh 0) {binCount 1;// 遍历链表for (NodeK,V e f;; binCount) {K ek;
// 找到相同的 keyif (e.hash hash ((ek e.key) key ||(ek ! null key.equals(ek)))) {oldVal e.val;// 更新if (!onlyIfAbsent)e.val value;break;}NodeK,V pred e;// 已经是最后的节点了, 新增 Node, 追加至链表尾if ((e e.next) null) {pred.next new NodeK,V(hash, key,value, null);break;}}}// 红黑树else if (f instanceof TreeBin) {NodeK,V p;binCount 2;// putTreeVal 会看 key 是否已经在树中, 是, 则返回对应的 TreeNodeif ((p ((TreeBinK,V)f).putTreeVal(hash, key,value)) ! null) {oldVal p.val;if (!onlyIfAbsent)p.val value;}}}// 释放链表头节点的锁}if (binCount ! 0) {if (binCount TREEIFY_THRESHOLD)// 如果链表长度 树化阈值(8), 进行链表转为红黑树treeifyBin(tab, i);if (oldVal ! null)return oldVal;break;}}}// 增加 size 计数addCount(1L, binCount);return null;}private final NodeK,V[] initTable() {NodeK,V[] tab; int sc;// 这个hash有没有被创建while ((tab table) null || tab.length 0) {if ((sc sizeCtl) 0)// 让出cpu的使用权如果cpu的时间片没有其它线程了那么还是会分给这个线程只是让他不至于充分利用cpu少占用一点cpu的时间。Thread.yield();// 尝试将 sizeCtl 设置为 -1表示初始化 table// 而其它的线程再次进入循环首先 不小于0了其次之前的 sc也已经变了cas失败再次循环的时候发现 tab已经不为空了结束循环else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {// 获得锁, 创建 table, 这时其它线程会在 while() 循环中 yield 直至 table 创建try {if ((tab table) null || tab.length 0) {int n (sc 0) ? sc : DEFAULT_CAPACITY;NodeK,V[] nt (NodeK,V[])new Node?,?[n];table tab nt;// 计算出下一次要扩容的阈值sc n - (n 2);}} finally {// 计算出下一次要扩容的阈值sizeCtl sc;}break;}}return tab;}// check 是之前 binCount 的个数// 运用了 longadder 的思想private final void addCount(long x, int check) {CounterCell[] as; long b, s;if (// 已经有了 counterCells, 向 cell 累加// 累加单元数组不为空(as counterCells) ! null ||// 还没有, 向 baseCount 累加// 一个基础数值累加!U.compareAndSwapLong(this, BASECOUNT, b baseCount, s b x)) {CounterCell a; long v; int m;boolean uncontended true;if (// 还没有 counterCellsas null || (m as.length - 1) 0 ||// 还没有 cell(a as[ThreadLocalRandom.getProbe() m]) null ||// cell cas 增加计数失败!(uncontended U.compareAndSwapLong(a, CELLVALUE, v a.value, v x))) {// 创建累加单元数组和cell, 累加重试fullAddCount(x, uncontended);return;}if (check 1)return;// 获取元素个数s sumCount();}if (check 0) {NodeK,V[] tab, nt; int n, sc;// 看看元素的个数是否大于扩容的阈值while (s (long)(sc sizeCtl) (tab table) ! null (n tab.length) MAXIMUM_CAPACITY) {int rs resizeStamp(n);if (sc 0) {if ((sc RESIZE_STAMP_SHIFT) ! rs || sc rs 1 ||sc rs MAX_RESIZERS || (nt nextTable) null ||transferIndex 0)break;// newtable 已经创建了帮忙扩容if (U.compareAndSwapInt(this, SIZECTL, sc, sc 1))// 首次调用因为是懒惰初始化的所以还没有创建transfer(tab, nt);}// 需要扩容这时 newtable 未创建else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs RESIZE_STAMP_SHIFT) 2))transfer(tab, null);s sumCount();}}}但是这里使用一个hashmap去记录去重的guid是有弊端的。
利用map或者set来记录guid本身比较占用内存还有性能。
guid是一个Long值8个byte如果咱们的用户规模达到了10亿那你这个set或者map中存储的guid就将达到10亿个。
相当于8 * 1000000000 /1024/1024 7629MB相当于一个Hashmap中存储了7G数据
那么有一些经典的数据结构bitmap、bloomfilter、hyperloglog都可以解决去统计个数、判重的场景。 输出个数有几个1就是出现过几种 guid
之前使用hashmap存直接存guid8个byte一个guid 64bit 现在bitmap里面记一个guid只要一个bit
并且hashmap是按需扩容的这个过程是很浪费性能的而bigmap要一开始初始化10亿个bit有点浪费空间。
但是api会对最原始的bitmap进行一些优化比如说稀疏型的向量的场景中例如roaringbitmap工具包进行了优化慢慢的进行了扩容。
public class consumer{public static void main(String[] args){// 使用roaringbitmap来记录RoaringBitmap bitmap RoaringBitmap.bigmapOf();// 启动数据消费线程new Thread(new ConsumeRunnable(bitmap)).start();// 启动统计及消费输出结果的线程每5s输出一次// 优雅一点实现定时调度可以用各种定时调度器有第三方的也可以用JDK自己的TimerTimer timer new Timer();timer.scheduleAtFixedRate(new StatisticTask(bitmap),0,1000);}
}class ConsumeRunnable implements Runnable{RoaringBitmap bitmap;public ConsumeRunnable(RoaringBitmap bitmap){this.bitmap bitmap;}public void run(){Properties props new Properties();// 定义 kakfa 服务的地址不需要将所有 broker 指定上props.put(bootstrap.servers, doitedu01);// key 的反序列化类props.put(key.deserializer,org.apache.kafka.common.serialization.StringDeserializer);// value 的反序列化类props.put(value.deserializer,org.apache.kafka.common.serialization.StringDeserializer);// 如果没有消费偏移量记录则自动重设为起始 offsetlatest, earliest, noneprops.put(auto.offset.reset,latest);// 制定 consumer groupprops.put(group.id, g1);while (true) {// 读取数据读取超时时间为 5000msConsumerRecordsString, String records consumer.poll(5000);for (ConsumerRecordString, String record : records)String eventJson record.value();UserEvent userEvent JSON.parseObject(eventJson,UserEvent.class);bitmap.add(userEvent.getGuid());}}
}class StatisticTask extends TimerTask{RoaringBitmap bitmap;public StatisticTask(RoaringBitmap bitmap){this.bitmap bitmap;}public void run(){System.out.println(截止到当前的用户总数为bitmap.getCardinality());}
}但是如果涉及到了多维聚合那么单纯的使用bitmap可能还不够
比如 省市区 用户数 、 省市 用户数
江苏省南京市鼓楼区5
江苏省南京市下关区6
…
从用户访问记录中统计各区域的用户数
江苏省南京市鼓楼区[0,1,0,1,1,1,0,1,0,0,0,0] 5
江苏省南京市下关区[1,1,0,0,0,0,0,0,1,1,1,1] 6
要到的江苏省南京市 [1,1,0,1,1,1,0,1,1,1,1,1] 10
这也就是使用bitmap来实现高效率的维度再均衡
所以这也就是数据结构与应用解耦
消费者
需求2写一个消费者不断地从kafka中消费如上用户行为事件并做出如下加工处理
给每一条数据添加一个字段来标识该条数据所属的用户今天是否第一次出现如是则标注1否则标注0
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789,“flag”:1}
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789,“flag”:0}
{“guid”:1,“eventId”:“pageview”,“timestamp”:163786834789,“flag”:1}
这个需求可以用HashMap来做
也可以用bitmap来做启动bitmap.contain()能够判断是是否包含
如果是判重字符串用bitmap就不太方便了可以使用bloomfilter但是bloomfilter会牺牲一定的准确率因为bloomfilter会误判
在判重方面布隆过滤器比bitmap.contain()更适合。
布隆过滤器具有以下优势
内存占用更小布隆过滤器使用位数组和哈希函数来表示数据集合相比于bitmap.contain()方法它可以以较小的内存占用来存储大规模的数据集合。高效的查询性能布隆过滤器可以在常数时间内完成判重操作即使数据集合非常庞大也能以较高的速度进行查询。而bitmap.contain()方法可能需要遍历整个位图数组时间复杂度较高。可扩展性布隆过滤器可以动态地添加新的元素并且支持删除操作。相比之下bitmap.contain()方法需要预先确定数据范围不便于动态扩展。
虽然布隆过滤器具有以上优势但也需要注意它存在一定的误判率即可能将不存在的元素判断为存在。因此在判重的关键场景中如果对结果的准确性要求非常高可以使用其他更精确的去重方法进行验证。
public class consumer{public static void main(String[] args){ // 启动数据消费线程new Thread(new ConsumeRunnable()).start();}
}class ConsumeRunnable implements Runnable{BloomFilterLong bloomFilter;Properties props;public ConsumeRunnable(){bloomFilter BloomFilter.create(Funnels.longFunnel(),1000000000,0.01);props new Properties();// 定义 kakfa 服务的地址不需要将所有 broker 指定上props.put(bootstrap.servers, doitedu01);// key 的反序列化类props.put(key.deserializer,org.apache.kafka.common.serialization.StringDeserializer);// value 的反序列化类props.put(value.deserializer,org.apache.kafka.common.serialization.StringDeserializer);// 如果没有消费偏移量记录则自动重设为起始 offsetlatest, earliest, noneprops.put(auto.offset.reset,latest);// 制定 consumer groupprops.put(group.id, g1);}public void run(){while (true) {// 读取数据读取超时时间为 5000msConsumerRecordsString, String records consumer.poll(5000);for (ConsumerRecordString, String record : records)String eventJson record.value();UserEvent userEvent JSON.parseObject(eventJson,UserEvent.class);// 去布隆过滤器判断一下boolean mightContain bloomFilter.mightContain(userEvent.getGuid());if(mightContain){userEvent.setFlag(0);}else{userEvent.setFlag(1);bloomFilter.put(userEvent.getGuid());}}}
}
