kafka api使用

1.生產者api

引入依賴

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.11.0.0</version> //版本爲0.11.0.0
</dependency>

生產者的配置項都在ProducerConfig類中說明，每一項配置都有對應的doc說明。

生產者使用api 帶回調函數demo，還有阻塞方式運行，返回Future對象，經過future對象get()到返回的值。

public class CustomProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.152.3:9092,192.168.152.2:9092,192.168.152.1:9092");//
        props.put(ProducerConfig.ACKS_CONFIG, "all"); //leader 確認機制 0 1 all
        props.put(ProducerConfig.RETRIES_CONFIG, 1);//重試次數
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);//生產者批發送大小
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);//生產者達不到批發送大小，最短等待時間
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);//RecordAccumulator 緩衝區大小
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); //key的序列化器
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); //value的序列化器
        Producer<String, String> producer = new KafkaProducer<>(props); 
        //ProducerConfig對象傳參到KafkaProducer構造函數，生成producer對象
        for (int i = 0; i < 10; i++) {
            //producer.send(),消息封裝成ProducerRecord對象
            //帶回調發送消息，若是發送失敗會自動重試
            producer.send(new ProducerRecord<String, String>("minerprofit", Integer.toString(i), Integer.toString(i)), (RecordMetadata metadata,Exception exception) -> {
                if (exception == null) {
                    System.out.println("success->" +
                            metadata.offset());
                } else {
                    exception.printStackTrace();
                }
            }
        });
    }
    producer.close(); //關閉消費者
}

2.消費者api

自動提交offset方式

public class CustomConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.152.3:9092,192.168.152.2:9092,192.168.152.1:9092");//
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "miner"); //消費者組
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); //開啓自動提交
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); //自動提交最短期
        //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");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("profit")); //消費者組訂閱的topic
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100); //拉取數據
            for (ConsumerRecord<String, String> record : records){
                System.out.printf("offset = %d, key = %s, value= %s%n", record.offset(), record.key(), record.value());
            }                
        }
    }
}

手動提交offset方式:

手動提交有兩種提交方式一種是同步提交，一種是異步提交。

public class CustomConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.152.3:9092,192.168.152.2:9092,192.168.152.1:9092");//
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "miner"); //消費者組
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); //開啓自動提交
        //props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); //自動提交最短期
        //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");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("profit")); //消費者組訂閱的topic
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100); //拉取數據
            for (ConsumerRecord<String, String> record : records){
                System.out.printf("offset = %d, key = %s, value= %s%n", record.offset(), record.key(), record.value());
            }
            consumer.commitSync(); //同步提交offset,會阻塞當前線程的運行
        }
    }
}

異步提交

consumer.commitAsync(new OffsetCommitCallback() {
    @Override
    public void onComplete(Map<TopicPartition,OffsetAndMetadata> offsets, Exception exception) {
        if (exception != null) {
            System.err.println("Commit failed：" +
                    offsets);
        }
    }
});

3.如何保證消息中間件冪等性

什麼是冪等性：

生產者生產的消息可以發送到消息中間件中，消息中間件不會重複接受也不會少接收；消費者進行消費消息，不會重複消費，也不會少消費。

kafka結合具體業務如何保證冪等性：

kafka 生產者確認acks使用all級別，生產者發送到kafka的消息只可能重複不可能丟失，保證at least once;消費者使用異步提交offset,在業務中將獲得的消息首先入數據庫，若是庫中已經存在了相同的消息，那麼若是獲得了新的相同的消息，那麼就能夠剔除重複的消息。