金华网站建设公司招聘,自建站shopify,wordpress 文章多图,建站公司推荐首推万维科技文章目录 1、从基础的客户端说起1.1、消息发送者主流程1.2、消息消费者主流程 2、从客户端属性来梳理客户端工作机制2.1、消费者分组消费机制 1、从基础的客户端说起
Kafka提供了非常简单的客户端API。只需要引入一个Maven依赖即可#xff1a; dependencygroupId… 文章目录 1、从基础的客户端说起1.1、消息发送者主流程1.2、消息消费者主流程 2、从客户端属性来梳理客户端工作机制2.1、消费者分组消费机制 1、从基础的客户端说起
Kafka提供了非常简单的客户端API。只需要引入一个Maven依赖即可 dependencygroupIdorg.apache.kafka/groupIdartifactIdkafka_2.13/artifactIdversion3.4.0/version/dependency1.1、消息发送者主流程
然后可以使用Kafka提供的Producer类快速发送消息。
public class MyProducer {private static final String BOOTSTRAP_SERVERS worker1:9092,worker2:9092,worker3:9092;private static final String TOPIC disTopic;public static void main(String[] args) throws ExecutionException, InterruptedException {//PART1:设置发送者相关属性Properties props new Properties();// 此处配置的是kafka的端口props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);// 配置key的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,org.apache.kafka.common.serialization.StringSerializer);// 配置value的序列化类props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,org.apache.kafka.common.serialization.StringSerializer);ProducerString,String producer new KafkaProducer(props);CountDownLatch latch new CountDownLatch(5);for(int i 0; i 5; i) {//Part2:构建消息ProducerRecordString, String record new ProducerRecord(TOPIC, Integer.toString(i), MyProducer i);//Part3:发送消息//单向发送不关心服务端的应答。producer.send(record);System.out.println(message i sended);//同步发送获取服务端应答消息前会阻塞当前线程。RecordMetadata recordMetadata producer.send(record).get();String topic recordMetadata.topic();int partition recordMetadata.partition();long offset recordMetadata.offset();String message recordMetadata.toString();System.out.println(message:[ message] sended with topic:topic; partition:partition ;offset:offset);//异步发送消息发送后不阻塞服务端有应答后会触发回调函数producer.send(record, new Callback() {Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if(null ! e){System.out.println(消息发送失败,e.getMessage());e.printStackTrace();}else{String topic recordMetadata.topic();long offset recordMetadata.offset();String message recordMetadata.toString();System.out.println(message:[ message] sended with topic:topic;offset:offset);}latch.countDown();}});}//消息处理完才停止发送者。latch.await();producer.close();}
} 整体来说构建Producer分为三个步骤
设置Producer核心属性 Producer可选的属性都可以由ProducerConfig类管理。比如ProducerConfig.BOOTSTRAP_SERVERS_CONFIG属性显然就是指发送者要将消息发到哪个Kafka集群上。这是每个Producer必选的属性。在ProducerConfig中对于大部分比较重要的属性都配置了对应的DOC属性进行描述。构建消息Kafka的消息是一个Key-Value结构的消息。其中key和value都可以是任意对象类型。其中key主要是用来进行Partition分区的业务上更关心的是value。使用Producer发送消息通常用到的就是单向发送、同步发送和异步发送者三种发送方式。
1.2、消息消费者主流程
接下来可以使用Kafka提供的Consumer类快速消费消息。
public class MyConsumer {private static final String BOOTSTRAP_SERVERS worker1:9092,worker2:9092,worker3:9092;private static final String TOPIC disTopic;public static void main(String[] args) {//PART1:设置发送者相关属性Properties props new Properties();//kafka地址props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);//每个消费者要指定一个groupprops.put(ConsumerConfig.GROUP_ID_CONFIG, test);//key序列化类props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);//value序列化类props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);ConsumerString, String consumer new KafkaConsumer(props);consumer.subscribe(Arrays.asList(TOPIC));while (true) {//PART2:拉取消息// 100毫秒超时时间ConsumerRecordsString, String records consumer.poll(Duration.ofNanos(100));//PART3:处理消息for (ConsumerRecordString, String record : records) {System.out.println(offset record.offset() ;key record.key() ; value record.value());}//提交offset消息就不会重复推送。consumer.commitSync(); //同步提交表示必须等到offset提交完毕再去消费下一批数据。
// consumer.commitAsync(); //异步提交表示发送完提交offset请求后就开始消费下一批数据了。不用等到Broker的确认。}}
} 整体来说Consumer同样是分为三个步骤
设置Consumer核心属性 可选的属性都可以由ConsumerConfig类管理。在这个类中同样对于大部分比较重要的属性都配置了对应的DOC属性进行描述。同样BOOTSTRAP_SERVERS_CONFIG是必须设置的属性。拉取消息Kafka采用Consumer主动拉取消息的Pull模式。consumer主动从Broker上拉取一批感兴趣的消息。处理消息提交位点消费者将消息拉取完成后就可以交由业务自行处理对应的这一批消息了。只是消费者需要向Broker提交偏移量offset。如果不提交OffsetBroker会认为消费者端消息处理失败了还会重复进行推送。 Kafka的客户端基本就是固定的按照这三个大的步骤运行。在具体使用过程中最大的变数基本上就是给生产者和消费者的设定合适的属性。这些属性极大的影响了客户端程序的执行方式。
2、从客户端属性来梳理客户端工作机制
渔与鱼Kafka的客户端API的重要目的就是想要简化客户端的使用方式所以对于API的使用尽量熟练就可以了。对于其他重要的属性都可以通过源码中的描述去学习并且可以设计一些场景去进行验证。其重点是要逐步在脑海之中建立一个Message在Kafka集群中进行流转的基础模型。
其实Kafka的设计精髓是在网络不稳定服务也随时会崩溃的这些作死的复杂场景下如何保证消息的高并发、高吞吐那才是Kafka最为精妙的地方。但是要理解那些复杂的问题都是需要建立在这个基础模型基础上的。
2.1、消费者分组消费机制
这是我们在使用kafka时最为重要的一个机制因此最先进行梳理。
在Consumer中都需要指定一个GROUP_ID_CONFIG属性这表示当前Consumer所属的消费者组。他的描述是这样的 public static final String GROUP_ID_CONFIG group.id;public static final String GROUP_ID_DOC A unique string that identifies the consumer group this consumer belongs to. This property is required if the consumer uses either the group management functionality by using codesubscribe(topic)/code or the Kafka-based offset management strategy.;既然这里提到了kafka-based offset management strategy那是不是也有非Kafka管理Offset的策略呢
另外还有一个相关的参数GROUP_INSTANCE_ID_CONFIG可以给组成员设置一个固定的instanceId这个参数通常可以用来减少Kafka不必要的rebalance。
从这段描述中看到对于Consumer如果需要在subcribe时使用组管理功能以及Kafka提供的offset管理策略那就必须要配置GROUP_ID_CONFIG属性。这个分组消费机制简单描述就是这样的
image.png
生产者往Topic下发消息时会尽量均匀的将消息发送到Topic下的各个Partition当中。而这个消息会向所有订阅了该Topic的消费者推送。推送时每个ConsumerGroup中只会推送一份。也就是同一个消费者组中的多个消费者实例只会共同消费一个消息副本。而不同消费者组之间会重复消费消息副本。这就是消费者组的作用。
与之相关的还有Offset偏移量。这个偏移量表示每个消费者组在每个Partiton中已经消费处理的进度。在Kafka中可以看到消费者组的Offset记录情况。
[operworker1 bin]$ ./kafka-consumer-groups.sh --bootstrap-server worker1:9092 --describe --group test
文章转载自： http://www.morning.smfbw.cn.gov.cn.smfbw.cn http://www.morning.gcftl.cn.gov.cn.gcftl.cn http://www.morning.xxwhz.cn.gov.cn.xxwhz.cn http://www.morning.kbynw.cn.gov.cn.kbynw.cn http://www.morning.fnkcg.cn.gov.cn.fnkcg.cn http://www.morning.dmzzt.cn.gov.cn.dmzzt.cn http://www.morning.qtbnm.cn.gov.cn.qtbnm.cn http://www.morning.ppllj.cn.gov.cn.ppllj.cn http://www.morning.mwkwg.cn.gov.cn.mwkwg.cn http://www.morning.ndcjq.cn.gov.cn.ndcjq.cn http://www.morning.xwlmr.cn.gov.cn.xwlmr.cn http://www.morning.rxgnn.cn.gov.cn.rxgnn.cn http://www.morning.mzhhr.cn.gov.cn.mzhhr.cn http://www.morning.rkxdp.cn.gov.cn.rkxdp.cn http://www.morning.lxbml.cn.gov.cn.lxbml.cn http://www.morning.lffrh.cn.gov.cn.lffrh.cn http://www.morning.glkhx.cn.gov.cn.glkhx.cn http://www.morning.fxqjz.cn.gov.cn.fxqjz.cn http://www.morning.zxwqt.cn.gov.cn.zxwqt.cn http://www.morning.kpcxj.cn.gov.cn.kpcxj.cn http://www.morning.nfyc.cn.gov.cn.nfyc.cn http://www.morning.gmztd.cn.gov.cn.gmztd.cn http://www.morning.tsrg.cn.gov.cn.tsrg.cn http://www.morning.gcfrt.cn.gov.cn.gcfrt.cn http://www.morning.kjdxh.cn.gov.cn.kjdxh.cn http://www.morning.nflpk.cn.gov.cn.nflpk.cn http://www.morning.xbtlt.cn.gov.cn.xbtlt.cn http://www.morning.pgmyn.cn.gov.cn.pgmyn.cn http://www.morning.qtqjx.cn.gov.cn.qtqjx.cn http://www.morning.nwjzc.cn.gov.cn.nwjzc.cn http://www.morning.xqtqm.cn.gov.cn.xqtqm.cn http://www.morning.bynf.cn.gov.cn.bynf.cn http://www.morning.rqsnl.cn.gov.cn.rqsnl.cn http://www.morning.clbzy.cn.gov.cn.clbzy.cn http://www.morning.nchlk.cn.gov.cn.nchlk.cn http://www.morning.rlfr.cn.gov.cn.rlfr.cn http://www.morning.jhrtq.cn.gov.cn.jhrtq.cn http://www.morning.xinyishufa.cn.gov.cn.xinyishufa.cn http://www.morning.ftmp.cn.gov.cn.ftmp.cn http://www.morning.rjynd.cn.gov.cn.rjynd.cn http://www.morning.nmrtb.cn.gov.cn.nmrtb.cn http://www.morning.nkddq.cn.gov.cn.nkddq.cn http://www.morning.mcgsq.cn.gov.cn.mcgsq.cn http://www.morning.cyhlq.cn.gov.cn.cyhlq.cn http://www.morning.xxwfq.cn.gov.cn.xxwfq.cn http://www.morning.klyyd.cn.gov.cn.klyyd.cn http://www.morning.cniedu.com.gov.cn.cniedu.com http://www.morning.3ox8hs.cn.gov.cn.3ox8hs.cn http://www.morning.hwcgg.cn.gov.cn.hwcgg.cn http://www.morning.wfzdh.cn.gov.cn.wfzdh.cn http://www.morning.zlkps.cn.gov.cn.zlkps.cn http://www.morning.wprxm.cn.gov.cn.wprxm.cn http://www.morning.gmztd.cn.gov.cn.gmztd.cn http://www.morning.bhpsz.cn.gov.cn.bhpsz.cn http://www.morning.bmsqq.cn.gov.cn.bmsqq.cn http://www.morning.slwfy.cn.gov.cn.slwfy.cn http://www.morning.zmwzg.cn.gov.cn.zmwzg.cn http://www.morning.chhhq.cn.gov.cn.chhhq.cn http://www.morning.qfgwx.cn.gov.cn.qfgwx.cn http://www.morning.dyfmh.cn.gov.cn.dyfmh.cn http://www.morning.sbkb.cn.gov.cn.sbkb.cn http://www.morning.psxxp.cn.gov.cn.psxxp.cn http://www.morning.mhnd.cn.gov.cn.mhnd.cn http://www.morning.ndmh.cn.gov.cn.ndmh.cn http://www.morning.wwjft.cn.gov.cn.wwjft.cn http://www.morning.qqnh.cn.gov.cn.qqnh.cn http://www.morning.rykgh.cn.gov.cn.rykgh.cn http://www.morning.mtktn.cn.gov.cn.mtktn.cn http://www.morning.sgmgz.cn.gov.cn.sgmgz.cn http://www.morning.rmyt.cn.gov.cn.rmyt.cn http://www.morning.gwjnm.cn.gov.cn.gwjnm.cn http://www.morning.yzdth.cn.gov.cn.yzdth.cn http://www.morning.smmby.cn.gov.cn.smmby.cn http://www.morning.yrcxg.cn.gov.cn.yrcxg.cn http://www.morning.sfnjr.cn.gov.cn.sfnjr.cn http://www.morning.jbhhj.cn.gov.cn.jbhhj.cn http://www.morning.glnxd.cn.gov.cn.glnxd.cn http://www.morning.hyryq.cn.gov.cn.hyryq.cn http://www.morning.ztfzm.cn.gov.cn.ztfzm.cn http://www.morning.kwnbd.cn.gov.cn.kwnbd.cn
