Kakfa学习笔记（三）——Java API发送消费消息

小灰灰 2022-10-01 10:57 82阅读 0赞

上一篇[Kakfa学习笔记（二）——体验Kafka][Kakfa_Kafka]

这次我们用Java API来发送和消费消息

## 一对一的发送消费 ##

首先启动zk和三个broker

> bin/zookeeper-server-start.sh config/zookeeper.properties
    > bin/kafka-server-start.sh config/server.properties
    > bin/kafka-server-start.sh config/server-1.properties
    > bin/kafka-server-start.sh config/server-2.properties
    复制代码

创建chat主题，分区和副本都为3

> bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic chat
    复制代码

maven引入依赖

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.2.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.2.0</version>
    </dependency>
    复制代码

编写生产者代码，指定topic为chat，发送10条消息

public class Sender {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("acks", "all");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            Producer<String, String> producer = new KafkaProducer<>(props);
            for (int i = 0; i < 10; i++)
                producer.send(new ProducerRecord<String, String>("chat", ""+Integer.toString(i), Integer.toString(i)));
    
            producer.close();
    
        }
    }
    复制代码

编写消费者代码，很简单，把消费的内容打印出来

public class Receiver {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", "localhost:9092");
            props.setProperty("group.id", "chat-room-1");//消费组id
            props.setProperty("enable.auto.commit", "true");
            props.setProperty("auto.commit.interval.ms", "1000");
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Collections.singletonList("chat"));//chat主题
          
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records)
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
    复制代码

我们可以先后启动一个消费者和一个生产者，可以看到消费者输出10条数据

offset = 8, key = 0, value = 0
    offset = 9, key = 2, value = 2
    offset = 10, key = 3, value = 3
    offset = 11, key = 9, value = 9
    offset = 4, key = 4, value = 4
    offset = 5, key = 6, value = 6
    offset = 8, key = 1, value = 1
    offset = 9, key = 5, value = 5
    offset = 10, key = 7, value = 7
    offset = 11, key = 8, value = 8
    复制代码

可以看到消息并不是有序的，因为消息分发到三个分区，这个消息者消费了三个分区

## 多消费者消费 ##

因为我们这个topic有三个分区，我们可以先挂起三个消费者(同一个消费组)，再起一个生产者，这次可以看到消息是分摊在三个消费者，不会出现重复消息

consumer-0:
    offset = 4, key = 0, value = 0
    offset = 5, key = 2, value = 2
    offset = 6, key = 3, value = 3
    offset = 7, key = 9, value = 9
    
    consumer-1:
    offset = 4, key = 1, value = 1
    offset = 5, key = 5, value = 5
    offset = 6, key = 7, value = 7
    offset = 7, key = 8, value = 8
    
    consumer-2:
    offset = 2, key = 4, value = 4
    offset = 3, key = 6, value = 6
    复制代码

如果我们起四个消费者，看一下结果

consumer-0:
    offset = 16, key = 0, value = 0
    offset = 17, key = 2, value = 2
    offset = 18, key = 3, value = 3
    offset = 19, key = 9, value = 9
    
    consumer-1:
    
    consumer-2:
    offset = 16, key = 1, value = 1
    offset = 17, key = 5, value = 5
    offset = 18, key = 7, value = 7
    offset = 19, key = 8, value = 8
    
    consumer-3
    offset = 8, key = 4, value = 4
    offset = 9, key = 6, value = 6
    复制代码

会发现consumer-1是消费不到消息的，因为只有3个分区，所以最多只有3个消费者同时工作，多出来的一个不会消费

## 有序消息 ##

如果我们想消息有序被送达并消费，在发送时可以指定partition

producer.send(new ProducerRecord<String, String>("chat",1, Integer.toString(i), Integer.toString(i)));//第二个参数指定partition=1
    复制代码

我们起两个消费者

consumer-0:
    
    consumer-1:
    offset = 10, key = 0, value = 0
    offset = 11, key = 1, value = 1
    offset = 12, key = 2, value = 2
    offset = 13, key = 3, value = 3
    offset = 14, key = 4, value = 4
    offset = 15, key = 5, value = 5
    offset = 16, key = 6, value = 6
    offset = 17, key = 7, value = 7
    offset = 18, key = 8, value = 8
    offset = 19, key = 9, value = 9
    复制代码

结果只有一个消费者消费了消息，并且可以看到，消息是有序的

转载于:https://juejin.im/post/5cdd2e3b51882538d859ba93

[Kakfa_Kafka]: https://juejin.im/post/5cdc19a96fb9a0323120b3e8