spring-kafka - 消费者未收到 Spring Cloud 流绑定器 Kafka 自定义标头

问题描述

我正在使用 Spring Cloud Stream binder kafka，Edgware.SR4 版本。

我已将自定义标头设置为消息有效负载并将其发布，但我在消费者端看不到这些标头。

我使用 Message 对象来绑定有效负载和标头。我尝试添加属性 spring.cloud.stream.kafka.binder.headers 但它不起作用

制片人：

应用程序.yml

spring:
  cloud:
    stream:
      bindings:
        sampleEvent:
          destination: sample-event
          content-type: application/json
      kafka:
        binder:
          brokers: localhost:9092
          zkNodes: localhost:2181
          autoCreateTopics: false
          zkConnectionTimeout: 36000

MessageChannelConstants.java

public class MessageChannelConstants {
    public static final String SAMPLE_EVENT = "sampleEvent";
        private MessageChannelConstants() {}
}

SampleMessageChannels.java

public interface SampleMessageChannels {

    @Output(MessageChannelConstants.SAMPLE_EVENT)
    MessageChannel sampleEvent();
}

SampleEventPublisher.java

@Service
@EnableBinding(SampleMessageChannels.class)
public class SampleEventPublisher{

@Autowired
private SampleMessageChannels sampleMessageChannels;

public void publishSampleEvent(SampleEvent sampleEvent) {
        final Message<SampleEvent> message = MessageBuilder.withPayload(sampleEvent).setHeader("appId", "Demo").build();
    MessageChannel messageChannel = SampleMessageChannels.sampleEvent();
    if (messageChannel != null) {
        messageChannel.send(message);
    }
}
}

消费者：

应用程序.yml

spring:
  cloud:
    stream:
      bindings:
        sampleEvent:
          destination: sample-event
          content-type: application/json
      kafka:
        binder:
          brokers: localhost:9092
          zkNodes: localhost:2181
          autoCreateTopics: false
          zkConnectionTimeout: 36000

MessageChannelConstants.java

public class MessageChannelConstants {

    public static final String SAMPLE_EVENT = "sampleEvent";

    private MessageChannelConstants() {}
}

SampleMessageChannels.java

public interface SampleMessageChannels {

    @Output(MessageChannelConstants.SAMPLE_EVENT)
    MessageChannel sampleEvent();
}

SampleEventListener.java

@Service
@EnableBinding(SampleMessageChannels.class)
public class SampleEventListener{

@StreamListener(MessageChannelConstants.SAMPLE_EVENT)
public void listenSampleEvent(@Payload SampleEvent event,
    @Header(required = true, name = "appId") String appId) {

     // do something 
}

下面是我得到的例外，

org.springframework.messaging.MessageHandlingException: Missing header 'appId' for method parameter type [class java.lang.String]
    at org.springframework.messaging.handler.annotation.support.HeaderMethodArgumentResolver.handleMissingValue(HeaderMethodArgumentResolver.java:100)
    at org.springframework.messaging.handler.annotation.support.AbstractNamedValueMethodArgumentResolver.resolveArgument(AbstractNamedValueMethodArgumentResolver.java:103)
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:112)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:135)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:107)
    at org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:55)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:425)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:375)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:360)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
    at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:70)
    at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:64)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:188)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$200(KafkaMessageDrivenChannelAdapter.java:63)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:372)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:352)
    at org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter$1.doWithRetry(RetryingAcknowledgingMessageListenerAdapter.java:79)
    at org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter$1.doWithRetry(RetryingAcknowledgingMessageListenerAdapter.java:73)
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180)
    at org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter.onMessage(RetryingAcknowledgingMessageListenerAdapter.java:73)
    at org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter.onMessage(RetryingAcknowledgingMessageListenerAdapter.java:39)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:792)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:736)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.access$2100(KafkaMessageListenerContainer.java:246)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer$ListenerInvoker.run(KafkaMessageListenerContainer.java:1025)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.lang.Thread.run(Thread.java:748)

注意：我也在使用 spring cloud sleuth 和 zipkin 依赖项。

标签： spring-kafkaspring-cloud-stream

spring-kafka - 消费者未收到 Spring Cloud 流绑定器 Kafka 自定义标头

问题描述

解决方案

推荐阅读