public void start(Function<String, List<byte[]>> payloads, double rate) {
        adjustRate(rate);
        adjustRateExecutor.scheduleWithFixedDelay(this::adjustRate, 0, ADJUST_RATE_INTERVAL_SECONDS, TimeUnit.SECONDS);
        int processors = Runtime.getRuntime().availableProcessors();
        // shard producers across processors
        int batchSize = Math.max(1, producers.size() / processors);
        for (int i = 0; i < producers.size(); i += batchSize) {
            int from = i;
            int to = Math.min(i + batchSize, producers.size());
            executor.submit(() -> start(producers.subList(from, to), payloads));
        }
    }
Start sending messages using the given payloads at the given rate.
tools/src/main/java/org/apache/kafka/tools/automq/perf/ProducerService.java