如何监听来自 Kafka 的正确 ACK 消息

我正在做一个POC使用 Spring Boot 和 Kafka 进行事务性项目，我有以下疑问：

设想：一个微服务MSPUB1接收来自客户的请求。该请求发布有关主题的消息TRANSACTION_TOPIC1在 Kafka 上，但微服务可以并行接收多个请求。微服务监听主题TRANSACTION_RESULT1检查交易是否完成。

在流媒体平台的另一边，另一个微服务MSSUB1正在听主题TRANSACTION_TOPIC1并处理所有消息并将结果发布到：TRANSACTION_RESULT1

最好的方法是什么MSPUB1了解消息是否与主题相关TRANSACTION_RESULT1符合他原来的要求吗？微服务MSPUB1可以有一个在初始主题上发布的任何消息的 IDTRANSACTION_TOPIC1并被移至TRANSACTION_RESULT1

问题：当你在读分区时，你移动了指针，但是在多个请求的并发环境下，如何检查消息是否在主题上TRANSACTION_RESULT1是预期的吗？

提前谢谢了

胡安·安东尼奥

一种方法是使用弹簧集成BarrierMessageHandler.

这是一个示例应用程序。希望这是不言自明的。需要 Kafka 0.11 或更高版本...

@SpringBootApplication
@RestController
public class So48349993Application {

    private static final Logger logger = LoggerFactory.getLogger(So48349993Application.class);

    private static final String TRANSACTION_TOPIC1 = "TRANSACTION_TOPIC1";

    private static final String TRANSACTION_RESULT1 = "TRANSACTION_RESULT1";

    public static void main(String[] args) {
        SpringApplication.run(So48349993Application.class, args);
    }

    private final Exchanger exchanger;

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public So48349993Application(Exchanger exchanger,
            KafkaTemplate<String, String> kafkaTemplate) {
        this.exchanger = exchanger;
        this.kafkaTemplate = kafkaTemplate;
        kafkaTemplate.setDefaultTopic(TRANSACTION_RESULT1);
    }

    @RequestMapping(path = "/foo/{id}/{other}", method = RequestMethod.GET)
    @ResponseBody
    public String foo(@PathVariable String id, @PathVariable String other) {
        logger.info("Controller received: " + other);
        String reply = this.exchanger.exchange(id, other);
        // if reply is null, we timed out
        logger.info("Controller replying: " + reply);
        return reply;
    }

    // Client side

    @MessagingGateway(defaultRequestChannel = "outbound", defaultReplyTimeout = "10000")
    public interface Exchanger {

        @Gateway
        String exchange(@Header(IntegrationMessageHeaderAccessor.CORRELATION_ID) String id,
                @Payload String out);

    }

    @Bean
    public IntegrationFlow router() {
        return IntegrationFlows.from("outbound")
                .routeToRecipients(r -> r
                        .recipient("toKafka")
                        .recipient("barrierChannel"))
                .get();
    }

    @Bean
    public IntegrationFlow outFlow(KafkaTemplate<String, String> kafkaTemplate) {
        return IntegrationFlows.from("toKafka")
                .handle(Kafka.outboundChannelAdapter(kafkaTemplate).topic(TRANSACTION_TOPIC1))
                .get();
    }

    @Bean
    public IntegrationFlow barrierFlow(BarrierMessageHandler barrier) {
        return IntegrationFlows.from("barrierChannel")
                .handle(barrier)
                .transform("payload.get(1)") // payload is list with input/reply
                .get();
    }

    @Bean
    public BarrierMessageHandler barrier() {
        return new BarrierMessageHandler(10_000L);
    }

    @KafkaListener(id = "clientReply", topics = TRANSACTION_RESULT1)
    public void result(Message<?> reply) {
        logger.info("Received reply: " + reply.getPayload() + " for id "
                + reply.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID));
        barrier().trigger(reply);
    }

    // Server side

    @KafkaListener(id = "server", topics = TRANSACTION_TOPIC1)
    public void service(String in,
            @Header(IntegrationMessageHeaderAccessor.CORRELATION_ID) String id) throws InterruptedException {
        logger.info("Service Received " + in);
        Thread.sleep(5_000);
        logger.info("Service Replying to " + in);
        // with spring-kafka 2.0 (and Boot 2), you can return a String and use @SendTo instead of this.
        this.kafkaTemplate.send(new GenericMessage<>("reply for " + in,
                Collections.singletonMap(IntegrationMessageHeaderAccessor.CORRELATION_ID, id)));
    }

    // Provision topics if needed

    // provided by Boot in 2.0
    @Bean
    public KafkaAdmin admin() {
        Map<String, Object> config = new HashMap<>();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        return new KafkaAdmin(config);
    }

    @Bean
    public NewTopic topic1() {
        return new NewTopic(TRANSACTION_TOPIC1, 10, (short) 1);
    }

    @Bean
    public NewTopic result1() {
        return new NewTopic(TRANSACTION_RESULT1, 10, (short) 1);
    }

}

Result

