@青牛 不一样是没问题的

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

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

}

}

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

@青牛 谢谢老师 ，资源是一定的 不能起两个storm 我用 多主题多分区 在一个spout里面订阅多个主题 循环这些主题的分区 取出每个分区的部分数据这样一定程度上解决吧

@青牛

@青牛 String sql 怎样包含多个建表语句？ CREATE TABLE IF NOT EXISTS ...; CREATE TABLE IF NOT EXISTS...;
这样执行statement.execute(allSql.toString());的时候就报 ParseException line 1:165 missing EOF at ';' near 'TEXTFILE'

@青牛 数据迁移到hive 需要将客户指定的表批量创建

@青牛 谢谢您 解决了 重启一次就好了

666

6666

@青牛 谢谢您

666

@青牛 没事了 谢谢您 balance_swith但是也没改 应该是默认的 反正就是好了

@shishuai19910217
@青牛 昨天有一个节点坏掉了 剩余的集群资源分配就正常了 今天把坏的节点修复了 又出现资源分配不均的情况了

@青牛 现在 数据分布的挺平均的

但是

还是这样