kafka 同一个消费组里 多个消费者 使用 consumer.assign (topicPartitions);会造成数据重复吗？

不会 kafka保证的了一条消费只能消费一次

@青牛 謝謝 但是我写了一个例子是起了两个线程（模拟有两个消费者）结果就是线程A消费分区1的时候线程B也在消费分区1 不是说在同一个分组下一个分区只能被一个消费者消费吗 ？那为什么线程A和线程2会出现同时消费一个分区的情况？官网上的解释是使用assign就不會被group管理了 那这是不是就说明使用assign后会造成数据重复消费的问题？

public class kafkaConsumer extends Thread {

private String topic;

public kafkaConsumer() {
    super();

}

@Override
public void run() {
    List<String> topicList = Arrays.asList("topic1","topic2","topic3","topic4","topic5","topic6");
    KafkaConsumer consumer = createConsumer("aa");
    List<TopicPartition> topicPartitions = new ArrayList<>();
    for(String topic : topicList){
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        for (PartitionInfo info : partitionInfos) {
            TopicPartition topicPartition = new TopicPartition(info.topic(), info.partition());
            topicPartitions.add(topicPartition);
        }
    }
    consumer1.assign(topicPartitions);
    while (true){
        ConsumerRecords<String, String> records = consumer.poll(1100);

        for (ConsumerRecord<String, String> record : records) {

            System.out.println("---------"+Thread.currentThread().getName()+"---"+record.topic()+"---"+record.partition()+"-----"+record.value());

        }
        try {
            Thread.sleep(2000);
        }catch (Exception e){}
    }
}

private KafkaConsumer createConsumer(String groupName) {
    Properties props = new Properties();
    props.put("bootstrap.servers", "127.0.0.1:9092");
    props.put("group.id", groupName);
    props.put("enable.auto.commit", "true");
    props.put("max.poll.records", "100");//原来500 减小为100
    props.put("auto.commit.interval.ms", "1000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    return consumer;
}

public static void main(String[] args) {
    new kafkaConsumer().start();// 使用kafka集群中创建好的主题 test
    new kafkaConsumer().start();// 使用kafka集群中创建好的主题 test

}

}

是不是我在代码中使用try {
Thread.sleep(2000);
}catch (Exception e){}
导致的？

KafkaConsumer consumer = createConsumer("aa"); 把consumer的名字改成每个线程不一样试试

@青牛 不一样是没问题的

你好，请问最后是怎么解决的？我也遇到同样问题