如何在 spring 集成中设置 KafkaHeaders.MESSAGE_KEY

Question

谁能帮我为下面的代码设置随机KafkaHeaders.MESSAGE_KEY，以便我可以将有效负载发布到不同的分区。

@Bean
public IntegrationFlow fileInboundChannelFlow() {
    FileInboundChannelAdapterSpec messageSourceSpec = Files
            .inboundAdapter(Paths.get(this.properties.getDirectory()).toFile());

    messageSourceSpec = messageSourceSpec.filter(getFilter());
    //messageSourceSpec.regexFilter(this.properties.getFilenameRegex());
    messageSourceSpec.preventDuplicates(this.properties.isPreventDuplicates());

    IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(messageSourceSpec)
            .split(new FileSplitter(true, true))
            .enrichHeaders(h -> h.header(KafkaHeaders.MESSAGE_KEY, "payload.flightNumber"));

    return flowBuilder.<Object, Class<?>>route(Object::getClass,
            m -> m.channelMapping(FileSplitter.FileMarker.class, "markers.input").channelMapping(String.class,
                    "lines.input"))
            .get();
}

我想根据 payload.prod_cd 设置密钥。 由于它是单例 bean，我想为每个有效负载初始化不同的 kafka 标头键。

Answer 1

你在正确的轨道上：

.enrichHeaders(h -> h.header(KafkaHeaders.MESSAGE_KEY, "payload.flightNumber"));

您只需要改用headerExpression()：

/**
 * Add a single header specification where the value is a String representation of a
 * SpEL {@link Expression}. If the header exists, it will <b>not</b> be overwritten
 * unless {@link #defaultOverwrite(boolean)} is true.
 * @param name the header name.
 * @param expression the expression.
 * @return the header enricher spec.
 */
public HeaderEnricherSpec headerExpression(String name, String expression) {