---

• kafka 整合 springboot

• pom.xml

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.5.RELEASE</version>
</parent>
<dependencies>

      <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter</artifactId>
       </dependency>

       <dependency>
           <groupId>org.springframework.kafka</groupId>
           <artifactId>spring-kafka</artifactId>
       </dependency>
       <!--测试-->
       <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-test</artifactId>
           <scope>test</scope>
       </dependency>

</dependencies>

• application.yml

spring:
  kafka:
    bootstrap-servers: kafka_1:9092,kafka_2:9092,kafka_3:9092
    consumer:
      auto-commit-interval: 100
      auto-offset-reset: earliest
      enable-auto-commit: true
      group-id: group1
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        isolation:
          level: read_committed
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      acks: all
      batch-size: 16384
      buffer-memory: 33554432
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        enable:
          idempotence: true
      retries: 5
      transaction-id-prefix: transaction-id-
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

• KafkaBootApp

package cn.qww.boot;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.KafkaListeners;
import org.springframework.messaging.handler.annotation.SendTo;

import java.io.IOException;


@SpringBootApplication

@EnableKafka
public class KafkaBootApp {
    public static void main(String[] args) throws IOException {
        SpringApplication.run(KafkaBootApp.class, args);
    }


    @KafkaListeners(value = {@KafkaListener(topics = {"topic04"})})
    @SendTo(value = {"topic02"})
    public String listener(ConsumerRecord<?, ?> cr) {
        System.out.println("receive:" + cr);
        return cr.value() + " qww";
    }


}

• kafkaTemplate 使用，及 事务

package cn.qww.boot;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.transaction.annotation.Transactional;

@SpringBootTest(classes = {KafkaBootApp.class})
@RunWith(SpringRunner.class)
public class KafkaTemplateTests {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 两种方式，注掉 application.yml 中   transaction-id-prefix: transaction-id-
     * 或者添加 @Transactional 注释
     */
    @Test
    // @Transactional
    public void sendTest() {
        kafkaTemplate.send(new ProducerRecord<>("topic01", "kafkaTemplate_key", "kafkaTemplate_value"));
    }

    /**
     * 注掉 application.yml 中 需要开启：  transaction-id-prefix: transaction-id- 配置；
     * 就是说和上边的方法相互冲突
     */
    @Test
    public void testKafkaTemplateTx() {
        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, String, Object>() {
            @Override
            public Object doInOperations(KafkaOperations<String, String> kafkaOperations) {
                return kafkaOperations.send(new ProducerRecord<>("topic01", "OperationsCallback_key", "OperationsCallback_OperationsCallback"));
            }
        });
    }

}