Java使用Kafka的方法

本篇內(nèi)容主要講解“Java使用Kafka的方法”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“Java使用Kafka的方法”吧!

1、maven依賴

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

2、Producer

2.1、producer發(fā)送消息

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
/**
 * @author Thomas
 * @Description:最簡單的kafka producer
 * @date 22:18 2019-7-5
 */
public class ProducerDemo {
    public static void main(String[] args) {
        Properties properties =new Properties();
        //zookeeper服務(wù)器集群地址，用逗號隔開
        properties.put("bootstrap.servers", "172.16.0.218:9092,172.16.0.219:9092,172.16.0.217:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //自定義producer攔截器
        properties.put("interceptor.classes", "com.lt.kafka.producer.MyProducerInterceptor");
        //自定義消息路由規(guī)則（消息發(fā)送到哪一個(gè)Partition中）
        //properties.put("partitioner.class", "com.lt.kafka.producer.MyPartition");

        Producer<string, string=""> producer = null;
        try {
            producer = new KafkaProducer<string, string="">(properties);
            for (int i = 20; i < 40; i++) {
                String msg = "This is Message:" + i;

                /**
                 * kafkaproducer中會同時(shí)調(diào)用自己的callback的onCompletion方法和producerIntercepter的onAcknowledgement方法。
                 * 關(guān)鍵源碼：Callback interceptCallback = this.interceptors == null
                 * callback : new InterceptorCallback<>(callback,
                 * this.interceptors, tp);
                 */
                producer.send(new ProducerRecord<string, string="">("leixiang", msg),new MyCallback());
            }
        } catch (Exception e) {
            e.printStackTrace();

        } finally {
            if(producer!=null)
                producer.close();
        }
    }
}

2.2、自定義producer攔截器

import java.util.Map;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
 * @author Thomas
 * @Description:自定義producer攔截器
 * @date 22:21 2019-7-5
 */
public class MyProducerInterceptor implements ProducerInterceptor<string,string>{
    /**
     * 打印配置相關(guān)信息
     */
    public void configure(Map<string,> configs) {
        // TODO Auto-generated method stub
        System.out.println(configs.toString());
    }

    /**
     * producer發(fā)送信息攔截方法
     */
    public ProducerRecord<string,string> onSend(ProducerRecord<string, string=""> record) {
        System.out.println("攔截處理前=============");
        String topic=record.topic();
        String value=record.value();
        System.out.println("攔截處理前的消息====："+value);
        ProducerRecord<string,string> record2=new ProducerRecord<string, string="">(topic, value+" (intercepted)");
        System.out.println("攔截處理后的消息："+record2.value());
        System.out.println("攔截處理后===============");
        return record2;
    }

    /**
     * 消息確認(rèn)回調(diào)函數(shù)，和callback的onCompletion方法相似。
     * 在kafkaProducer中，如果都設(shè)置，兩者都會調(diào)用。
     */
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (metadata != null)
            System.out.println("MyProducerInterceptor onAcknowledgement:RecordMetadata=" + metadata.toString());
        if (exception != null)
            exception.printStackTrace();
    }
    /**
     * interceptor關(guān)閉回調(diào)
     */
    public void close() {
        System.out.println("MyProducerInterceptor is closed!");
    }
}

2.3、自定義消息路由規(guī)則

自定義路由規(guī)則，可以根據(jù)自己的需要定義消息發(fā)送到哪個(gè)分區(qū)。自定義路由規(guī)則需要實(shí)現(xiàn)Partitioner。

import java.util.Map;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
/**
 * @author Thomas
 * @Description:
 * @date 22:24 2019-7-5
 */
public class MyPartition implements Partitioner {
    public void configure(Map<string,> arg0) {
        // TODO Auto-generated method stub
    }

    public void close() {
        // TODO Auto-generated method stub
    }

    public int partition(String arg0, Object arg1, byte[] arg2, Object arg3, byte[] arg4, Cluster arg5) {
        // TODO Auto-generated method stub
        return 0;
    }
}

3、Consumer

3.1、自動提交

import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
/**
 * @author Thomas
 * @Description:
 * @date 22:26 2019-7-5
 */