2018-01-20 17:27:54.668  INFO 98522 --- [   server-1-C-1] com.example.So48349993Application        : Service Received foo
2018-01-20 17:27:55.782  INFO 98522 --- [nio-8080-exec-2] com.example.So48349993Application        : Controller received: bar
2018-01-20 17:27:55.788  INFO 98522 --- [   server-0-C-1] com.example.So48349993Application        : Service Received bar
2018-01-20 17:27:59.673  INFO 98522 --- [   server-1-C-1] com.example.So48349993Application        : Service Replying to foo
2018-01-20 17:27:59.702  INFO 98522 --- [ientReply-1-C-1] com.example.So48349993Application        : Received reply: reply for foo for id 1
2018-01-20 17:27:59.705  INFO 98522 --- [nio-8080-exec-1] com.example.So48349993Application        : Controller replying: reply for foo
2018-01-20 17:28:00.792  INFO 98522 --- [   server-0-C-1] com.example.So48349993Application        : Service Replying to bar
2018-01-20 17:28:00.798  INFO 98522 --- [ientReply-0-C-1] com.example.So48349993Application        : Received reply: reply for bar for id 2
2018-01-20 17:28:00.800  INFO 98522 --- [nio-8080-exec-2] com.example.So48349993Application        : Controller replying: reply for bar

Pom

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>so48349993</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>so48349993</name>
    <description>Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.9.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-kafka</artifactId>
            <version>2.3.0.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>1.3.2.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>


</project>

应用程序属性

spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.concurrency=2

EDIT

这是一个在服务器端使用 Spring Integration 的版本，而不是@KafkaListener...

@SpringBootApplication
@RestController
public class So483499931Application {

    private static final Logger logger = LoggerFactory.getLogger(So483499931Application.class);

    private static final String TRANSACTION_TOPIC1 = "TRANSACTION_TOPIC3";

    private static final String TRANSACTION_RESULT1 = "TRANSACTION_RESULT3";

    public static void main(String[] args) {
        SpringApplication.run(So483499931Application.class, args);
    }

    private final Exchanger exchanger;

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public So483499931Application(Exchanger exchanger,
            KafkaTemplate<String, String> kafkaTemplate) {
        this.exchanger = exchanger;
        this.kafkaTemplate = kafkaTemplate;
        kafkaTemplate.setDefaultTopic(TRANSACTION_RESULT1);
    }

    @RequestMapping(path = "/foo/{id}/{other}", method = RequestMethod.GET)
    @ResponseBody
    public String foo(@PathVariable String id, @PathVariable String other) {
        logger.info("Controller received: " + other);
        String reply = this.exchanger.exchange(id, other);
        logger.info("Controller replying: " + reply);
        return reply;
    }

    // Client side

    @MessagingGateway(defaultRequestChannel = "outbound", defaultReplyTimeout = "10000")
    public interface Exchanger {

        @Gateway
        String exchange(@Header(IntegrationMessageHeaderAccessor.CORRELATION_ID) String id,
                @Payload String out);

    }

    @Bean
    public IntegrationFlow router() {
        return IntegrationFlows.from("outbound")
                .routeToRecipients(r -> r
                        .recipient("toKafka")
                        .recipient("barrierChannel"))
                .get();
    }

    @Bean
    public IntegrationFlow outFlow(KafkaTemplate<String, String> kafkaTemplate) {
        return IntegrationFlows.from("toKafka")
                .handle(Kafka.outboundChannelAdapter(kafkaTemplate).topic(TRANSACTION_TOPIC1))
                .get();
    }

    @Bean
    public IntegrationFlow barrierFlow(BarrierMessageHandler barrier) {
        return IntegrationFlows.from("barrierChannel")
                .handle(barrier)
                .transform("payload.get(1)") // payload is list with input/reply
                .get();
    }

    @Bean
    public BarrierMessageHandler barrier() {
        return new BarrierMessageHandler(10_000L);
    }

    @KafkaListener(id = "clientReply", topics = TRANSACTION_RESULT1)
    public void result(Message<?> reply) {
        logger.info("Received reply: " + reply.getPayload() + " for id "
                + reply.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID));
        barrier().trigger(reply);
    }

    // Server side

    @Bean
    public IntegrationFlow server(ConsumerFactory<String, String> consumerFactory,
            KafkaTemplate<String, String> kafkaTemplate) {
        return IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(consumerFactory, TRANSACTION_TOPIC1))
            .handle("so483499931Application", "service")
            .handle(Kafka.outboundChannelAdapter(kafkaTemplate).topic(TRANSACTION_RESULT1))
            .get();
    }

    public String service(String in) throws InterruptedException {
        logger.info("Service Received " + in);
        Thread.sleep(5_000);
        logger.info("Service Replying to " + in);
        return "reply for " + in;
    }

    // Provision topics if needed

    // provided by Boot in 2.0
    @Bean
    public KafkaAdmin admin() {
        Map<String, Object> config = new HashMap<>();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        return new KafkaAdmin(config);
    }

    @Bean
    public NewTopic topic1() {
        return new NewTopic(TRANSACTION_TOPIC1, 10, (short) 1);
    }

    @Bean
    public NewTopic result1() {
        return new NewTopic(TRANSACTION_RESULT1, 10, (short) 1);
    }

}

and

spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.concurrency=2
spring.kafka.consumer.group-id=server