通过Spring Boot Webflux实现Reactor Kafka

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor.kafka</groupId>
<artifactId>reactor-kafka</artifactId>
<version>1.1.0.RELEASE</version>
</dependency>

/ ** 
     *调用返回的Mono将被发送到Spring Webflux，后者依赖于multi-reactor 事件循环和NIO 
     *以非阻塞方式处理请求，从而实现更多的并发请求。结果将
     通过一个名为Server Sent Events 发送。
     ** /
@PostMapping(value = "/payment")
    public Mono<Void> doPayment(@RequestBody CreatePaymentCommand payment) {
    / ** 
             当调用doPayment方法时，我们发送付款信息，获得Mono <Void>作为响应。
             当我们的付款成功发送事件到Kafka主题
             ** / 
            return paymentGateway.doPayment(payment);
}

public PaymentGatewayImpl() {
    final Map<String, Object> producerProps = new HashMap<>();
    producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    final SenderOptions<Integer, String> producerOptions = SenderOptions.create(producerProps);
    kafkaProducer = KafkaSender.create(producerOptions);
}

@Override
    public Mono<Void> doPayment(final CreatePaymentCommand createPayment) {
    final PaymentEvent payment = new PaymentEvent(createPayment.getId(), createPayment.getCreditCardNumber(), createPayment.getAmount(), gatewayName);
    String payload = toBinary(payment);
    SenderRecord<Integer, String, Integer> message = SenderRecord.create(new ProducerRecord<>("unconfirmed-transactions", payload), 1);
    return kafkaProducer.send(Mono.just(message)).next();
}
private String toBinary(Object object) {
    try {
        return objectMapper.writeValueAsString(object);
    }
    catch (JsonProcessingException e) {
        throw new IllegalArgumentException(e);
    }
}

public PaymentValidatorListenerImpl(PaymentValidator paymentValidator) {
    this.paymentValidator = paymentValidator;
    final Map<String, Object> consumerProps = new HashMap<>();
    consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
    consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "payment-validator-1");
    consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "payment-validator");
    consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    ReceiverOptions<Object, Object> consumerOptions = ReceiverOptions.create(consumerProps)
                    .subscription(Collections.singleton("unconfirmed-transactions"))
                    .addAssignListener(partitions -> log.debug("onPartitionsAssigned {}", partitions))
                    .addRevokeListener(partitions -> log.debug("onPartitionsRevoked {}", partitions));
    kafkaReceiver = KafkaReceiver.create(consumerOptions);
    /**
         * We create a receiver for new unconfirmed transactions
         */
    ((Flux<ReceiverRecord>) kafkaReceiver.receive())
                    .doOnNext(r -> {
        /**
                     * Each unconfirmed payment we receive, we convert to a PaymentEvent and process it
                     */
        final PaymentEvent paymentEvent = fromBinary((String) r.value(), PaymentEvent.class);
        processEvent(paymentEvent);
        r.receiverOffset().acknowledge();
    }
    )
                    .subscribe();
}
private void processEvent(PaymentEvent paymentEvent) {
    paymentValidator.calculateResult(paymentEvent);
}
private <T> T fromBinary(String object, Class<T> resultType) {
    try {
        return objectMapper.readValue(object, resultType);
    }
    catch (IOException e) {
        throw new IllegalArgumentException(e);
    }
}