spring cloud producer exception-java.lang.illegalstateexception:关闭producer后无法执行操作

efzxgjgh  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(839)

spring云生产者引发异常-java.lang.illegalstateexception:生产者关闭后无法执行操作
嗨,我们给出了一个基于spring云的微服务应用程序。在该应用程序中,我们在向kafka生成消息时出错。我们正在使用下面的spring云版本。
Spring Cloud版
格林威治.sr1
以下是Kafka制作人使用的接口定义

String INPUT = "jobmanager-in";
String OUTPUT_C = "collection-out";
String OUTPUT_P = "parser-out";
String OUTPUT_CMP = "compare-out";
String OUTPUT_R = "report-out";
String OUTPUT_N = "notification-out";

@Input(INPUT)
SubscribableChannel inboundJobManager();

@Output(OUTPUT_C)
    MessageChannel outboundCollections();

@Output(OUTPUT_P)
MessageChannel outboundParse();

@Output(OUTPUT_CMP)
MessageChannel outboundCompare();

@Output(OUTPUT_R)
MessageChannel outboundReport();

@Output(OUTPUT_N)
MessageChannel outboundNotification();

生产商代码是

public void sendCollectionTask(final MessageT message) {

        logger.info("Sending Collection Task :: " + message.toString());
        MessageChannel messageChannel = collectionStream.outboundCollections();
        boolean l_bool = messageChannel.send(MessageBuilder.withPayload(message)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build());
        logger.info("After sending message for Collection Task send status :: " + message.getTask().getId() + " : "
                + l_bool);
    }

当我们启动应用程序时,生产者会正确地生成消息,但过了一段时间后,我们会收到错误

2019-09-12 02:07:45,215 ERROR com.ericsson.tmo.cm.ccm.jobmanager.util.JobOrchestrator [scheduling-1] Error Trace ::
org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$ProducerConfigurationMessageHandler@11328ab9]; nested exception is java.lang.IllegalStateException: Cannot perform operation after producer has been closed
        at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:189)
        at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:179)

如果我们重新启动应用程序,问题就解决了。谢谢你的帮助。
我们的 Spring 云配置是

cloud:
    stream:
      kafka:
        binder:
          brokers: kafka
          autoCreateTopic: false
          configuration:
            auto.offset.reset: latest
            max.request.size: 16777216
            buffer.memory: 67108864
      bindings:
        collection-out:
          destination: collection
          contentType: application/json
          group: collection
          producer:
            partitionCount: 5
            autoAddPartitions: true
c0vxltue

c0vxltue1#

例外信息很清楚。
生产者关闭后无法执行操作
可能会发生这样的情况:生产者可能被一个线程关闭,而生产者正在调用另一个线程 send() 或者生产者网络线程仍在发送消息。
既然你用的是Spring,我想 KafkaProducer 对象是为您创建的,并通过依赖注入注入。
你所要做的就是找出谁在用同样的东西 KafkaProducer 对象以及 close() 被称为。
可能的原因
有时候,也会发生同样的事情 KafkaProducer 可能由同一应用程序的其他几个组件使用(如果您没有在多个生产者之间进行划分),其中一个可能已经关闭了它。
更清楚的是,您可能正在使用同一个producerbean示例为同一示例中的不同组件生成消息,而在某个地方它是关闭的。
通常情况下,我们有不同的生产者生产不同的主题,因为我们有不同的生产者配置和不同类型的数据主题。
我在添加调用的shutdown钩子时遇到了类似的异常 producer.close() 而另一个线程正在尝试生成。

相关问题