时间戳

首页 > 解决方案 > 消息未传递给 Kafka 主题

java - 消息未传递给 Kafka 主题

问题描述

我正在开发一个简单的应用程序，使用 spring boot 和 Kafka 将对象保存到 Kafka 主题，然后有一个消费者会监听并将该记录添加到内存数据库中。在以下配置之后，Kafka 和 Spring Boot 都可以正常工作，但是对 Kafka 的 API 调用会挂起一段时间，并且 Kafka 主题中也不会有任何消息。

我按照下面的教程

https://www.confluent.io/blog/apache-kafka-spring-boot-application/

但我稍微改变了我的服务。

package com.dilshan.shoppingcart.cart;

import org.springframework.stereotype.Service;

import java.io.IOException;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;

@Service
public class ShoppingService {
    private static final String TOPIC = "users";

    @Autowired
    private KafkaTemplate<String, ShoppingCart> kafkaTemplate;

    private ShoppingCartRepository cartRepository;

    public void publish(ShoppingCart cart) {
        this.kafkaTemplate.send(TOPIC, cart);
    }

    @KafkaListener(topics = "users", groupId = "group_id")
    public void subscribe(ShoppingCart cart) throws IOException {
        System.out.println("Hello-"+cart.getId());
        cartRepository.save(cart);

    }

}

这是我的 application.properties 文件

spring.datasource.url=jdbc:h2:mem:testdb
spring.datasource.driverClassName=org.h2.Driver
spring.datasource.username=sa
spring.datasource.password=
spring.h2.console.enabled=true
spring.h2.console.path=/h2
server.port=8080
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=group_id
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.listener.missing-topics-fatal=false

这是我用来访问 kafka 方法的控制器。

package com.dilshan.shoppingcart.cart;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class ShoppingCartController {

    @Autowired
    private ShoppingService shoppingService;

    @RequestMapping(method = RequestMethod.POST, value = "/shoppingcart")
    private ResponseEntity<ShoppingCart> selectProducts( @RequestBody ShoppingCart cart) {
        shoppingService.publish(cart);
        return ResponseEntity.ok().build();
    }

}

以下日志不断显示在应用程序日志中。

2020-02-18 19:22:12.951[0;39m [33m WARN[0;39m [35m6986[0;39m [2m---[0;39m [2m[ntainer#0-0-C-1][0;39m [36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-1, groupId=group_id] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
[2m2020-02-18 19:22:13.958[0;39m [33m WARN[0;39m [35m6986[0;39m [2m---[0;39m [2m[ntainer#0-0-C-1][0;39m [36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-1, groupId=group_id] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
[2m2020-02-18 19:22:15.115[0;39m [33m WARN[0;39m [35m6986[0;39m [2m---[0;39m [2m[ntainer#0-0-C-1][0;39m [36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-1, groupId=group_id] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
[2m2020-02-18 19:22:16.272[0;39m [33m WARN[0;39m [35m6986[0;39m [2m---[0;39m [2m[ntainer#0-0-C-1][0;39m [36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-1, groupId=group_id] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

标签： javaspring-bootapache-kafka

解决方案

Kafka 代理，有时称为 Kafka 服务器，应该在您的 PC 或远程机器上运行。

请检查此链接 https://kafka.apache.org/quickstart

首先，您需要运行 Zookeeper 和 Kafka 代理，然后您可以使用您的应用程序发送消息。

推荐阅读