JHipster 与 kafka 真实应用使用示例

Question

在 JHipster 版本 6.6.0 中，Kafka 使用模型已从标准的生产者/消费者类更改为 WebResource 级别。没有实际示例，这种变化的优势是什么，以及如何在实际应用中使用这种变化。

假设我们有 Service A 和 Service B。这两个服务之间的通信必须通过 Kafka 事件来完成。

问题是 - 我必须这样做 Service B 开始监听来自 Service A 主题的事件。在当前的配置中，看起来我必须手动触发/consumes端点，但这没有任何意义，因为我期望该服务将在应用程序启动并运行后开始侦听指定的主题列表。

如果对此主题有任何评论以帮助我理解这一点，我将不胜感激。

示例： jhipster 7.1.0 生成此资源：

服务 A - 网关

package com.stukans.refirmware.gateway.web.rest;

import com.stukans.refirmware.gateway.config.KafkaProperties;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderOptions;
import reactor.kafka.sender.SenderRecord;
import reactor.kafka.sender.SenderResult;

@RestController
@RequestMapping("/api/gateway-kafka")
public class GatewayKafkaResource {

    private final Logger log = LoggerFactory.getLogger(GatewayKafkaResource.class);

    private final KafkaProperties kafkaProperties;
    private KafkaSender<String, String> sender;

    public GatewayKafkaResource(KafkaProperties kafkaProperties) {
        this.kafkaProperties = kafkaProperties;
        this.sender = KafkaSender.create(SenderOptions.create(kafkaProperties.getProducerProps()));
    }

    @PostMapping("/publish/{topic}")
    public Mono<PublishResult> publish(
        @PathVariable String topic,
        @RequestParam String message,
        @RequestParam(required = false) String key
    ) {
        log.debug("REST request to send to Kafka topic {} with key {} the message : {}", topic, key, message);
        return Mono
            .just(SenderRecord.create(topic, null, null, key, message, null))
            .as(sender::send)
            .next()
            .map(SenderResult::recordMetadata)
            .map(
                metadata ->
                    new PublishResult(metadata.topic(), metadata.partition(), metadata.offset(), Instant.ofEpochMilli(metadata.timestamp()))
            );
    }

    @GetMapping("/consume")
    public Flux<String> consume(@RequestParam("topic") List<String> topics, @RequestParam Map<String, String> consumerParams) {
        log.debug("REST request to consume records from Kafka topics {}", topics);
        Map<String, Object> consumerProps = kafkaProperties.getConsumerProps();
        consumerProps.putAll(consumerParams);
        consumerProps.remove("topic");

        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create(consumerProps).subscription(topics);
        return KafkaReceiver.create(receiverOptions).receive().map(ConsumerRecord::value);
    }

    private static class PublishResult {

        public final String topic;
        public final int partition;
        public final long offset;
        public final Instant timestamp;

        private PublishResult(String topic, int partition, long offset, Instant timestamp) {
            this.topic = topic;
            this.partition = partition;
            this.offset = offset;
            this.timestamp = timestamp;
        }
    }
}

服务 B - 代理

package com.stukans.refirmware.agent.web.rest;

import com.stukans.refirmware.agent.config.KafkaProperties;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderOptions;
import reactor.kafka.sender.SenderRecord;
import reactor.kafka.sender.SenderResult;

@RestController
@RequestMapping("/api/agent-kafka")
public class AgentKafkaResource {

    private final Logger log = LoggerFactory.getLogger(AgentKafkaResource.class);

    private final KafkaProperties kafkaProperties;
    private KafkaSender<String, String> sender;

    public AgentKafkaResource(KafkaProperties kafkaProperties) {
        this.kafkaProperties = kafkaProperties;
        this.sender = KafkaSender.create(SenderOptions.create(kafkaProperties.getProducerProps()));
    }

    @PostMapping("/publish/{topic}")
    public Mono<PublishResult> publish(
        @PathVariable String topic,
        @RequestParam String message,
        @RequestParam(required = false) String key
    ) {
        log.debug("REST request to send to Kafka topic {} with key {} the message : {}", topic, key, message);
        return Mono
            .just(SenderRecord.create(topic, null, null, key, message, null))
            .as(sender::send)
            .next()
            .map(SenderResult::recordMetadata)
            .map(
                metadata ->
                    new PublishResult(metadata.topic(), metadata.partition(), metadata.offset(), Instant.ofEpochMilli(metadata.timestamp()))
            );
    }

    @GetMapping("/consume")
    public Flux<String> consume(@RequestParam("topic") List<String> topics, @RequestParam Map<String, String> consumerParams) {
        log.debug("REST request to consume records from Kafka topics {}", topics);
        Map<String, Object> consumerProps = kafkaProperties.getConsumerProps();
        consumerProps.putAll(consumerParams);
        consumerProps.remove("topic");

        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create(consumerProps).subscription(topics);
        return KafkaReceiver.create(receiverOptions).receive().map(ConsumerRecord::value);
    }

    private static class PublishResult {

        public final String topic;
        public final int partition;
        public final long offset;
        public final Instant timestamp;

        private PublishResult(String topic, int partition, long offset, Instant timestamp) {
            this.topic = topic;
            this.partition = partition;
            this.offset = offset;
            this.timestamp = timestamp;
        }
    }
}

这是唯一可用的与 Kafka 相关的代码。

在 6.6.0 版本之前，JHipster 生成标准的生产者/消费者类，我可以使用这些类来定义要收听的主题。现在还不清楚如何使用生成的代码来发出/监听事件。

Answer 1

首先，由于我们看不到您的代码，我们不知道它是如何（或应该）工作的......

无论如何，您都不应该通过 REST “触发”消费者；它们应该在服务启动时自动启动。如果没有消息，则它们在后台轮询并空闲，直到“生产者服务”向主题推送消息。

Answer 2

我做了这样一个自动触发消费者的服务：

@Service
public class KafkaConsumerService {

    private final Logger log = LoggerFactory.getLogger(KafkaConsumerService.class);

    private static final String GROUP_USER_CREATE_TOPIC = "GROUP_STORE.GROUP_USER.SAVE"; //<application name>.<dataset name>.<event>

    private final KafkaProperties kafkaProperties;

    private KafkaReceiver<String, String> kafkaReceiver;

    private final ObjectMapper objectMapper = new ObjectMapper();

    private final GroupMemberService groupMemberService;

    public KafkaConsumerService(KafkaProperties kafkaProperties, GroupMemberService groupMemberService) {
        this.kafkaProperties = kafkaProperties;
        this.groupMemberService = groupMemberService;
    }

    @PostConstruct
    public void start() {
        log.info("Kafka consumer starting...");
        Map<String, Object> consumerProps = kafkaProperties.getConsumerProps();
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create(consumerProps)
            .subscription(Collections.singletonList(GROUP_USER_CREATE_TOPIC));
        this.kafkaReceiver = KafkaReceiver.create(receiverOptions);
        consumeGroupMember().subscribe();
    }

    public Flux<GroupMemberDTO> consumeGroupMember() {
        log.debug("consumer group member....");
        return this.kafkaReceiver
            .receive()
            .map(ConsumerRecord::value)
            .flatMap(
                record -> {
                    try {
                        GroupMemberDTO groupMemberDTO = objectMapper.readValue(record, GroupMemberDTO.class);
                        log.debug("Complete convert object: {}", groupMemberDTO);
                        return groupMemberService.insert(groupMemberDTO);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                    }
                }
            );
    }

    public void shutdown() {
        log.info("Shutdown kafka consumer");
    }
}