Kafka在Java项目中的应用
Docker 安装Kafka
一.首先需要安装docker,可看这篇文章安装docker
二.拉取zookeeper和KafKa镜像
docker pull wurstmeister/zookeeper
docker pull wurstmeister/kafka
Kafka组件需要向zookeeper进行注册,所以也需要安装zookeeper
三.启动zookeeper、kafka组件
docker run -d --name zookeeper -p 2181:2181 wurstmeister/zookeeper
docker run -d --name kafka --publish 9092:9092 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 --env KAFKA_ADVERTISED_HOST_NAME=localhost --env KAFKA_ADVERTISED_PORT=9092 wurstmeister/kafka
启动成功界面如下,status即为running(运行中)
四.创建Springboot项目
4.1 添加依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</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>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
4.2 application.yml文件
server:
port: 9090
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
# 配置消费者消息offset是否自动重置(消费者重连会能够接收最开始的消息)
auto-offset-reset: earliest
producer:
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
retries: 3 # 重试次数
kafka:
topic:
my-topic: my-topic
my-topic2: my-topic2
4.3 创建实体类Book
public class Book {
private Long id;
private String name;
public Book() {
}
public Book(Long id, String name) {
this.id = id;
this.name = name;
}
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "Book{" +
"id=" + id +
", name='" + name + '\'' +
'}';
}
}
4.4 配置KafKa信息
@Configuration
public class KafkaConfig {
@Value("${kafka.topic.my-topic}")
String myTopic;
@Value("${kafka.topic.my-topic2}")
String myTopic2;
/**
* JSON消息转换器
*/
@Bean
public RecordMessageConverter jsonConverter() {
return new StringJsonMessageConverter();
}
/**
* 通过注入一个 NewTopic 类型的 Bean 来创建 topic，如果 topic 已存在，则会忽略。
*/
@Bean
public NewTopic myTopic() {
return new NewTopic(myTopic, 2, (short) 1);
}
@Bean
public NewTopic myTopic2() {
return new NewTopic(myTopic2, 1, (short) 1);
}
}
4.5 controller代码
@RestController
@RequestMapping(value = "/book")
public class BookController {
@Value("${kafka.topic.my-topic}")
String myTopic;
@Value("${kafka.topic.my-topic2}")
String myTopic2;
BookProducerService producer;
private AtomicLong atomicLong = new AtomicLong();
BookController(BookProducerService producer) {
this.producer = producer;
}
@GetMapping("/send")
public String sendMessageToKafkaTopic(@RequestParam("name") String name) {
this.producer.sendMessage(myTopic, new Book(atomicLong.addAndGet(1), name));
this.producer.sendMessage(myTopic2, new Book(atomicLong.addAndGet(1), name));
return name+" : 消息已经发送!";
}
}
4.6 book 的生成者业务
@Service
public class BookProducerService {
private static final Logger logger = LoggerFactory.getLogger(BookProducerService.class);
private final KafkaTemplate<String, Object> kafkaTemplate;//通过构造方法进行注入
public BookProducerService(KafkaTemplate<String, Object> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void sendMessage(String topic, Object o) {
ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, o);
future.addCallback(
result -> logger.info("生产者成功发送消息到topic:{} partition:{}的消息",
result.getRecordMetadata().topic(),
result.getRecordMetadata().partition()),
ex -> logger.error("生产者发送消失败，原因：{}", ex.getMessage()));
}
}
4.7 book的消费者业务
@Service
public class BookConsumerService {
@Value("${kafka.topic.my-topic}")
private String myTopic;
@Value("${kafka.topic.my-topic2}")
private String myTopic2;
private final Logger logger = LoggerFactory.getLogger(BookProducerService.class);
private final ObjectMapper objectMapper = new ObjectMapper();
@KafkaListener(topics = {"${kafka.topic.my-topic}"}, groupId = "group1")
public void consumeMessage(ConsumerRecord<String, String> bookConsumerRecord) {
try {
Book book = objectMapper.readValue(bookConsumerRecord.value(), Book.class);
logger.info("消费者消费topic:{} partition:{}的消息 -> {}", bookConsumerRecord.topic(), bookConsumerRecord.partition(), book.toString());
} catch (JsonProcessingException e) {
e.printStackTrace();
}
}
@KafkaListener(topics = {"${kafka.topic.my-topic2}"}, groupId = "group2")
public void consumeMessage2(Book book,ConsumerRecord<String,String> bookConsumerRecord) throws JsonProcessingException {
Book value = objectMapper.readValue(bookConsumerRecord.value(), Book.class);
logger.info("消费者消费topic:{} partition:{}的消息 -> {}", bookConsumerRecord.topic(), bookConsumerRecord.partition(), value.toString());
logger.info("消费者消费{}的消息 -> {}", myTopic2, book.toString());
}
}
代码整体目录如下
4.8 启动成功界面
4.9 浏览器访问
4.10 控制台显示
