海汼部落
  • 首页
  • 课程
  • 实战
  • 云平台
  • 部落
  • 公告
  • 分享
  • 问答
    登 录 注册

shishuai19910217

第 2944 位会员

0 关注者
22 评论
16 话题

  • Ta 发布的话题
  • Ta 发表的回复
  • Ta 关注的用户
  • Ta 赞过的话题
  1. 个人中心
  2. Ta 发表的回复(22)
  • 信息被删除或无权限查看
  • 信息被删除或无权限查看
  • kafka 同一个消费组里 多个消费者 使用 consumer.assign (topicPartitions);会造成数据重复吗?

    @青牛 不一样是没问题的

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

    file

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

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

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

    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
    
    }

    }

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

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

  • storm 怎样均衡的处理 kafka 数据?

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

  • spark shuffle 在原始的 Hash Shuffle 机制中怎么生存 bucket?

    @青牛

    file

  • hive 怎样批量创建表?

    @青牛 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 怎样批量创建表?

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

  • 使用 jdbc 连接 hive 出错 hadoop is not allowed to impersonate hadoop????

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

  • 海牛部落 spark 系列教程(四十六):kafka 介绍与安装、kafka-java-API、spark-streaming-kafka、cogroup

    666

  • spark 基于内存的分布式计算框架

    6666

  • 急求 spark 视频教程?

    @青牛 谢谢您

  • «
  • 1
  • 2
  • »

为技术学习者提供一个动手实战、分享创造、结识伙伴、协同互助的平台。    

  京公网安备 11011402010672号        京ICP备17041118号-2