第三章 Kafka集成 SpringBoot
SpringBoot 是一个在 JavaEE 开发中非常常用的组件。可以用于 Kafka 的生产者，也可以 用于 SpringBoot 的消费者。
在初始化springboot环境的时候要勾选kafka依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>3.1 SpringBoot 生产者
（1）修改 SpringBoot 核心配置文件 application.propeties, 添加生产者相关信息
server:
port: 8080
spring:
kafka:
bootstrap-servers: hadoop102:9092,hadoop103:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
group-id: test
（2）创建 controller 从浏览器接收数据, 并写入指定的 topic
package cn.jxust.springbootkafka.Controller;/*
*
* @author pengjx
*
* */
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class ProducerController {
@Autowired
private KafkaTemplate<String,String> stringKafkaTemplate;
@RequestMapping("/atguigu")
public String data(String msg){
stringKafkaTemplate.send("first",msg);
return "ok";
}
}
（3）在浏览器中给/atguigu 接口发送数据 http://localhost:8080/atguigu?msg=hello
3.2 SpringBoot 消费者
（1）修改 SpringBoot 核心配置文件 application.propeties
server:
port: 8080
spring:
kafka:
bootstrap-servers: hadoop102:9092,hadoop103:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
group-id: test
（2）创建类消费 Kafka 中指定 topic 的数据
package cn.jxust.springbootkafka.Consumer;/*
*
* @author pengjx
*
* */
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
@Configuration
public class KafkaConsumer {
@KafkaListener(topics = "first")
public void getData(String msg){
System.out.println("消息是："+msg);
}
}
3.3 工具类
KafkaProducer
package cn.jxust.springbootkafka.utils;/*
*
* @author pengjx
*
* */
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
@Component
@Slf4j
public class KafkaProducer {
@Autowired
private KafkaTemplate<String,Object> kafkaTemplate;
private static final String TOPIC="first";
public void send(Object object){
String jsonStr = JSONUtil.toJsonStr(object);
log.info("准备发送消息为：{}", jsonStr);
ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC, jsonStr);
future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable ex) {
//发送失败的处理
log.info(TOPIC + " - 生产者 发送消息失败：" + ex.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> result) {
//成功的处理
log.info(TOPIC + " - 生产者 发送消息成功：" + result.toString());
}
});
}
}
KafkaConsumer
package cn.jxust.springbootkafka.utils;/*
*
* @author pengjx
*
* */
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.Optional;
@Component
@Slf4j
public class KafkaConsumer {
@KafkaListener(topics = "first")
public void topicTest(ConsumerRecord<?,?> record){
Optional<?> message = Optional.ofNullable(record.value());
if(message.isPresent()){
Object object = message.get();
log.info("topic.group1 消费了： Topic:" + "first" + ",Message:" + object);
}
}
}
