使用Reactive Kafka Stream和Spring WebFlux

@Component
public class StockPriceProducer {
    public static final String[] STOCKS = {"AAPL", "GOOG", "MSFT", "AMZN", "TSLA"};
    private static final String CURRENCY = "USD";

    private final ReactiveKafkaProducerTemplate<String, StockUpdate> kafkaProducer;
    private final NewTopic topic;
    private final Random random = new Random();

    public StockPriceProducer(KafkaProperties properties,
                              @Qualifier(TopicConfig.STOCK_PRICES_IN) NewTopic topic) {
        this.kafkaProducer = new ReactiveKafkaProducerTemplate<>(
                SenderOptions.create(properties.buildProducerProperties())
        );
        this.topic = topic;
    }

    public Flux<SenderResult<Void>> produceStockPrices(int count) {
        return Flux.range(0, count)
                .map(i -> {
                    String stock = STOCKS[random.nextInt(STOCKS.length)];
                    double price = 100 + (200 * random.nextDouble());
                    return MessageBuilder.withPayload(new StockUpdate(stock, price, CURRENCY, Instant.now()))
                            .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                            .build();
                })
                .flatMap(stock -> {
                    var newRecord = new ProducerRecord<>(
                            topic.name(),
                            stock.getPayload().symbol(),
                            stock.getPayload());

                    stock.getHeaders()
                            .forEach((key, value) -> newRecord.headers().add(key, value.toString().getBytes()));

                    return kafkaProducer.send(newRecord);
                });
    }
}

@Configuration
public class StockPriceProcessor {
    private static final String USD = "USD";
    private static final String EUR = "EUR";

    @Bean
    public Function<Flux<Message<StockUpdate>>, Flux<Message<StockUpdate>>> processStockPrices(
            ClickHouseRepository repository,
            CurrencyRate currencyRate
    ) {
        return stockPrices -> stockPrices.flatMapSequential(message -> {
            StockUpdate stockUpdate = message.getPayload();
            return repository.saveStockPrice(stockUpdate)
                    .flatMap(success -> Boolean.TRUE.equals(success) ? Mono.just(stockUpdate) : Mono.empty())
                    .flatMap(stock -> currencyRate.convertRate(USD, EUR, stock.price()))
                    .map(newPrice -> convertPrice(stockUpdate, newPrice))
                    .map(priceInEuro -> MessageBuilder.withPayload(priceInEuro)
                            .setHeader(KafkaHeaders.KEY, stockUpdate.symbol())
                            .copyHeaders(message.getHeaders())
                            .build());
        });
    }

    private StockUpdate convertPrice(StockUpdate stockUpdate, double newPrice) {
        return new StockUpdate(stockUpdate.symbol(), newPrice, EUR, stockUpdate.timestamp());
    }
}

public interface CurrencyRate {
    Mono<Double> convertRate(String from, String to, double amount);
}

public interface ClickHouseRepository {
    Mono<Boolean> saveStockPrice(StockUpdate stockUpdate);
    Flux<StockUpdate> findMinuteAvgStockPrices(Instant from, Instant to);
}

@Component
public class StockPriceConsumer {
    private final ReactiveKafkaConsumerTemplate<String, StockUpdate> kafkaConsumerTemplate;

    public StockPriceConsumer(@NonNull KafkaProperties properties,
                              @Qualifier(TopicConfig.STOCK_PRICES_OUT) NewTopic topic) {
        var receiverOptions = ReceiverOptions
                .<String, StockUpdate>create(properties.buildConsumerProperties())
                .subscription(List.of(topic.name()));
        this.kafkaConsumerTemplate = new ReactiveKafkaConsumerTemplate<>(receiverOptions);
    }

    @PostConstruct
    public void consume() {
        kafkaConsumerTemplate
                .receiveAutoAck()
                .doOnNext(consumerRecord -> {
                    // simulate processing
                    log.info(
                            "received key={}, value={} from topic={}, offset={}, partition={}", consumerRecord.key(),
                            consumerRecord.value(),
                            consumerRecord.topic(),
                            consumerRecord.offset(),
                            consumerRecord.partition());
                })
                .doOnError(e -> log.error("Consumer error",  e))
                .doOnComplete(() -> log.info("Consumed all messages"))
                .subscribe();
    }
}

@RestController
public class StocksApi {
    private final ClickHouseRepository repository;

    @Autowired
    public StocksApi(ClickHouseRepository repository) {
        this.repository = repository;
    }

    @GetMapping("/stock-prices-out")
    public Flux<StockUpdate> getAvgStockPrices(@RequestParam("from") @NotNull Instant from,
                                               @RequestParam("to") @NotNull Instant to) {
        if (from.isAfter(to)) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "'from' must come before 'to'");
        }

        return repository.findMinuteAvgStockPrices(from, to);
    }
}