精品国产精品乱码一区二区三区,国产av不卡一二区

消費順序
一個partition同一個時刻在一個consumer group中只能有一個consumer instance在消費，從而保證消費順序。
consumer group中的consumer instance的數(shù)量不能比一個Topic中的partition的數(shù)量多，否則，多出來的consumer消費不到消息。
Kafka只在partition的范圍內(nèi)保證消息消費的局部順序性，不能在同一個topic中的多個partition中保證總的消費順序性。
如果有在總體上保證消費順序的需求，那么我們可以通過將topic的partition數(shù)量設(shè)置為1，將consumer group中的consumer instance數(shù)量也設(shè)置為1，但是這樣會影響性能，所以kafka的順序消費很少用。
Java客戶端訪問Kafka引入maven依賴<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.4.1</version> </dependency>
消息發(fā)送端代碼(生產(chǎn)者)package com.yundasys.usercenter.collect.api.vo.req;import com.alibaba.fastjson.JSON;import com.fasterxml.jackson.databind.ser.std.StringSerializer;import org.apache.kafka.clients.producer.*;import java.util.Properties;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutionException;import java.util.concurrent.TimeUnit;/** * @program: usercenter-portrait-collect * @description: MsgProducer * @author: yxh-word * @create: 2021-07-14 * @version: v1.0.0 創(chuàng)建文件, yxh-word, 2021-07-14 **/public class MsgProducer {private final static String TOPIC_NAME = "my-replicated-topic";public static void main(String[] args) throws InterruptedException, ExecutionException {Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");/*發(fā)出消息持久化機制參數(shù)（1）acks=0：表示producer不需要等待任何broker確認(rèn)收到消息的回復(fù)，就可以繼續(xù)發(fā)送下一條消息。性能最高，但是最容易丟消息。（2）acks=1：至少要等待leader已經(jīng)成功將數(shù)據(jù)寫入本地log，但是不需要等待所有follower是否成功寫入。就可以繼續(xù)發(fā)送下一條消息。這種情況下，如果follower沒有成功備份數(shù)據(jù)，而此時leader又掛掉，則消息會丟失。（3）acks=-1或all：需要等待 min.insync.replicas(默認(rèn)為1，推薦配置大于等于2) 這個參數(shù)配置的副本個數(shù)都成功寫入日志，這種策略會保證只要有一個備份存活就不會丟失數(shù)據(jù) 。這是最強的數(shù)據(jù)保證。一般除非是金融級別，或跟錢打交道的場景才會使用這種配置。*//*props.put(ProducerConfig.ACKS_CONFIG, "1");*//*發(fā)送失敗會重試，默認(rèn)重試間隔100ms，重試能保證消息發(fā)送的可靠性，但是也可能造成消息重復(fù)發(fā)送，比如網(wǎng)絡(luò)抖動，所以需要在接收者那邊做好消息接收的冪等性處理*//*props.put(ProducerConfig.RETRIES_CONFIG, 3);//重試間隔設(shè)置props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);//設(shè)置發(fā)送消息的本地緩沖區(qū)，如果設(shè)置了該緩沖區(qū)，消息會先發(fā)送到本地緩沖區(qū)，可以提高消息發(fā)送性能，默認(rèn)值是33554432，即32MBprops.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);*//*kafka本地線程會從緩沖區(qū)取數(shù)據(jù)，批量發(fā)送到broker，設(shè)置批量發(fā)送消息的大小，默認(rèn)值是16384，即16kb，就是說一個batch滿了16kb就發(fā)送出去*//*props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);*//*默認(rèn)值是0，意思就是消息必須立即被發(fā)送，但這樣會影響性能一般設(shè)置10毫秒左右，就是說這個消息發(fā)送完后會進(jìn)入本地的一個batch，如果10毫秒內(nèi)，這個batch滿了16kb就會隨batch一起被發(fā)送出去如果10毫秒內(nèi)，batch沒滿，那么也必須把消息發(fā)送出去，不能讓消息的發(fā)送延遲時間太長*//*props.put(ProducerConfig.LINGER_MS_CONFIG, 10);*///把發(fā)送的key從字符串序列化為字節(jié)數(shù)組props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//把發(fā)送消息value從字符串序列化為字節(jié)數(shù)組props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());Producer<String, String> producer = new KafkaProducer<String, String>(props);int msgNum = 5;final CountDownLatch countDownLatch = new CountDownLatch(msgNum);for (int i = 1; i <= msgNum; i++) {Order order = new Order(i, 100 + i, 1, 1000.00);//指定發(fā)送分區(qū)/*ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, 0, order.getOrderId().toString(), JSON.toJSONString(order));*///未指定發(fā)送分區(qū)，具體發(fā)送的分區(qū)計算公式：hash(key)%partitionNumProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, order.getOrderId().toString(), JSON.toJSONString(order));//等待消息發(fā)送成功的同步阻塞方法/*RecordMetadata metadata = https://www.520longzhigu.com/diannao/producer.send(producerRecord).get();System.out.println("同步方式發(fā)送消息結(jié)果：" + "topic-" + metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" + metadata.offset());*///異步回調(diào)方式發(fā)送消息producer.send(producerRecord, new Callback() {public void onCompletion(RecordMetadata metadata, Exception exception) {if (exception != null) {System.err.println("發(fā)送消息失?。? + exception.getStackTrace());}if (metadata != null) {System.out.println("異步方式發(fā)送消息結(jié)果：" + "topic-" + metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" + metadata.offset());}countDownLatch.countDown();}});//送積分 TODO}countDownLatch.await(5, TimeUnit.SECONDS);producer.close();}}消費者代碼package com.yundasys.usercenter.collect.api.vo.req;import com.fasterxml.jackson.databind.deser.std.StringDeserializer;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;import java.util.Arrays;import java.util.Properties;/** * @program: usercenter-portrait-collect * @description: MsgConsumer * @author: yxh-word * @create: 2021-07-14 * @version: v1.0.0 創(chuàng)建文件, yxh-word, 2021-07-14 **/public class MsgConsumer {private final static String TOPIC_NAME = "my-replicated-topic";private final static String CONSUMER_GROUP_NAME = "testGroup";public static void main(String[] args) {Properties props = new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");// 消費分組名props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);// 是否自動提交offset，默認(rèn)就是trueprops.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");// 自動提交offset的間隔時間props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");//props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");/*當(dāng)消費主題的是一個新的消費組，或者指定offset的消費方式，offset不存在，那么應(yīng)該如何消費latest(默認(rèn)) ：只消費自己啟動之后發(fā)送到主題的消息earliest：第一次從頭開始消費，以后按照消費offset記錄繼續(xù)消費，這個需要區(qū)別于consumer.seekToBeginning(每次都從頭開始消費)*///props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");/*consumer給broker發(fā)送心跳的間隔時間，broker接收到心跳如果此時有rebalance發(fā)生會通過心跳響應(yīng)將rebalance方案下發(fā)給consumer，這個時間可以稍微短一點*/props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);/*服務(wù)端broker多久感知不到一個consumer心跳就認(rèn)為他故障了，會將其踢出消費組，對應(yīng)的Partition也會被重新分配給其他consumer，默認(rèn)是10秒*/props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);//一次poll最大拉取消息的條數(shù)，如果消費者處理速度很快，可以設(shè)置大點，如果處理速度一般，可以設(shè)置小點props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);/*如果兩次poll操作間隔超過了這個時間，broker就會認(rèn)為這個consumer處理能力太弱，會將其踢出消費組，將分區(qū)分配給別的consumer消費*/props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);consumer.subscribe(Arrays.asList(TOPIC_NAME));// 消費指定分區(qū)//consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));//消息回溯消費/*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));*///指定offset消費/*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);*///從指定時間點開始消費/*List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);//從1小時前開始消費long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;Map<TopicPartition, Long> map = new HashMap<>();for (PartitionInfo par : topicPartitions) {map.put(new TopicPartition(topicName, par.partition()), fetchDataTime);}Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {TopicPartition key = entry.getKey();OffsetAndTimestamp value = https://www.520longzhigu.com/diannao/entry.getValue();if (key == null || value == null) continue;Long offset = value.offset();System.out.println("partition-" + key.partition() + "|offset-" + offset);System.out.println();//根據(jù)消費里的timestamp確定offsetif (value != null) {consumer.assign(Arrays.asList(key));consumer.seek(key, offset);}}*/while (true) {/** poll() API 是拉取消息的長輪詢*/ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = https://www.520longzhigu.com/diannao/%s%n", record.partition(),record.offset(), record.key(), record.value());}/*if (records.count() > 0) {// 手動同步提交offset，當(dāng)前線程會阻塞直到offset提交成功// 一般使用同步提交，因為提交之后一般也沒有什么邏輯代碼了consumer.commitSync();// 手動異步提交offset，當(dāng)前線程提交offset不會阻塞，可以繼續(xù)處理后面的程序邏輯consumer.commitAsync(new OffsetCommitCallback() {@Overridepublic void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {if (exception != null) {System.err.println("Commit failed for " + offsets);System.err.println("Commit failed exception: " + exception.getStackTrace());}}});}*/}}}Spring Boot整合Kafka引入spring boot kafka依賴:<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>

以上關(guān)于本文的內(nèi)容，僅作參考！溫馨提示：如遇健康、疾病相關(guān)的問題，請您及時就醫(yī)或請專業(yè)人士給予相關(guān)指導(dǎo)!

「愛刨根生活網(wǎng)」www.malaban59.cn小編還為您精選了以下內(nèi)容，希望對您有所幫助：

kafka簡單的入門案例 kafka應(yīng)用實例( 五 )