springboot 整合 kafka-蒲公英云

springboot 整合 kafka

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(); 
    }
}