Kafka的java API交互方式、Topic串联、Kafka版本兼容 06

╰+攻爆jí腚メ 2022-04-24 01:08 184阅读 0赞

# 1. 普通Topic生成消息&消费消息 #

*  Java工程-maven，依赖

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

*  生产者生产消息

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    /** * */
    public class KafkaProducerOneTopic { 
        public static void main(String[] args) throws InterruptedException { 
            //1. 连接集群
            Properties props = new Properties();
            props.put("bootstrap.servers", "node01: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<String, String>(props);
            for (int i = 20; i < 30; i++) { 
                //指定key,不指定parition
                producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), "真正的生成消息->"+Integer.toString(i)));
                Thread.sleep(1000);
            }
    
            producer.close();
        }
    }

*  消费者消费消息

import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    /** * */
    public class KafkaConsumerOneTopic { 
        public static void main(String[] args) { 
            //1. 获取连接
            Properties props = new Properties();
            props.put("bootstrap.servers", "node01:9092");
            props.put("group.id", "test");
            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<String, String>(props);
            consumer.subscribe(Arrays.asList("my-topic"));
    
            //真正的消费消息
            while (true) { 
                ConsumerRecords<String, String> records = consumer.poll(100);
    
                for (ConsumerRecord<String, String> record : records) { 
    
                    System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
                    System.out.println();
                }
    
            }
        }
    }

# 2. Topic串联 #

**需求：从一个kafka的topic中消费数据，然后放到另一个topic当中**

## 2.1 图解 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]

## 2.2 代码详解 ##

*  生产者A像mytopicA中生产数据

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    /** * 消息生产者 * 连接kafka集群 * 确定主题 * 生产者入口 */
    public class ProducerA { 
    
        public static void main(String[] args) throws InterruptedException { 
            //1. 连接集群
            Properties props = new Properties();
            props.put("bootstrap.servers", "node01: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");
    
            //2. 生产者入口
            Producer<String, String> producer = new KafkaProducer<String, String>(props);
            for (int i = 0; i < 5; i++) { 
                producer.send(new ProducerRecord<String, String>("mytopicA", Integer.toString(i), "真正的生产消息->"+Integer.toString(i)));
                Thread.sleep(2000);
            }
            producer.close();
        }
    }

*  创建消费者A消费mytopicA中的消息，同时生产者B生产消息到mytopicB中

import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.io.IOException;
    import java.util.Arrays;
    import java.util.Properties;
    
    /** * 1. 消费mytopicA中的消息 * 2. 消费消息后调用生产消息到mytopicB中的方法，将消息生产到mytopicB中 */
    public class ConsumerAToTopicB { 
    
        public static void main(String[] args) throws InterruptedException, IOException { 
            //1. 获取连接
            Properties props = new Properties();
            props.put("bootstrap.servers", "node01:9092");
            props.put("group.id", "test");
            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");
    
            //2. 创建消费入口
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
            consumer.subscribe(Arrays.asList("mytopicA"));
    
            String ms = "";
    
            while (true) { 
                //3. 真正的消费消息
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) { 
                    //消费消息的时候, 生产消息, 生产到另一主题, MytopicB中
                    ms = record.value();
                    System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
                    System.out.println();
                    //调用生产消息的方法，生产消息
                    sendMessage(ms);
    
                }
                
            }
    
        }
    
        static Properties props = null;
        static Producer<String, String> producerB = null;
    
        static { 
            props = new Properties();
            props.put("bootstrap.servers", "node01: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");
        }
        //消息消费定义一个方法,
        public static void sendMessage(String ms) throws InterruptedException { 
            //2. 生产者入口
            producer = new KafkaProducer<String, String>(props);
            producer.send(new ProducerRecord<String, String>("mytopicB", ms));
            producer.close();
        }
    }

*  创建消费者B消费mytopicB中的消息

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    /** * 消息生产者 * 连接kafka集群 * 确定主题 * 生产者入口 */
    public class ProducerB { 
    
        public static void main(String[] args) throws InterruptedException { 
            //1. 连接集群
            Properties props = new Properties();
            props.put("bootstrap.servers", "node01: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");
    
            //2. 生产者入口
            Producer<String, String> producer = new KafkaProducer<String, String>(props);
            for (int i = 0; i < 5; i++) { 
                producer.send(new ProducerRecord<String, String>("mytopicB", Integer.toString(i), "真正的生产消息->"+Integer.toString(i)));
                Thread.sleep(2000);
            }
            producer.close();
        }
    }

# 3. Kafka版本兼容 #

**需求：生产者使用的版本是0.8.0.0，消费者使用的版本是0.10.0.0，但是生产者和消费者必须使用同一个版本，如何解决？**

## 3.1 图解 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]

## 3.3 伪代码 ##

*  创建两个module， 其中一个module的maven引入kafak0.8.0.0版本，另一个maven引入kafka0.10.0.0版本。
 *  和module平级的src中创建一个context，保存旧消息，当做消息过渡这。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]: /images/20220221/4e10784fb9124ff7ab4c3dccaa59f748.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]: /images/20220221/f061feb907284061b5d15526ef4fe7ff.png