Function runAsyncProducer

tools/kafka-producer-performance/main.go:318–363

(topic string, partition, messageLoad, messageSize int,
	config *sarama.Config, brokers []string, throughput int)

}

func runAsyncProducer(topic string, partition, messageLoad, messageSize int,
	config *sarama.Config, brokers []string, throughput int) {
	producer, err := sarama.NewAsyncProducer(brokers, config)
	if err != nil {
		printErrorAndExit(69, "Failed to create producer: %s", err)
	}
	defer func() {
		// Print final metrics.
		printMetrics(os.Stdout, config.MetricRegistry)
		if err := producer.Close(); err != nil {
			printErrorAndExit(69, "Failed to close producer: %s", err)
		}
	}()

	messages := generateMessages(topic, partition, messageLoad, messageSize)

	messagesDone := make(chan struct{})
	go func() {
		for range messageLoad {
			select {
			case <-producer.Successes():
			case err = <-producer.Errors():
				printErrorAndExit(69, "%s", err)
			}
		}
		messagesDone <- struct{}{}
	}()

	if throughput > 0 {
		ticker := time.NewTicker(time.Second)
		for idx, message := range messages {
			producer.Input() <- message
			if (idx+1)%throughput == 0 {
				<-ticker.C
			}
		}
		ticker.Stop()
	} else {
		for _, message := range messages {
			producer.Input() <- message
		}
	}

	<-messagesDone
	close(messagesDone)
}

func runSyncProducer(topic string, partition, messageLoad, messageSize, routines int,
	config *sarama.Config, brokers []string, throughput int) {