kafka 0.10.0.0 版本

偏执的太偏执、 2022-09-24 09:24 96阅读 0赞

### kafka 0.10.0.0 版本 ###

#### 一、安装kafka ####

1）下载路径：[http://apache.fayea.com/kafka/0.10.0.0/kafka\_2.11-0.10.0.0.tgz][http_apache.fayea.com_kafka_0.10.0.0_kafka_2.11-0.10.0.0.tgz]

2）解压安装包

tar -xzf kafka_2.11-0.10.0.0.tgz
    cd kafka_2.11-0.10.0.0

#### 二、启动zookeeper ####

运行kafka需要使用Zookeeper，所有你需要先启动一个Zookeeper服务器，如果你没有Zookeeper，你可以使用kafka自带打包和配置好的Zookeeper

bin/zookeeper-server-start.sh config/zookeeper.properties

#### 三、启动kafka ####

bin/kafka-server-start.sh config/server.properties

#### 四、编写produce生成者 ####

pom.xml

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

produce.java

package kafka.produce;
    
    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;
    
    public class Produce {
    
        public static void main(String[] args) {
            System.out.println("begin produce");
            connectionKafka();
            System.out.println("finish produce");
        }
    
        public static void connectionKafka() {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("acks", "all");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("linger.ms", 1);
            props.put("buffer.memory", 33554432);
            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>("my-topic", Integer.toString(i), Integer.toString(i))); 
            }
            producer.close();
        }
    
    }

#### 五、编写consumer消费者 ####

pom.xml

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

consumer.java

package kafka.consumer;
    
    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;
    
    public class Consumer {
    
        public static void main(String[] args) {
            System.out.println("begin consumer");
            connectionKafka();
            System.out.println("finish consumer");
        }
    
        @SuppressWarnings("resource")
        public static void connectionKafka() {
    
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("group.id", "testConsumer");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            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");
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList("my-topic","test-topic"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                try {
                    Thread.sleep(20000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
                }
            }
        }
    }

#### 六、注意事项 ####

客户端pom.xml配置的kafka-clients版本一定要与kafka版本（0.10.0.0）一致，否则会报错

1) 错误一：

Error reading field 'topic_metadata': Error reading array of size 1139567, only 45 bytes available

2) 错误二：

Exception in thread "main" org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'brokers': Error reading field 'host': Error reading string of length 12601, only 210 bytes available
        at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:73)
        at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:380)
        at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:449)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:269)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:183)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:973)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:937)
        at kafka.consumer.Consumer.connectionKafka(Consumer.java:32)
        at kafka.consumer.Consumer.main(Consumer.java:14)

#### 七、shell脚本 ####

shell启动kafka生成者客户端

./kafka-console-producer.sh --broker-list localhost:9092 --topic zkTopic

shell启动kafka消费者客户端

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic zkTopic --from-beginning --zookeeper localhost:2181

shell查看topic分区信息

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic zkTopic1

运行结果

Topic:zkTopic1  PartitionCount:4    ReplicationFactor:1 Configs:
        Topic: zkTopic1 Partition: 0    Leader: 0   Replicas: 0 Isr: 0
        Topic: zkTopic1 Partition: 1    Leader: 0   Replicas: 0 Isr: 0
        Topic: zkTopic1 Partition: 2    Leader: 0   Replicas: 0 Isr: 0
        Topic: zkTopic1 Partition: 3    Leader: 0   Replicas: 0 Isr: 0

#### 八、参考地址 ####

1）官方文档：[http://kafka.apache.org/documentation.html\#producerconfigs][http_kafka.apache.org_documentation.html_producerconfigs]  
2）中文地址：[http://orchome.com/295][http_orchome.com_295]

[http_apache.fayea.com_kafka_0.10.0.0_kafka_2.11-0.10.0.0.tgz]: http://apache.fayea.com/kafka/0.10.0.0/kafka_2.11-0.10.0.0.tgz
[http_kafka.apache.org_documentation.html_producerconfigs]: http://kafka.apache.org/documentation.html#producerconfigs
[http_orchome.com_295]: http://orchome.com/295