zoukankan      html  css  js  c++  java
  • kafka的暂停消费和重新开始消费问题

    //暂停kafka的消费 暂停分区的分配
    consumer.unsubscribe();//此处不取消订阅暂停太久会出现订阅超时的错误
    consumer.pause(consumer.assignment());



    //重新消费分区,此处不重新分配会出错
    this.open(null,null,null);
        if (null == consumer) {
    Properties props = new Properties();
    props.put("bootstrap.servers", PropertiesUtil.getValue("bootstrap.servers"));
    // 消费者的组id
    props.put("group.id", constant.kafka_groupName);//Spider2
    props.put("enable.auto.commit", "false");
    // max.poll.interval.ms(官网给得默认值为3000)的意思为,当我们从kafkaServer端poll消息时,poll()的调用之间的最大延迟。
    // 这提供了消费者在获取更多记录之前可以空闲的时间量的上限。 如果在此超时到期之前未调用poll(),则认为使用者失败,并且消费
    // 者组将重新平衡以便将分区重新分配给其他消费者,而恰好这里我们设置了Thread.sleep(6000) > max.poll.interval.ms值,
    // 也就是我们在手动提交的时候,实际上分区信息已经被分配到整个消费者组里面的其它消费者了
    props.put("auto.commit.interval.ms", "3000");
    // 从poll(拉)的回话处理时长
    props.put("session.timeout.ms", "100000");
    props.put("request.timeout.ms", "200000");
    props.put("max.poll.records", "2");
    // poll的数量限制
    // props.put("max.poll.records", "100");
    /* props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");*/
    props.put("key.deserializer", StringDeserializer.class.getName());
    props.put("value.deserializer", StringDeserializer.class.getName());
    props.put("group.name", UUID.randomUUID().toString().replaceAll("-", ""));
    consumer = new KafkaConsumer<String, String>(props);

    // 订阅主题列表topic
    //consumer.subscribe(Arrays.asList("test_input"));
    }
    //注册kafka rebalanceListener
    //consumer.subscribe(Arrays.asList("test_etl"), new ConsumerRebalanceListener(){

    listener = new ConsumerRebalanceListener(){
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    System.out.printf("threadId = {}, onPartitionsRevoked.", Thread.currentThread().getId());
    consumer.commitSync(offsetsMap);
    consumer.commitSync();
    }
    @Override
    public void onPartitionsAssigned(
    Collection<TopicPartition> partitions) {
    System.out.printf("threadId = {}, onPartitionsAssigned.", Thread.currentThread().getId());
    consumer.commitSync();
    offsetsMap.clear();
    }};

    consumer.subscribe(Arrays.asList(topicName.split(",")[0],topicName.split(",")[1],topicName.split(",")[2]), listener);
    consumer.resume(consumer.assignment());
  • 相关阅读:
    生成缩略图时报GDI+中发生一般性错误
    跨数据库服务器查询和跨表更新
    javascript,jquery(闭包概念)
    聚集索引和非聚集索引(整理)
    ASP.NET MVC的过滤器
    如何在标题栏title前添加网站logo?
    sql语句执行顺序
    详解ASP.NET MVC的请求生命周期
    在事务中执行批量复制操作SqlBulkCopy,SqlTransaction .
    基于TCP 协议的RPC
  • 原文地址:https://www.cnblogs.com/yaohaitao/p/12172867.html
Copyright © 2011-2022 走看看