Kafka 的 Java Api 与 SpringBoot Api

Myth丶恋晨 2022-11-18 04:54 123阅读 0赞

# Java 操作 Kafka #

### 依赖 ###

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

### Producer ###

方式一：同步调用

public class MyKafKaProducer extends Thread{ 
        
        public void testProducer() { 
        	//================= 1、构造 KafkaProducer ======================
            // 构建连接配置
            Properties properties = new Properties();
           
            // 若要配多个服务器，用逗号隔开
            // 注：服务器要开放端口，若云服务器还要在server.properties配置内网IP和外网IP
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.129:9092");
            // 指定生产者 id
            properties.put(ProducerConfig.CLIENT_ID_CONFIG, "my-producer");
            // KEY_SERIALIZER_CLASS_CONFIG, key 的序列化方式，指分区 id，所以是 int 型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
                           IntegerSerializer.class.getName());
            // VALUE_SERIALIZER_CLASS_CONFIG，value 的序列化方式，就是消息内容的序列化方式，所以是 string
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
                           StringSerializer.class.getName());
    
            // 构造Client无非是：new 或 工厂模式
            KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(properties);
    		
    		//================= 2、发送消息 ======================
    		String topic = "test";
            String msg = "kafka practice msg: ";
            
            for(int i = 0 ; i < 20 ; i ++){ 
            	// 发送消息send()！！! 同步调用
    	        // Future.get()会阻塞，等待返回结果......
    	        RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, msg + i)).get();
    	      	// 等上面get到结果了，才能执行这里
    	      	// recordMetadata.offset()：消息偏移
    	      	// recordMetadata.partition()：消息发到哪个分区了
    	        System.out.println(recordMetadata.offset() + "->" + recordMetadata.partition() + 
    	                             "->" + recordMetadata.topic());
            }
        }
    }

方式二：异步调用

public void testProducer() {         
    	// ....
    	
        for(int i = 0 ; i < 20 ; i ++){ 
            // 异步调用 
            producer.send(new ProducerRecord<>(topic, msg + i), new Callback() {      
                 @Override 
                 // 回调函数
                 // 注：这里其实还能通过lamada用函数式接口(metadata, exception) -> {}
                 public void onCompletion(RecordMetadata recordMetadata, Exception e) {    
                 System.out.println("callback: "+ recordMetadata.offset()
                                       +"->"+recordMetadata.partition());                 
                 }               
            });               
            TimeUnit.SECONDS.sleep(2);         
      }

### Consumer ###

public class MykafkaConsumer extends Thread{ 
    
        KafkaConsumer<Integer, String> consumer;
        String topic;
    
        public MykafkaConsumer(String topic) { 
            // 构建连接配置，这里是ConsumerConfig
            Properties properties = new Properties();
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.129:9092");
            properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "my-consumer");
            // 反序列化，这里是Deserializer
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                           IntegerDeserializer.class.getName());
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                           StringDeserializer.class.getName());
                           
            /** * 以下是Producer没有的配置 */
            // 要加入的 group。
            // 一条消息只能被一个 group 里的一个 consumer 所消费，但是可以被不同 group 里的 consumer 所消费
            // 因为 kafka 保存了每个 group 对其订阅 topic 消费的偏移，所以每个 group 的消费互不干扰
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-gid"); // 要加入的group
            // 新group消费为位置。
            // 指的就是上面说的一条消息可以被不同 group 里的 consumer 所消费
            // 对应的还有 latest（从其他消费者最后消费的oﬀset处开始消费Topic的消息）
            properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 
            // 超时，心跳。
            properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); 
            // 自动提交（批量），即每隔 1000 ms 自动提交一次
            properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 自动提交（批量）
    
            consumer = new KafkaConsumer<>(properties);
            this.topic = topic;
        }
    
        public void run() { 
            // 死循环不断消费消息
            while (true) { 
                // 绑定订阅主题
                // 注：Collections.singleton返回一个单元素&不可修改Set集合，
                // 同样的还有singletonList，singletonMap
                consumer.subscribe(Collections.singleton(this.topic));
                // 接收消息 POLL()！！！
                ConsumerRecords<Integer, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
                consumerRecords.forEach(record -> System.out.println(record.key() + "->" + 
                                                     record.value() + "->" + record.offset()));
            }
        }
    
        public static void main(String[] args) { 
            // 拉取test主题的消息
            new MykafkaConsumer("test").start();
        }
    }

消息提交主要有以下两种方式：

*  自动提交：enable.auto.commit 可以配置自动提交时间
    
     *  可配合 auto.commit.interval.ms 设置自动提交频率
     *  可通过 enable.auto.commit=false关闭自动提交  
        重复消费问题：设置自动提交后，若在提交间隔Consumer挂了，就可能出现已经被消费了的消息由于未被确认而再次被消费；
 *  手动提交：即使存在重复提交，也可以有有效控制重复消息数
    
     *  异步提交：consumer.commitAsync()
     *  同步提交：consumer.commitSync()，存在阻塞，因此一般批量提交，以减少阻塞频率

# SpringBoot 整合 Kafka #

### 依赖 ###

<dependency>   
        <groupId>org.springframework.kafka</groupId>  
        <artifactId>spring-kafka</artifactId>  
        <version>2.2.0.RELEASE</version> 
    </dependency>

注意

*  SpringBoot 的版本和 Kafka-clinets 的版本有一个对照表格，如果没有按照正确的版本来引入，那么会存在版本问题导致ClassNotFound的问题，具体请参考 https://spring.io/projects/spring-kafka
    
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70][]

### Producer ###

application.properties

# kafka服务器
    spring.kafka.bootstrapservers=192.168.56.129:9092
    # 编码（Producer）
    spring.kafka.producer.keyserializer=org.apache.kafka.common.serialization.IntegerSerializer 
    spring.kafka.producer.valueserializer=org.apache.kafka.common.serialization.StringSerializer

发送消息

@Component
    public class MyKafKaProducer { 
    
        @Resource
        // Integer：消息key的类型，每条消息都需一个 key，它决定了该消息会去哪个 Partition
        // String：消息类型
        private KafkaTemplate<Integer, String> kafkaTemplate;  
    
        public void send() { 
        	// 通过kafka进行消息发送
        	// ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
            kafkaTemplate.send("test", 1, "msgData");
        }
    }

### Consumer ###

application.properties

# kafka服务器
    spring.kafka.bootstrapservers=192.168.56.129:9092
    # 解码（Consumer）
    spring.kafka.consumer.keydeserializer=org.apache.kafka.common.serialization.IntegerDeserializer
    spring.kafka.consumer.valuedeserializer=org.apache.kafka.common.serialization.StringDeserializer
    # 对Consumer的特殊配置
    spring.kafka.consumer.group-id=test-consumer-group 
    spring.kafka.consumer.auto-offset-reset=earliest 
    spring.kafka.consumer.enable-auto-commit=true

消费消息

@Component
    public class MyKafkaConsumer { 
    	
        // 订阅相应主题。接收到的消息会被注入ConsumerRecord 
        @KafkaListener(topics = { "test"})  
        public void listener(ConsumerRecord record) { 
        
            // Optional，jdk8新特性，用于null处理
            Optional msg = Optional.ofNullable(record.value()); 
            // isPresent判斷值是否存在
            if (msg.isPresent()) {   
                // 打印消息 
                System.out.println(msg.get()); // get，从optional中取出对应值
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70]: /images/20221022/7f022ef7e76f401dae1194b573829260.png