kafka_2.13-2.7.0安装

#broker的全局唯一编号，不能重复

broker.id=1

#用来监听链接的端口，producer或consumer将在此端口建立连接

port=9092

#处理网络请求的线程数量

num.network.threads=3

#用来处理磁盘IO的线程数量

num.io.threads=8

#topic在当前broker上的分片个数

num.partitions=3

#用来恢复和清理data下数据的线程数量

num.recovery.threads.per.data.dir=1

#发送套接字的缓冲区大小

socket.send.buffer.bytes=102400

#接受套接字的缓冲区大小

socket.receive.buffer.bytes=102400

#请求套接字的缓冲区大小

socket.request.max.bytes=104857600

#kafka消息存放的路径

log.dirs=/usr/local/kafka/logs

#partion buffer中，消息的条数达到阈值，将触发flush到磁盘

log.flush.interval.messages=10000

#消息buffer的时间，达到阈值，将触发flush到磁盘

log.flush.interval.ms=3000

#segment文件保留的最长时间，超时将被删除

log.retention.hours=168

#滚动生成新的segment文件的最大时间

log.roll.hours=168

#日志文件中每个segment的大小，默认为1G

log.segment.bytes=1073741824

#周期性检查文件大小的时间

log.retention.check.interval.ms=300000

#日志清理是否打开

log.cleaner.enable=true

#broker需要使用zookeeper保存meta数据

zookeeper.connect=192.168.31.101:2181,192.168.31.102:2181,192.168.31.103:2181

#zookeeper链接超时时间

zookeeper.connection.timeout.ms=6000

#删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除

delete.topic.enable=true

#此处的host.name为本机IP(重要),如果不改,则客户端会抛出错误!

host.name=192.168.31.101

advertised.host.name=192.168.31.101

advertised.listeners=PLAINTEXT://192.168.31.101:9092

default.replication.factor=3

auto.create.topics.enable=true

message.max.byte=5242880

replica.fetch.max.bytes=5242880

metadata.broker.list=192.168.31.101:9092,192.168.31.102:9092,192.168.31.103:9092

zookeeper.connect=192.168.31.101:2181,192.168.31.102:2181,192.168.31.103:2181

public class MyProducer {

    private static KafkaProducer<String, String> producer;

    //初始化

    static {

        Properties properties = new Properties();

        //kafka启动，生产者建立连接broker的地址

        properties.put("bootstrap.servers", "127.0.0.1:9092");

        //kafka序列化方式

        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //自定义分区分配器

        properties.put("partitioner.class", "com.imooc.kafka.CustomPartitioner");

        producer = new KafkaProducer<>(properties);

    }

    /**

     * 创建topic：.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181

     * --replication-factor 1 --partitions 1 --topic kafka-study

     * 创建消费者：.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092

     * --topic imooc-kafka-study --from-beginning

     */

    //发送消息，发送完后不做处理

    private static void sendMessageForgetResult() {

        ProducerRecord<String, String> record = new ProducerRecord<>("kafka-study", "name", "ForgetResult");

        producer.send(record);

        producer.close();

    }

    //发送同步消息，获取发送的消息

    private static void sendMessageSync() throws Exception {

        ProducerRecord<String, String> record = new ProducerRecord<>("kafka-study", "name", "sync");

        RecordMetadata result = producer.send(record).get();

        System.out.println(result.topic());//imooc-kafka-study

        System.out.println(result.partition());//分区为0

        System.out.println(result.offset());//已发送一条消息，此时偏移量+1

        producer.close();

    }

    /**

     * 创建topic：.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181

     * --replication-factor 1 --partitions 3 --topic kafka-study-x

     * 创建消费者：.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092

     * --topic kafka-study-x --from-beginning

     */

    private static void sendMessageCallback() {

        ProducerRecord<String, String> record = new ProducerRecord<>("kafka-study-x", "name", "callback");

        producer.send(record, new MyProducerCallback());

        //发送多条消息

        record = new ProducerRecord<>("kafka-study-x", "name-x", "callback");

        producer.send(record, new MyProducerCallback());

        producer.close();

    }

    //发送异步消息

    //场景：每条消息发送有延迟，多条消息发送，无需同步等待，可以执行其他操作，程序会自动异步调用

    private static class MyProducerCallback implements Callback {

        @Override

        public void onCompletion(RecordMetadata recordMetadata, Exception e) {

            if (e != null) {

                e.printStackTrace();

                return;

            }

            System.out.println("*** MyProducerCallback ***");

            System.out.println(recordMetadata.topic());

            System.out.println(recordMetadata.partition());

            System.out.println(recordMetadata.offset());

        }

    }

    public static void main(String[] args) throws Exception {

        //sendMessageForgetResult();

        //sendMessageSync();

        sendMessageCallback();

    }

}

public class CustomPartitioner implements Partitioner {

    @Override

    public int partition(String topic, Object key, byte[] keyBytes,

                         Object value, byte[] valueBytes, Cluster cluster) {

        //获取topic所有分区

        List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);

        int numPartitions = partitionInfos.size();

        //消息必须有key

        if (null == keyBytes || !(key instanceof String)) {

            throw new InvalidRecordException("kafka message must have key");

        }

        //如果只有一个分区，即0号分区

        if (numPartitions == 1) {return 0;}

