partition生成规则

partition生成规则

  • 不指定key

    private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
            TopicPartition tp = null;
    
            try {
                this.throwIfProducerClosed();
                long nowMs = this.time.milliseconds();
    
                KafkaProducer.ClusterAndWaitTime clusterAndWaitTime;
                try {
                    clusterAndWaitTime = this.waitOnMetadata(record.topic(), record.partition(), nowMs, this.maxBlockTimeMs);
                } catch (KafkaException var22) {
                    if (this.metadata.isClosed()) {
                        throw new KafkaException("Producer closed while send in progress", var22);
                    }
    
                    throw var22;
                }
    
                nowMs += clusterAndWaitTime.waitedOnMetadataMs;
                long remainingWaitMs = Math.max(0L, this.maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
                Cluster cluster = clusterAndWaitTime.cluster;
    
                byte[] serializedKey;
                try {
                    serializedKey = this.keySerializer.serialize(record.topic(), record.headers(), record.key());
                } catch (ClassCastException var21) {
                    throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() + " to class " + this.producerConfig.getClass("key.serializer").getName() + " specified in key.serializer", var21);
                }
    
                byte[] serializedValue;
                try {
                    serializedValue = this.valueSerializer.serialize(record.topic(), record.headers(), record.value());
                } catch (ClassCastException var20) {
                    throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() + " to class " + this.producerConfig.getClass("value.serializer").getName() + " specified in value.serializer", var20);
                }
    
                int partition = this.partition(record, serializedKey, serializedValue, cluster);
                tp = new TopicPartition(record.topic(), partition);
                this.setReadOnly(record.headers());
                Header[] headers = record.headers().toArray();
                int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(this.apiVersions.maxUsableProduceMagic(), this.compressionType, serializedKey, serializedValue, headers);
                this.ensureValidRecordSize(serializedSize);
                long timestamp = record.timestamp() == null ? nowMs : record.timestamp();
                if (this.log.isTraceEnabled()) {
                    this.log.trace("Attempting to append record {} with callback {} to topic {} partition {}", new Object[]{record, callback, record.topic(), partition});
                }
    
                Callback interceptCallback = new KafkaProducer.InterceptorCallback(callback, this.interceptors, tp);
                if (this.transactionManager != null && this.transactionManager.isTransactional()) {
                    this.transactionManager.failIfNotReadyForSend();
                }
    
                RecordAppendResult result = this.accumulator.append(tp, timestamp, serializedKey, serializedValue, headers, interceptCallback, remainingWaitMs, true, nowMs);
                if (result.abortForNewBatch) {
                    int prevPartition = partition;
                    this.partitioner.onNewBatch(record.topic(), cluster, partition);
                    partition = this.partition(record, serializedKey, serializedValue, cluster);
                    tp = new TopicPartition(record.topic(), partition);
                    if (this.log.isTraceEnabled()) {
                        this.log.trace("Retrying append due to new batch creation for topic {} partition {}. The old partition was {}", new Object[]{record.topic(), partition, prevPartition});
                    }
    
                    interceptCallback = new KafkaProducer.InterceptorCallback(callback, this.interceptors, tp);
                    result = this.accumulator.append(tp, timestamp, serializedKey, serializedValue, headers, interceptCallback, remainingWaitMs, false, nowMs);
                }
    
                if (this.transactionManager != null && this.transactionManager.isTransactional()) {
                    this.transactionManager.maybeAddPartitionToTransaction(tp);
                }
    
                if (result.batchIsFull || result.newBatchCreated) {
                    this.log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
                    this.sender.wakeup();
                }
    
                return result.future;
            } catch (ApiException var23) {
                this.log.debug("Exception occurred during message send:", var23);
                if (callback != null) {
                    callback.onCompletion((RecordMetadata)null, var23);
                }
    
                this.errors.record();
                this.interceptors.onSendError(record, tp, var23);
                return new KafkaProducer.FutureFailure(var23);
            } catch (InterruptedException var24) {
                this.errors.record();
                this.interceptors.onSendError(record, tp, var24);
                throw new InterruptException(var24);
            } catch (KafkaException var25) {
                this.errors.record();
                this.interceptors.onSendError(record, tp, var25);
                throw var25;
            } catch (Exception var26) {
                this.interceptors.onSendError(record, tp, var26);
                throw var26;
            }
        }
    
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster, int numPartitions) {
            return keyBytes == null ? this.stickyPartitionCache.partition(topic, cluster) : Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    
    public int partition(String topic, Cluster cluster) {
            Integer part = (Integer)this.indexCache.get(topic);
            return part == null ? this.nextPartition(topic, cluster, -1) : part;
        }
    
    public int nextPartition(String topic, Cluster cluster, int prevPartition) {
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            Integer oldPart = (Integer)this.indexCache.get(topic);
            Integer newPart = oldPart;
            if (oldPart != null && oldPart != prevPartition) {
                return (Integer)this.indexCache.get(topic);
            } else {
                List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
                Integer random;
                if (availablePartitions.size() < 1) {
                    random = Utils.toPositive(ThreadLocalRandom.current().nextInt());
                    newPart = random % partitions.size();
                } else if (availablePartitions.size() == 1) {
                    newPart = ((PartitionInfo)availablePartitions.get(0)).partition();
                } else {
                    while(newPart == null || newPart.equals(oldPart)) {
                        random = Utils.toPositive(ThreadLocalRandom.current().nextInt());
                        newPart = ((PartitionInfo)availablePartitions.get(random % availablePartitions.size())).partition();
                    }
                }
    
                if (oldPart == null) {
                    this.indexCache.putIfAbsent(topic, newPart);
                } else {
                    this.indexCache.replace(topic, prevPartition, newPart);
                }
    
                return (Integer)this.indexCache.get(topic);
            }
        }
    
    

由上可知,如果没有指定key 会根据topic ,headers,value 从现有的partition列表中随机选一个,但是一但选定(也就是存在prepartition)就直接用原来的。

  • 指定key

    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster, int numPartitions) {
            return keyBytes == null ? this.stickyPartitionCache.partition(topic, cluster) : Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    public static int murmur2(byte[] data) {
            int length = data.length;
            int seed = -1756908916;
            int m = 1540483477;
            int r = true;
            int h = seed ^ length;
            int length4 = length / 4;
    
            for(int i = 0; i < length4; ++i) {
                int i4 = i * 4;
                int k = (data[i4 + 0] & 255) + ((data[i4 + 1] & 255) << 8) + ((data[i4 + 2] & 255) << 16) + ((data[i4 + 3] & 255) << 24);
                k *= 1540483477;
                k ^= k >>> 24;
                k *= 1540483477;
                h *= 1540483477;
                h ^= k;
            }
    
            switch(length % 4) {
            case 3:
                h ^= (data[(length & -4) + 2] & 255) << 16;
            case 2:
                h ^= (data[(length & -4) + 1] & 255) << 8;
            case 1:
                h ^= data[length & -4] & 255;
                h *= 1540483477;
            default:
                h ^= h >>> 13;
                h *= 1540483477;
                h ^= h >>> 15;
                return h;
            }
        }
    
    public static int toPositive(int number) {
            return number & 2147483647;
        }
    
  • 指定partition,刚直接使用指定的partition,但是要尽量不要把所有的,特别是大量数据的情况放,放入同一个partition,不然可以导致kafka死掉。

上一篇:四条命令利用docker创建redis集群


下一篇:RedGate 工具SQLTEST 1.0.15.1