Kafka的Offset详解
- 1、生产者Offset
- 2、消费者Offset
- 2.1、消费者
- 2.2、生产者
- 2.3、实体类对象
- 2.4、JSON工具类
- 2.5、项目配置文件
- 2.6、测试类
- 2.7、测试
- 2.8、总结
1、生产者Offset
2、消费者Offset
2.1、消费者
package com.power.consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class EventConsumer {
/**
* topics 用于指定从哪个主题中消费消息
* concurrency 用于指定有多少个消费者
* @param record
*/
@KafkaListener(topics = {"offSetTopic"}, groupId = "offSetGroup")
public void onEventA(ConsumerRecord<String, String> record) {
System.out.println(Thread.currentThread().getId()+"---> 消费消息 record = " + record);
}
}
2.2、生产者
package com.power.producer;
import com.power.model.User;
import com.power.util.JSONUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Date;
@Component
public class EventProducer {
@Resource
private KafkaTemplate<String,Object> kafkaTemplate;
public void sendEvent(){
for (int i = 0; i < 2; i++) {
User user = User.builder().id(i).phone("1567676767"+i).birthday(new Date()).build();
String userJson = JSONUtils.toJSON(user);
kafkaTemplate.send("offSetTopic","k"+i, userJson);
}
}
}
2.3、实体类对象
package com.power.model;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Date;
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Data
public class User {
private Integer id;
private String phone;
private Date birthday;
}
2.4、JSON工具类
package com.power.util;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class JSONUtils {
private static final ObjectMapper OBJECTMAPPER = new ObjectMapper();
public static String toJSON(Object object){
try {
return OBJECTMAPPER.writeValueAsString(object);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
public static <T> T toBean(String json,Class<T> clazz){
try {
return OBJECTMAPPER.readValue(json,clazz);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
}
2.5、项目配置文件
spring:
application:
#应用名称
name: spring-boot-06-kafka-offset
#kafka连接地址（ip+port）
kafka:
bootstrap-servers: <你的kafka服务器IP>:9092
#配置消费者的反序列化
consumer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
2.6、测试类
package com.power;
import com.power.producer.EventProducer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
@SpringBootTest
public class SpringBoot07KafkaBaseApplication {
@Resource
private EventProducer eventProducer;
@Test
void sendInterceptor(){
eventProducer.sendEvent();
}
}
2.7、测试
-
先启动生产者，会发送两条消息到kafka服务器
-
再启动消费者监听，此时我们发现，启动后的消费者并不会监听到生产者已发送的两条消息
-
在kafka安装目录的bin文件夹下执行命令：
./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group offSetGroup --describe
- 根据命令结果：查看kafka消费者的偏移量offset，我们发现当前消费者偏移量CURRENT-OFFSET值为2 ，当前日志记录的生产者消息偏移量LOG-END-OFFSET值为2，消费者偏移量和日志记录的生产者消息偏移量差值LAG值为0 ，所以消费者查询不到生产者发送的消息。
- 关闭消费者，再次使用生产者发送消息，再次执行命令查看消费者偏移量
- 此时我们发现消费者偏移量为4，日志记录的偏移量为6，两者差值为2，此时启动消费者，读取到了差值为2的数据
2.8、总结
- 消费者从什么地方开始消费，就看消费者的offset是多少，消费者启动后他的offset是多少。
- 消费者offset是多少，可以通过命令查看
./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group offSetGroup --describe
