Kafka 的基本介绍与适用场景

    $ wget https://www.apache.org/dyn/closer.lua?path=/kafka/2.7.0/kafka_2.13-2.7.0.tgz -O kafka_2.13-2.7.0.tgz
    $ tar xzf kafka_2.13-2.7.0.tgz
    
      
    
    代码解读

    $ bin/zookeeper-server-start.sh config/zookeeper.properties
    
    
    代码解读

    $ bin/kafka-server-start.sh config/server.properties
    
    
    代码解读

    $ bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1
    
    
    代码解读

    public class SimpleProducer {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
    
        Producer<String, String> producer = new KafkaProducer<>(properties);
    
        for (int i = 0; i < 10; i++) {
            System.out.println("Producing message : " + i);
            ProducerRecord<String, String> record =
                    new ProducerRecord<>("test", Integer.toString(i),
                            "Hello World" + Integer.toString(i));
    
            RecordMetadata metadata = producer.send(record).get();
    
            System.out.println(metadata.topic());
            System.out.println(metadata.partition());
            System.out.println(metadata.offset());
        }
    
        producer.flush();
        producer.close();
    }
    }
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    public class SimpleConsumer {
    
    public static void main(String[] args) {
        // Create a Kafka consumer for topic `test` with group id `my-group`.
        KafkaConsumer<String, String> consumer =
                new KafkaConsumer<>(consumerProperties());
    
        // Subscribe the consumer to the topic `test`.
        consumer.subscribe(Collections.singletonList("test"));
    
        while (true) {
            // Poll records from Kafka until some data is available or timeout reached.
            final ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
            if (!records.isEmpty()) {
                System.out.printf("%d records received.\n", records.count());
    
                // Process each record.
                for (final ConsumerRecord<String, String> record : records) {
                    System.out.printf("Received record: %s\n", record);
    
                    // Commit offset so that next time we will start from where we left off last time.
                    try {
                        consumer.commitAsync();
                    } catch (CommitFailedException e) {
                        log.error("Unable to commit offset for partition {}.",
                                e.partition(), e);
                    }
                }
            } else {
                // There are no more records in this poll() operation, so let's wait and see if there is any change later on.
                Thread.sleep(100);
            }
        } finally {
            // Close down the Kafka consumer gracefully.
            consumer.close();
        }
    }
    
    private static Map<String, Object> consumerProperties() {
        return Collections.<String, Object>singletonMap(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
               .put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
               .put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
               .put(ConsumerConfig.GROUP_ID_CONFIG, "my-group")
               .put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    }
    }
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    $ bin/kafka-server-start.sh config/server.properties
    
    
    代码解读

    $ java -cp target/simple-kafka-client-example-0.1.jar com.github.charlemaznable.examples.SimpleProducer
    Producing message : 0
    Producing message : 1
    Producing message : 2
    Producing message : 3
    Producing message : 4
    Producing message : 5
    Producing message : 6
    Producing message : 7
    Producing message : 8
    Producing message : 9
    
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    $ java -cp target/simple-kafka-client-example-0.1.jar com.github.charlemaznable.examples.SimpleConsumer
    Received record: ConsumerRecord(topic=test, partition=0, leaderEpoch=null, offset=0, CreateTime=1625950445998, serializedKeySize=-1, serializedValueSize=12, key=null, value=Hello World0)
    Received record: ConsumerRecord(topic=test, partition=0, leaderEpoch=null, offset=1, CreateTime=1625950445999, serializedKeySize=-1, serializedValueSize=12, key=null, value=Hello World1)
    Received record: ConsumerRecord(topic=test, partition=0, leaderEpoch=null, offset=2, CreateTime=1625950446000, serializedKeySize=-1, serializedValueSize=12, key=null, value=Hello World2)
    ......
    
      
      
      
      
    
    代码解读