- 引入依赖
<!--kafka依赖-->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
- application.yml配置文件，配置生产者和消费者
spring:
application:
name: simple-spring-boot-demo
kafka:
# kafka 集群地址,逗号分隔
bootstrap-servers: 127.0.0.1:9092
# 生产者
producer:
# 设置大于0的值，则客户端会将发送失败的消息重新发送，默认重试间隔100ms，retry-backoff-ms重试间隔
retries: 3
# 本地线程会从缓冲区中一次拉16k数据，发送到kafka集群，如果线程拉不到16k数据，隔10ms，也会将拉到的数据发送，这个间隔也可以配linger-ms
batch-size: 16384
# 消息本地缓冲区，用来存放要发送的数据,可以提高消息发送性能，32M
buffer-memory: 33554432
#procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
#acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
#acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
#acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器follower仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。(里面有默认的配置min.insync.replicas=2,默认值为1，这种情况和ack=1相同了，推荐配置大于等于2)
#可以设置的值为：all, -1, 0, 1
acks: 1
# 指定消息key和消息体的编解码方式
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
# 消费者
consumer:
# 消费组
group-id: com.jenson
# 自动提交，如果enable-auto-commit为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000，auto-commit-interval(自动提交频率设置)
enable-auto-commit: false
# earliest 新的消费组的消费者启动时，默认会从第一个位置开始消费,latest 默认的，消费新消息
auto-offset-reset: earliest
# 一次最多拉多少条
max-poll-records: 500
# 指定消息key和消息体的编解码方式
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
listener:
# 当每一条记录被消费者监听器 (ListenerConsumer） 处理之后提交
# RECORD
# 当每一批poll()的数据被消费者监听器 (ListenerConsumer） 处理之后提交
# BATCH
# 当每一批poll()的数据被消费者监听器 (ListenerConsumer） 处理之后，距离上次提交时间大于TIME时提交
# TIME
# 当每一批poll()的数据被消费者监听器 (ListenerConsumer） 处理之后，被处理record数量大于等于COUNT时提交
# COUNT
# TIME | COUNT 有一个条件满足时提交
# COUNT_IIME
# 当每一批poll()的数据被消费者监听器 (ListenerConsumer）处理之后，手动调用Acknowledgment.acknowledge()后提交
# MANUAL
# 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种
# MANUAL_IMMEDIATE
ack-mode: MANUAL_IMMEDIATE
- 生产者代码编写
这里通过接口触发消息发送
package com.jenson.controller.v1;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
/**
* @author Jenson
*/
@RestController
@RequestMapping("/kafka")
@Slf4j
public class KafkaController {
@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
@GetMapping("/send")
public Boolean sendMessage(@RequestParam String message){
String topic = "topic1";
// 发送消息
kafkaTemplate.send(topic, 0 , "key", message);
return Boolean.TRUE;
}
}
KafkaTemplate 有很多send的方法，此处使用的这个源码注释如下：
/**
* Send the data to the provided topic with the provided key and partition.
* @param topic the topic.
* @param partition the partition.
* @param key the key.
* @param data the data.
* @return a Future for the {@link SendResult}.
*/
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
- 消费者编写
package com.jenson.consumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
/**
* @author Jenson
*/
@Component
@Slf4j
public class MyConsumer {
@KafkaListener(topics = "topic1", groupId = "com.jenson")
public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
String value = record.value();
log.info("value : " + value);
log.info("record : {} ", record);
// 手动提交offset ，只有在配置文件中配置了手动提交模式，ack才有用
ack.acknowledge();
}
}
内容来源于B站教学视频：https://www.bilibili.com/video/BV1Xy4y1G7zA?p=23&share_source=copy_web
网友评论