[大数据]springboot 整合kafka

1.kafka集群配置的话需要在本地host文件里配置
如下：

192.168.5.11 kafka1
192.168.5.12 kafka2
192.168.5.13 kafka3
192.168.5.14 kafka4
192.168.5.15 kafka5
192.168.5.16 kafka6

2.先引依赖

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

这是对应的版本，千万不要引错版本，负责回报找不到类这种异常
3.在application.yml中进行配置kafka

kafka:
    bootstrap-servers: 192.168.5.11:9092,192.168.5.12:9092,192.168.5.13:9092,192.168.5.14:9092,192.168.5.15:9092
    producer:
      retries: 0
      batch-size: 16384
      buffer-memory: 33554432
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        linger.ms: 1

    consumer:
      enable-auto-commit: false
      auto-commit-interval: 100ms
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        session.timeout.ms: 15000
      group-id: test-group-id

    listener:
      # 在侦听器容器中运行的线程数。
      concurrency: 5
      #listner负责ack，每调用一次，就立即commit
      ack-mode: manual_immediate
      missing-topics-fatal: false

4.可配置kafkaconfig也可让springboot自动装配

@Configuration
@EnableKafka
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;
    @Value("${spring.kafka.producer.retries}")
    private int retries;
    @Value("${spring.kafka.producer.batch-size}")
    private int batchSize;
    @Value("${spring.kafka.producer.properties.linger.ms}")
    private int linger;
    @Value("${spring.kafka.producer.buffer-memory}")
    private int bufferMemory;

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfig());
    }

    @Bean
    public Map<String, Object> producerConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<String, Object>(producerFactory());
    }

    @Bean//通过bean创建(bean的名字为initialTopic)
    public NewTopic initialTopic() {
        return new NewTopic("test",3, (short) 1 );
    }

    @Bean //创建一个kafka管理类，相当于rabbitMQ的管理类rabbitAdmin,没有此bean无法自定义的使用adminClient创建topic
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> props = new HashMap<>();
        //配置Kafka实例的连接地址
        //kafka的地址，不是zookeeper
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        KafkaAdmin admin = new KafkaAdmin(props);
        return admin;
    }

    @Bean  //kafka客户端，在spring中创建这个bean之后可以注入并且创建topic,用于集群环境，创建对个副本
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin().getConfig());
    }

}

5.注入kafkaTemplate,生产者发送消息

 @Slf4j
public class TestHandlerKafka {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    private static final String TOPIC_NOTIFY = "test";

    @Autowired
    private KafkaSendResultHandler producerListener;


    public ResponseStatusList producer() {
        ResponseStatusList responseStatusList = new ResponseStatusList();

        try {
            //发送消息前配置回调
            kafkaTemplate.setProducerListener(producerListener);
            String key = "test";
            ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send(TOPIC_NOTIFY, key, JSON.toJSONString(entity));
            SendResult<String, Object> stringObjectSendResult = send.get();
            log.info("发送消息成功,", stringObjectSendResult);
            responseStatusList.setLocalDate(new Date());
            responseStatusList.setStatusCode("0");
            responseStatusList.setStatusString("正常");
            return responseStatusList;
        } catch (Exception e) {
            log.error(Status.KAFKA_SEND_ERROR.getMsg(), e);
            responseStatusList.setLocalDate(new Date());
            responseStatusList.setStatusCode("1");
            responseStatusList.setStatusString("其他未知错误");
            return responseStatusList;
        }
    }
}


	

6.可进行简单的消费

@Component
public class KafkaConsumer {


    private static final String TOPIC_NOTIFY = "test";

    // 消费监听
    @KafkaListener(topics = {TOPIC_NOTIFY})
    public void onMessage1(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.err.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
    }

}