PartitionManager分区管理器怎么使用

这篇文章主要介绍"PartitionManager分区管理器怎么使用"，在日常操作中，相信很多人在PartitionManager分区管理器怎么使用问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答"PartitionManager分区管理器怎么使用"的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

阅读背景：对于java内部类有一个粗浅的认识

阅读目的：了解kafka 分区是如何在Storm接口之中进行管理的

最终主题：详尽的梳理PartitionManager的整个过程

package com.mixbox.storm.kafka;import backtype.storm.Config;import backtype.storm.metric.api.CombinedMetric;import backtype.storm.metric.api.CountMetric;import backtype.storm.metric.api.MeanReducer;import backtype.storm.metric.api.ReducedMetric;import backtype.storm.spout.SpoutOutputCollector;import com.google.common.collect.ImmutableMap;import com.mixbox.storm.kafka.KafkaSpout.EmitState;import com.mixbox.storm.kafka.KafkaSpout.MessageAndRealOffset;import com.mixbox.storm.kafka.trident.MaxMetric;import kafka.api.OffsetRequest;import kafka.javaapi.consumer.SimpleConsumer;import kafka.javaapi.message.ByteBufferMessageSet;import kafka.message.MessageAndOffset;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.util.*;/** * 分区的管理器 *  * @author Yin Shuai *  */public class PartitionManager {        public static final Logger LOG = LoggerFactory                        .getLogger(PartitionManager.class);        private final CombinedMetric _fetchAPILatencyMax;        private final ReducedMetric _fetchAPILatencyMean;        private final CountMetric _fetchAPICallCount;        private final CountMetric _fetchAPIMessageCount;        /**         * kafka MessageID 封装了 partition 和offset         *          * @author Yin Shuai         */        static class KafkaMessageId {                public Partition partition;                public long offset;                public KafkaMessageId(Partition partition, long offset) {                        this.partition = partition;                        this.offset = offset;                }        }        // 被发送的偏移量        Long _emittedToOffset;        SortedSet _pending = new TreeSet();        // 已经提交的        Long _committedTo;        // 等待去发射        LinkedList _waitingToEmit = new LinkedList();        // 分区        Partition _partition;        // Storm Spout的配置文件        SpoutConfig _spoutConfig;        // topology 的实例ID        String _topologyInstanceId;        // kafka 底层的消费者ID        SimpleConsumer _consumer;        // 动态的分区Connection        DynamicPartitionConnections _connections;        //ZKState 状态的维护        ZkState _state;        //Storm的配置文件        Map _stormConf;        //        @SuppressWarnings("unchecked")        public PartitionManager(DynamicPartitionConnections connections,                        String topologyInstanceId, ZkState state, Map stormConf,                        SpoutConfig spoutConfig, Partition id) {                _partition = id;                _connections = connections;                _spoutConfig = spoutConfig;                _topologyInstanceId = topologyInstanceId;                _consumer = connections.register(id.host, id.partition);                _state = state;                _stormConf = stormConf;                String jsonTopologyId = null;                Long jsonOffset = null;                String path = committedPath();                try {                                                Map json = _state.readJSON(path);                        LOG.info("Read partition information from: " + path + " --> "                                        + json);                        if (json != null) {                                jsonTopologyId = (String) ((Map) json                                                .get("topology")).get("id");                                jsonOffset = (Long) json.get("offset");                        }                } catch (Throwable e) {                        LOG.warn("Error reading and/or parsing at ZkNode: " + path, e);                }                if (jsonTopologyId == null || jsonOffset == null) { // failed to parse                                                                                                                        // JSON?                        _committedTo = KafkaUtils.getOffset(_consumer, spoutConfig.topic,                                        id.partition, spoutConfig);                        LOG.info("No partition information found, using configuration to determine offset");                } else if (!topologyInstanceId.equals(jsonTopologyId)                                && spoutConfig.forceFromStart) {                        _committedTo = KafkaUtils.getOffset(_consumer, spoutConfig.topic,                                        id.partition, spoutConfig.startOffsetTime);                        LOG.info("Topology change detected and reset from start forced, using configuration to determine offset");                } else {                        _committedTo = jsonOffset;                        LOG.info("Read last commit offset from zookeeper: " + _committedTo                                        + "; old topology_id: " + jsonTopologyId                                        + " - new topology_id: " + topologyInstanceId);                }                LOG.info("Starting " + _partition + " from offset " + _committedTo);                _emittedToOffset = _committedTo;                _fetchAPILatencyMax = new CombinedMetric(new MaxMetric());                _fetchAPILatencyMean = new ReducedMetric(new MeanReducer());                _fetchAPICallCount = new CountMetric();                _fetchAPIMessageCount = new CountMetric();        }        public Map getMetricsDataMap() {                Map ret = new HashMap();                ret.put(_partition + "/fetchAPILatencyMax",                                _fetchAPILatencyMax.getValueAndReset());                ret.put(_partition + "/fetchAPILatencyMean",                                _fetchAPILatencyMean.getValueAndReset());                ret.put(_partition + "/fetchAPICallCount",                                _fetchAPICallCount.getValueAndReset());                ret.put(_partition + "/fetchAPIMessageCount",                                _fetchAPIMessageCount.getValueAndReset());                return ret;        }        // returns false if it's reached the end of current batch        public EmitState next(SpoutOutputCollector collector) {                                //等待去发送的 为空了。                if (_waitingToEmit.isEmpty()) {                        // 开始装载                        fill();                }                                while (true) {                                                //检索并移除List中间的第一个元素                        MessageAndRealOffset toEmit = _waitingToEmit.pollFirst();                                                //要发送的为空的时候, 没有发生的                        if (toEmit == null) {                                return EmitState.NO_EMITTED;                        }                                                // 这里的tups                        Iterable> tups = KafkaUtils.generateTuples(                                        _spoutConfig, toEmit.msg);                        if (tups != null) {                                for (List