public class AutoCommitConsumerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "172.16.0.218:9092,172.16.0.219:9092,172.16.0.217:9092");
        props.put("group.id", "leixiang");
        props.put("enable.auto.commit", "true");
        //想要讀取之前的數(shù)據(jù)，必須加上
        //props.put("auto.offset.reset", "earliest");
        /* 自動確認(rèn)offset的時(shí)間間隔 */
        props.put("auto.commit.interval.ms", "1000");
        /*
         * 一旦consumer和kakfa集群建立連接，
         * consumer會以心跳的方式來高速集群自己還活著，
         * 如果session.timeout.ms 內(nèi)心跳未到達(dá)服務(wù)器，服務(wù)器認(rèn)為心跳丟失，會做rebalence
         */
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //配置自定義的攔截器，可以在攔截器中引入第三方插件實(shí)現(xiàn)日志記錄等功能。
        //props.put("interceptor.classes", "com.lt.kafka.consumer.MyConsumerInterceptor");

        @SuppressWarnings("resource")
        KafkaConsumer<string, string=""> consumer = new KafkaConsumer<string, string="">(props);
        try {
            /* 消費(fèi)者訂閱的topic, 可同時(shí)訂閱多個(gè) ，用逗號隔開*/
            consumer.subscribe(Arrays.asList("leixiang"));
            while (true) {
                //輪詢數(shù)據(jù)。如果緩沖區(qū)中沒有數(shù)據(jù)，輪詢等待的時(shí)間為毫秒。如果0，立即返回緩沖區(qū)中可用的任何記錄，則返回空
                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());
            }
        } catch (Exception e) {
            // TODO: handle exception
            e.printStackTrace();
        }
    }
}

3.2、手動提交

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
/**
 * @author Thomas
 * @Description:
 * @date 22:28 2019-7-5
 */
public class ManualCommitConsumerDemo {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "172.16.0.218:9092,172.16.0.219:9092,172.16.0.217:9092");
        props.put("group.id", "leixiang");
        props.put("enable.auto.commit", "false");//手動確認(rèn)
        /* 自動確認(rèn)offset的時(shí)間間隔 */
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "earliest");//想要讀取之前的數(shù)據(jù)，必須加上
        /*
         * 一旦consumer和kakfa集群建立連接，
         * consumer會以心跳的方式來高速集群自己還活著，
         * 如果session.timeout.ms 內(nèi)心跳未到達(dá)服務(wù)器，服務(wù)器認(rèn)為心跳丟失，會做rebalence
         */
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //配置自定義的攔截器，可以在攔截器中引入第三方插件實(shí)現(xiàn)日志記錄等功能。
        props.put("interceptor.classes", "com.lt.kafka.consumer.MyConsumerInterceptor");

        KafkaConsumer<string, string=""> consumer = new KafkaConsumer<string, string="">(props);
        /* 消費(fèi)者訂閱的topic, 可同時(shí)訂閱多個(gè) ，用逗號隔開*/
        consumer.subscribe(Arrays.asList("leixiang"));
        while (true) {
            ConsumerRecords<string, string=""> records = consumer.poll(100);
            for (ConsumerRecord<string, string=""> record : records) {
                //處理消息
                saveMessage(record);
                //手動提交，并且設(shè)置Offset提交回調(diào)方法
                //consumer.commitAsync(new MyOffsetCommitCallback());
                consumer.commitAsync();
            }
        }
    }

    public static void saveMessage(ConsumerRecord<string, string=""> record){
        System.out.printf("處理消息：offset = %d, key = %s, value = %s%n", record.offset(), record.key(),
                record.value());
    }
}

自定義Consumer攔截器

import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
/**
 * @author Thomas
 * @Description:
 * @date 22:29 2019-7-5
 */
public class MyConsumerInterceptor implements ConsumerInterceptor<string, string="">{
public void configure(Map<string,> configs) {
        System.out.println("MyConsumerInterceptor configs>>>"+configs.toString());
        }

public ConsumerRecords<string, string=""> onConsume(ConsumerRecords<string, string=""> records) {
        System.out.println("onConsume");
        return records;
        }

public void onCommit(Map<topicpartition, offsetandmetadata=""> offsets) {
        System.out.println("onCommit");
        }

public void close() {
        System.out.println("MyConsumerInterceptor is closed!");
        }
}

自定義Offset提交回調(diào)方法

import java.util.Map;

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
/**
 * @author Thomas
 * @Description:
 * @date 22:31 2019-7-5
 */
public class MyOffsetCommitCallback implements OffsetCommitCallback {

    public void onComplete(Map<topicpartition, offsetandmetadata=""> offsets, Exception exception) {
        if (offsets != null)
            System.out.println("offsets>>>" + offsets.toString());
        if (exception != null)
            exception.printStackTrace();
    }
}

到此，相信大家對“Java使用Kafka的方法”有了更深的了解，不妨來實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！