如何使用 Spring-Cloud-Stream-Binder-Kafka 将"Kafka" -消息打印到控制台

Question

我的问题非常接近打印kafka流输入到控制台？有点差异：我使用的是streamsbuilder而不是KStreamBuilder。(不幸的是，由于404误差，没有分辨率的Git。(

我的听众是：

    @StreamListener(LoansStreams.INPUT)
    public void handleLoans(@Payload Loans loans) {

        final Serde<String> stringSerde = Serdes.String();
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream(stringSerde, stringSerde, "in-stream");
        source.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                ArrayList<String> keywords = new ArrayList<String>();
                // apply regex to value and for each match add it to keywords
                return keywords;
            }
        }
    }

我遇到的错误是"类型流构建器中的方法流(字符串，消耗(不适用于参数(Serde，Serde，Serde，String("，显然是因为它是KStreamBuilder的一部分。

。

假设我正确理解，我必须贴上收到的消息，但是我该怎么做？我想补充一点，这是我第一次使用kafka-cloud-stream。

我的目标是创建一个具有两个不同微服务的POC，访问不同的数据库，并通过Kafka"连接"两个微服务，以应用传奇模式(或至少开始应用其原理(。

。

如果重要的话，我启动了Kafka，只有要打印的消息：

    #log4j.rootLogger=WARN, stderr
    log4j.rootLogger=OFF, stdout 
    log4j.appender.stderr=org.apache.log4j.ConsoleAppender
    log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
    log4j.appender.stderr.layout.ConversionPattern=[%d] %p %m (%c)%n
    #log4j.appender.stderr.Target=System.err

从不打印Kafka-Console-Consumer警告

另外，基于keeperErrorCode =/admin/preferred_replica_election的nonode，我忽略了此zookeeper例外：

[2019-04-30 14:15:11,617] INFO Got user-level KeeperException when processing sessionid:0x100003793980001 type:multi cxid:0x71 zxid:0xbc txntype:-1 reqpath:n/a aborting remaining multi ops. Error Path:/admin/preferred_replica_election Error:KeeperErrorCode = NoNode for /admin/preferred_replica_election (org.apache.zookeeper.server.PrepRequestProcessor)

最后但并非最不重要的一点是，我的发件人是：

public void sendLoan(final Loans loans) {
    log.info("Sending loans {}", loans);
    MessageChannel messageChannel = loansStreams.outboundLoans();
    messageChannel.send(MessageBuilder
            .withPayload(loans)
            .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
            .build());
}

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>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
        <relativePath /> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.mybank</groupId>
    <artifactId>kafka-cloud-stream</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka-cloud-stream</name>
    <description>Spring Cloud Stream With Kafka</description>
    <properties>
        <java.version>11</java.version>
        <spring-cloud.version>Greenwich.SR1</spring-cloud.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-test-support</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.springframework/spring-web -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
            <!-- version>5.1.5.RELEASE</version -->
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

Answer 1

KStreamBuilder是由 StreamsBuilder替换的旧建筑商类(在apache kafka 1.0.0版本中(。不仅更改了名称，还更改了方法参数。

正如错误消息指示的那样，新方法期望stream(String, Consumed)而不是旧的stream(Serde, Serde, String)。

比较相应的升级指南：https：//docs.confluent.io/4.0.0/streams/upgrade-guide.html#building-and-and-rinning-running-running-a-topology