springboot 整合 kafka

「爱情、让人受尽委屈。」 2022-05-09 14:12 215阅读 0赞

## maven依赖 ##

springboot 依赖省略

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

## 代码部分 ##

### 配置部分 ###

spring:
      kafka:
        bootstrap-servers: 127.0.0.1:9092
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:
          # 应该可以消费者类指定
          group-id: 9 
          enable-auto-commit: true
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

### 生产者 ###

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.scheduling.annotation.EnableScheduling;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    /** * @author lf * @描述 生产者 * @date 2018-09-07 */
    @EnableScheduling
    @Component
    public class Producer { 
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        // 3秒发送一个数据
        @Scheduled(fixedDelay=3000)
        public void send() { 
            Date date = new Date();
            SimpleDateFormat simpleDateFormat=new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
            String format = simpleDateFormat.format(date);
            // topic 名字 test
            kafkaTemplate.send("test",format+" boot");
        }
    
    }

### 消费者 ###

消费者只是简单的试用了几个配置

import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.kafka.annotation.PartitionOffset;
    import org.springframework.kafka.annotation.TopicPartition;
    import org.springframework.kafka.support.Acknowledgment;
    import org.springframework.stereotype.Component;
    
    /** * @author lf * @描述 消费者 * @date 2018-09-07 */
    @Component
    public class Consumer { 
    
        @KafkaListener(id = "Consumer1",//管理此端点的容器的唯一标识符
                // groupId = "",仅为此侦听器使用此值覆盖使用者工厂的属性。
                // 消费topic属性配置
                //配置topic和分区：监听两个topic，分别为test1、test2，
                // test1只接收分区0，1的消息，
                //test2接收分区0和分区2的消息，但是分区1的消费者初始位置(offset)为8
                topicPartitions =
                        { 
                                @TopicPartition(topic = "test1", partitions = {  "0", "1" }),
                                @TopicPartition(topic = "test2", partitions = "0",
                                        partitionOffsets = @PartitionOffset(partition = "2", initialOffset = "7"))
                        })
        public void listen (ConsumerRecord<?, ?> record, Acknowledgment ack) throws Exception { 
            System.out.printf("topic = %s, offset = %d, value = %s partition=%s\n", record.topic(), record.offset(), record.value(),record.partition());
            // 开启即手动提交
            //ack.acknowledge(); 
        }
    }

属性配置参考官网https://docs.spring.io/spring-kafka/api/org/springframework/kafka/annotation/KafkaListener.html