05_kafka-整合springboot

---

• 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:9092consumer:auto-commit-interval: 100auto-offset-reset: earliestenable-auto-commit: truegroup-id: group1key-deserializer: org.apache.kafka.common.serialization.StringDeserializerproperties:isolation:level: read_committedvalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerproducer:acks: allbatch-size: 16384buffer-memory: 33554432key-serializer: org.apache.kafka.common.serialization.StringSerializerproperties:enable:idempotence: trueretries: 5transaction-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 {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;/*** 两种方式，注掉 application.yml 中   transaction-id-prefix: transaction-id-* 或者添加 @Transactional 注释*/@Test// @Transactionalpublic void sendTest() {kafkaTemplate.send(new ProducerRecord<>("topic01", "kafkaTemplate_key", "kafkaTemplate_value"));}/*** 注掉 application.yml 中 需要开启：  transaction-id-prefix: transaction-id- 配置；* 就是说和上边的方法相互冲突*/@Testpublic void testKafkaTemplateTx() {kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, String, Object>() {@Overridepublic Object doInOperations(KafkaOperations<String, String> kafkaOperations) {return kafkaOperations.send(new ProducerRecord<>("topic01", "OperationsCallback_key", "OperationsCallback_OperationsCallback"));}});}}