Kakfa Producer - Spring Boot应用程序 - 无法生成消息

Question

我也是刚刚学习spring boot和kakfa。我探索了一下，并配置了一个示例生产者应用，如下图。但是我无法发布消息。如果能得到帮助就更好了，因为我在这里错过了什么。我已经启动了zookeeper服务和kakfa服务，并确保主题是可用的。

Config:

    import java.util.HashMap;
    import java.util.Map;

    import com.jpmorgan.sample.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.common.serialization.StringSerializer;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.kafka.core.DefaultKafkaProducerFactory;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.kafka.core.ProducerFactory;

    @Configuration
    public class KafkaConfig {

        @Value("${kafka.bootstrap-servers}")
        private String bootstrapServers;

        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<>();
            // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    bootstrapServers);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                    StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                    StringSerializer.class);

            return props;
        }

        @Bean
        public ProducerFactory<String, String> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }

        @Bean
        public KafkaTemplate<String, String> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }

        @Bean
        public KafkaProducer sender() {
            return new KafkaProducer();
        }
    }


Producer Class:
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

public class KafkaProducer {

    private static final Logger LOGGER =
            LoggerFactory.getLogger(KafkaProducer.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void send(String payload) {
        LOGGER.info("sending payload='{}'", payload);
        kafkaTemplate.send("test", payload);
    }
}

Sample Application Class:
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaProducerSampleApplication {

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

}

我只是通过IntelliJ运行。我看到moneta启动已经开始了。

         main] n$WebApplicationLoggingAutoConfiguration : Enabled Moneta Request Logging with exclude-url-patterns: [], http.log-request-headers: [false], http.log-response-headers: [false], http.log-request-entity: [false], http.log-response-entity: [false] and http.max-entity-bytes: [1024]
2020-05-06 18:51:57.539  INFO 4952 --- [           main] c.j.m.b.a.s.cors.CorsAutoConfiguration   : Moneta CORS has been disabled because neither [moneta.cors.allowed-origins] nor [moneta.cors.allowed-origins-regex] has been set
2020-05-06 18:51:57.628  INFO 4952 --- [           main] .m.b.a.a.MonetaActuatorAutoConfiguration : Enabled Moneta defaults for Spring Boot Actuator
2020-05-06 18:51:57.677  INFO 4952 --- [           main] o.s.boot.web.servlet.RegistrationBean    : Filter corsFilter was not registered (disabled)
2020-05-06 18:51:57.860  INFO 4952 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-05-06 18:51:57.956  WARN 4952 --- [           main] c.j.m.b.startup.ApplicationInfoLoader    : !!!! SEAL ID property [application.seal.id] should be provided as either as a system property or in the application properties file !!!!
2020-05-06 18:51:58.054  INFO 4952 --- [           main] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 4 endpoint(s) beneath base path '/actuator'
2020-05-06 18:51:58.111  INFO 4952 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
2020-05-06 18:51:58.114  INFO 4952 --- [           main] c.j.s.KafkaProducerSampleApplication     : Started KafkaProducerSampleApplication in 2.547 seconds (JVM running for 4.376)

Answer 1

试试这种方式，你不需要在生产者配置中使用@EnableKafka注释

从producerConfigs中删除@Bean，然后再试一下，我就可以了。

    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }