org.springframework.kafka.listener.listenerexecutionfailedexception:无法使用传入消息调用侦听器方法

luaexgnf 于 2021-06-06 发布在 Kafka

关注(0)|答案(1)|浏览(1053)

我是apachekafka新手，能够从发送者发送消息（json格式），但不能在消费者服务中使用。

这是我的密码：

发件人服务

@Service
public class SenderService {
    private static final Logger LOG = LoggerFactory.getLogger(SenderService.class);

    @Autowired
    private KafkaTemplate<String, IdName> kafkaTemplate;

    @Value("${app.topic.email}")
    private String topic;

    public void send(IdName idName) {
        LOG.info("Sending Data='{}' to topic='{}' ", idName, topic);

        Message<IdName> message = MessageBuilder.withPayload(idName).setHeader(KafkaHeaders.TOPIC, topic)
            .setHeader(KafkaHeaders.MESSAGE_KEY, "TestMessage")
            .build();
        kafkaTemplate.send(message);
    }
}

### Consumer Service

@Service
public class ConsumerService {
    private static final Logger LOG = LoggerFactory.getLogger(ConsumerService.class);

    @Autowired
    MailSender mailSender;

    @KafkaListener(topics = "${app.topic.email}")
    public void receive(@Payload IdName data,
        @Header MessageHeaders headers) throws Exception{
        LOG.info("Received data='{}'", data);
    }
 }

我收到以下例外情况

2018-09-21 11:01:41.738 ERROR 63487 --- [ntainer#0-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: ConsumerRecord(topic = emailclient, partition = 0, offset = 4, CreateTime = 1537507901660, serialized key size = 11, serialized value size = 122, headers = RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 98, 99, 112, 108, 117, 115, 100, 46, 97, 112, 112, 108, 105, 99, 97, 116, 105, 111, 110, 46, 100, 111, 109, 97, 105, 110, 46, 73, 100, 78, 97, 109, 101])], isReadOnly = false), key = TestMessage, value = com.*****.****.domain.IdName@4a2d3006)

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.*****.****.service.ConsumerService.receive(com.*****.****.domain.IdName,org.springframework.messaging.MessageHeaders) throws java.lang.Exception]

Bean [com.*****.****.service.ConsumerService@4649d70a]; nested exception is org.springframework.messaging.MessageHandlingException: Missing header 'headers' for method parameter type [class org.springframework.messaging.MessageHeaders], failedMessage=GenericMessage [payload=com.*****.****.domain.IdName@4a2d3006, headers={kafka_offset=4, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@53fdf4bb, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=TestMessage, kafka_receivedPartitionId=0, kafka_receivedTopic=emailclient, kafka_receivedTimestamp=1537507901660, __TypeId__=[B@707343c2}]

有人能帮我吗？

Java apache-kafka

来源：https://stackoverflow.com/questions/52437219/org-springframework-kafka-listener-listenerexecutionfailedexception-listener-me

1条答案

按热度按时间

kmb7vmvb1#

如果您想访问侦听器中的所有头，那么您使用的注解是错误的，应该是 @Headers 而不是 @Header . 生成的代码如下所示：

@Service
public class ConsumerService {
private static final Logger LOG = LoggerFactory.getLogger(ConsumerService.class);

    @Autowired
    MailSender mailSender;

    @KafkaListener(topics = "${app.topic.email}")
    public void receive(@Payload IdName data,
        @Headers MessageHeaders headers) throws Exception{
        LOG.info("Received data='{}'", data);
    }
 }

当然，您也只能注入消息键，如下所示：

@Service
public class ConsumerService {
private static final Logger LOG = LoggerFactory.getLogger(ConsumerService.class);

    @Autowired
    MailSender mailSender;

    @KafkaListener(topics = "${app.topic.email}")
    public void receive(@Payload IdName data,
        @Header(KafkaHeaders.MESSAGE_KEY) String messageKey) throws Exception{
        LOG.info("Received data='{}'", data);
    }
 }

赞(0）回复(0）举报 2021-06-07

我来回答

org.springframework.kafka.listener.listenerexecutionfailedexception:无法使用传入消息调用侦听器方法

这是我的密码：

发件人服务

1条答案

相关问题

热门标签

最新问答