        //如果key为name，发送至最后一个分区

        if (key.equals("name")) {return numPartitions - 1;}

        return Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1);

    }

    @Override

    public void close() {}

    @Override

    public void configure(Map<String, ?> map) {}

}

public class MyConsumer {

    private static KafkaConsumer<String, String> consumer;

    private static Properties properties;

    //初始化

    static {

        properties = new Properties();

        //建立连接broker的地址

        properties.put("bootstrap.servers", "127.0.0.1:9092");

        //kafka反序列化

        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        //指定消费者组

        properties.put("group.id", "KafkaStudy");

    }

    //自动提交位移：由consume自动管理提交

    private static void generalConsumeMessageAutoCommit() {

        //配置

        properties.put("enable.auto.commit", true);

        consumer = new KafkaConsumer<>(properties);

        //指定topic

        consumer.subscribe(Collections.singleton("kafka-study-x"));

        try {

            while (true) {

                boolean flag = true;

                //拉取信息，超时时间100ms

                ConsumerRecords<String, String> records = consumer.poll(100);

                //遍历打印消息

                for (ConsumerRecord<String, String> record : records) {

                    System.out.println(String.format(

                            "topic = %s, partition = %s, key = %s, value = %s",

                            record.topic(), record.partition(), record.key(), record.value()

                    ));

                    //消息发送完成

                    if (record.value().equals("done")) { flag = false; }

                }

                if (!flag) { break; }

            }

        } finally {

            consumer.close();

        }

    }

    //手动同步提交当前位移，根据需求提交，但容易发送阻塞，提交失败会进行重试直到抛出异常

    private static void generalConsumeMessageSyncCommit() {

        properties.put("auto.commit.offset", false);

        consumer = new KafkaConsumer<>(properties);

        consumer.subscribe(Collections.singletonList("kafka-study-x"));

        while (true) {

            boolean flag = true;

            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records) {

                System.out.println(String.format(

                        "topic = %s, partition = %s, key = %s, value = %s",

                        record.topic(), record.partition(), record.key(), record.value()

                ));

                if (record.value().equals("done")) { flag = false; }

            }

            try {

                //手动同步提交

                consumer.commitSync();

            } catch (CommitFailedException ex) {

                System.out.println("commit failed error: " + ex.getMessage());

            }

            if (!flag) { break; }

        }

    }

    //手动异步提交当前位移，提交速度快，但失败不会记录

    private static void generalConsumeMessageAsyncCommit() {

        properties.put("auto.commit.offset", false);

        consumer = new KafkaConsumer<>(properties);

        consumer.subscribe(Collections.singletonList("kafka-study-x"));

        while (true) {

            boolean flag = true;

            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records) {

                System.out.println(String.format(

                        "topic = %s, partition = %s, key = %s, value = %s",

                        record.topic(), record.partition(), record.key(), record.value()

                ));

                if (record.value().equals("done")) { flag = false; }

            }

            //手动异步提交

            consumer.commitAsync();

            if (!flag) { break; }

        }

    }

    //手动异步提交当前位移带回调

    private static void generalConsumeMessageAsyncCommitWithCallback() {

        properties.put("auto.commit.offset", false);

        consumer = new KafkaConsumer<>(properties);

        consumer.subscribe(Collections.singletonList("kafka-study-x"));

        while (true) {

            boolean flag = true;

            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records) {

                System.out.println(String.format(

                        "topic = %s, partition = %s, key = %s, value = %s",

                        record.topic(), record.partition(), record.key(), record.value()

                ));

                if (record.value().equals("done")) { flag = false; }

            }

            //使用java8函数式编程

            consumer.commitAsync((map, e) -> {

                if (e != null) {

                    System.out.println("commit failed for offsets: " + e.getMessage());

                }

            });

            if (!flag) { break; }

        }

    }

    //混合同步与异步提交位移

    @SuppressWarnings("all")

    private static void mixSyncAndAsyncCommit() {

        properties.put("auto.commit.offset", false);

        consumer = new KafkaConsumer<>(properties);

        consumer.subscribe(Collections.singletonList("kafka-study-x"));

        try {

            while (true) {

                //boolean flag = true;

                ConsumerRecords<String, String> records = consumer.poll(100);

                for (ConsumerRecord<String, String> record : records) {

                    System.out.println(String.format(

                            "topic = %s, partition = %s, key = %s, " + "value = %s",

                            record.topic(), record.partition(),

                            record.key(), record.value()

                    ));

                    //if (record.value().equals("done")) { flag = false; }

                }

                //手动异步提交，保证性能

                consumer.commitAsync();

                //if (!flag) { break; }

            }

        } catch (Exception ex) {

            System.out.println("commit async error: " + ex.getMessage());

        } finally {

            try {

                //异步提交失败，再尝试手动同步提交

                consumer.commitSync();

            } finally {

                consumer.close();

            }

        }

    }

    public static void main(String[] args) {

        //自动提交位移

        generalConsumeMessageAutoCommit();

        //手动同步提交当前位移

        //generalConsumeMessageSyncCommit();

        //手动异步提交当前位移

        //generalConsumeMessageAsyncCommit();

        //手动异步提交当前位移带回调

        //generalConsumeMessageAsyncCommitWithCallback()

        //混合同步与异步提交位移

        //mixSyncAndAsyncCommit();

    }

}