如何使用kafka发送顺序消息

　　　　Properties props = new Properties();
       // props.put("bootstrap.servers", "10.4.4.91:9092,10.4.4.92:9092,10.4.4.93:9092");
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");//"acks"
        props.put(ProducerConfig.RETRIES_CONFIG, 120000);//retries
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);//"batch.size"
        props.put(ProducerConfig.LINGER_MS_CONFIG, 0);//"linger.ms"
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);//"buffer.memory"
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);//"request.timeout.ms"
        props.put(ProducerConfig.CLIENT_ID_CONFIG,"drtest");//"client.id"
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"gzip");//"compression.type"
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//"key.serializer"
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");// "value.serializer"
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");//max.in.flight.requests.per.connection

Producer producer = new KafkaProducer<>(props);

Producer<String,String> producer = new KafkaProducer<>(props);

RecordMetadata recordMetadata = producer.send(new ProducerRecord<>("test",
            "testV")).get();

...

public final class RecordAccumulator {

...
  private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches;
...

kafkaProducer初始化时：

this.accumulator = new RecordAccumulator(logContext,
        config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
        this.totalMemorySize,
        this.compressionType,
        config.getLong(ProducerConfig.LINGER_MS_CONFIG),
        retryBackoffMs,
        metrics,
        time,
        apiVersions,
        transactionManager);

}

....
this.sender = new Sender(logContext,
                    client,
                    this.metadata,
                    this.accumulator,
                    maxInflightRequests == 1, //guaranteeMessageOrder属性配置
                    config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
                    acks,
                    retries,
                    metricsRegistry.senderMetrics,
                    Time.SYSTEM,
                    this.requestTimeoutMs,
                    config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
                    this.transactionManager,
                    apiVersions);

String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();

....

public boolean canSendMore(String node) {
        Deque<NetworkClient.InFlightRequest> queue = requests.get(node);
　　　　　　//能继续发送数据的条件之一为队列大小小于max.in.flight.requests.per.connection配置的值
        return queue == null || queue.isEmpty() ||
               (queue.peekFirst().send.completed() && queue.size() < this.maxInFlightRequestsPerConnection);
    }

final class InFlightRequests {

    private final int maxInFlightRequestsPerConnection;//此项配置即为max.in.flight.requests.per.connection
    private final Map<String, Deque<NetworkClient.InFlightRequest>> requests = new HashMap<>();
    /** Thread safe total number of in flight requests. */
    private final AtomicInteger inFlightRequestCount = new AtomicInteger(0);

// remove any nodes we aren‘t ready to send to
        Iterator<Node> iter = result.readyNodes.iterator();
        long notReadyTimeout = Long.MAX_VALUE;
        while (iter.hasNext()) {
            Node node = iter.next();
            if (!this.client.ready(node, now)) {
                iter.remove();
                notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, now));
            }
        }

Send send = request.toSend(destination, header);
        InFlightRequest inFlightRequest = new InFlightRequest(
                clientRequest,
                header,
                isInternalRequest,
                request,
                send,
                now);
        this.inFlightRequests.add(inFlightRequest);
        selector.send(send);