《二十:消费者端统计加工消息》
2022/1/13 23:33:31
本文主要是介绍《二十:消费者端统计加工消息》,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!
自定义拦截器实现ProducerInterceptor
package com.wangbiao.kafka.interceptor; import org.apache.kafka.clients.producer.ProducerInterceptor; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import java.util.Map; /** * 拦截改造消费的消息 记录改造value用竖线分割(下游自行解析) */ public class TimeInterceptor implements ProducerInterceptor<String,String> { @Override public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) { String value = record.value(); //记录改造value用竖线分割 return new ProducerRecord(record.topic(),record.partition(),record.key(),System.currentTimeMillis()+"|"+value); } @Override public void onAcknowledgement(RecordMetadata metadata, Exception exception) { } @Override public void close() { } @Override public void configure(Map<String, ?> configs) { } }
统计消费了多少数据
package com.wangbiao.kafka.interceptor; import org.apache.kafka.clients.producer.ProducerInterceptor; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import java.util.Map; /** * 统计拦截多少失败多少数据 */ public class CounterInterceptor implements ProducerInterceptor { public int success; public int error; @Override public ProducerRecord onSend(ProducerRecord record) { return record; } @Override public void onAcknowledgement(RecordMetadata metadata, Exception exception) { if (metadata!=null){ if(success==Integer.MAX_VALUE){ success=0; success++; } success++; }else { if(error==Integer.MAX_VALUE){error=0; error++; } error++; } } @Override public void close() { System.out.println("success:"+success); System.out.println("error:"+error); } @Override public void configure(Map<String, ?> configs) { } }
这篇关于《二十:消费者端统计加工消息》的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!
- 2024-05-25外企也半夜发布上线吗?
- 2024-05-24鸿蒙原生应用再新丁!芒果TV 入局鸿蒙
- 2024-05-22基本概念
- 2024-05-22检索数据
- 2024-05-22排序数据
- 2024-05-22基础过滤数据
- 2024-05-22通过逻辑操作符过滤数据
- 2024-05-22通过通配符过滤数据
- 2024-05-22字段的拼接与计算
- 2024-05-22聚合函数