环境
kafka 版本: 0.11
springboot 版本: 2.4.4
集群: 三台
情境:
主题: 名称: topic1, 6个分区 副本数为3
如下面代码: 生产者没有指定分区,没有指定key, 连续发送消息
消费者, 默认消费组, 没有指定消费分区
问题:
我有三个消费者, 一个生产者, 但是生产者,生产的消息全部被其中一个消费者消费, 查看消息内容发现,生产者在没有指定分区,没有指定key的时候,全部分配到了一个分区,分区编号为0,所以被同一个消费者消费
但是在没有指定分区,没有指定key的时候,Kafka的分区策略不应该是没有 partition 值又没有 key 值的情况下,第一次调用时随机生成一个整数(后面每次调用在这个整数上自增),将这个值与 topic 可用的 partition 总数取余得到 partition 值。
// topic 配置
@Configuration
public class KafkaInitConfig {
@Bean
public NewTopic topic1() {
return TopicBuilder.name("topic1").partitions(6).replicas(3).compact().build();
}
}
//生产者:
@Slf4j
@RestController
@RequestMapping("/kafka-tests/")
public class KafkaTestController {
private static final String TOPIC = "topic1";
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
/**
* 生产消息
*/
@PostMapping("actions/send-messages/")
public void sendMessage() {
for (int i = 0; i < 10; i++) {
String message = UUID.randomUUID().toString();
kafkaTemplate.send(TOPIC, message).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable ex) {
log.error("消息发送失败 ", ex);
}
@Override
public void onSuccess(SendResult<String, Object> result) {
log.info("消息发送成功,{}", result);
}
});
}
}
}
// 消费者
@Slf4j
@Service
public class DefaultKafkaConsumerService {
@KafkaListener(topics = { "topic" })
public void retrieveMessage(ConsumerRecord<?, ?> message) {
System.out.println(message);
}
}
配置信息:
###########【Kafka集群】###########
spring.kafka.bootstrap-servers=192.168.13.3:9092,192.168.13.4:9092,192.168.13.5:9092
###########【初始化生产者配置】###########
# 重试次数
spring.kafka.producer.retries=0
# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks=1
# 批量大小
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了
# 生产端缓冲区大小
spring.kafka.producer.buffer-memory = 33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
###########【初始化消费者配置】###########
# 默认的消费组ID
spring.kafka.consumer.properties.group.id=defaultConsumerGroup
# 是否自动提交offset
spring.kafka.consumer.enable-auto-commit=true
# 提交offset延时(接收到消息后多久提交offset)
spring.kafka.consumer.auto.commit.interval.ms=1000
# 当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
spring.kafka.consumer.auto-offset-reset=latest
# 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
spring.kafka.consumer.properties.session.timeout.ms=120000
# 消费请求超时时间
spring.kafka.consumer.properties.request.timeout.ms=180000
# Kafka提供的序列化和反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消费端监听的topic不存在时,项目启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal=false
kafka客户端如果用的是默认的话,没有修改默认分区发送规则,那应该是平均分布到6个分区上的。
从代码上看不出来,你是一个消费者,还是三个消费者。
基于你的描述,无法真正确认是否只集中到了某个分区,你可以使用kafka自带的命令来确认:
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group
来自:kafka命令大全
如果区分真的只发送到了1分区,那你重点检查你的发送逻辑,而不需要关注消费者。
你的答案