执行器指标中未显示Spring Cloud stream kafka指标|Springboot2.2.2|SpringCloud Hoxton. SR4

Question

我没有看到使用执行器的kafka消费者/生产者指标 /actuator/metrics.只显示了基本的jvm指标。

{
"names": [
"jvm.buffer.memory.used",
"jvm.threads.states",
"jvm.memory.committed",
"spring.integration.channels",
"process.uptime",
"jvm.memory.used",
"jvm.gc.pause",
"logback.events",
"http.server.requests",
"jvm.memory.max",
"tomcat.sessions.active.current",
"jvm.buffer.total.capacity",
"system.cpu.usage",
"jvm.threads.live",
"jvm.classes.unloaded",
"jvm.classes.loaded",
"jvm.threads.peak",
"tomcat.sessions.rejected",
"tomcat.sessions.alive.max",
"jvm.gc.memory.promoted",
"jvm.buffer.count",
"jvm.gc.memory.allocated",
"tomcat.sessions.expired",
"tomcat.sessions.created",
"jvm.gc.max.data.size",
"system.cpu.count",
"spring.integration.handlers",
"spring.integration.sources",
"process.start.time",
"jvm.threads.daemon",
"tomcat.sessions.active.max",
"jvm.gc.live.data.size",
"process.cpu.usage"
]
}

我的应用程序工作正常，绑定看起来也可以，在 /actuator/bindings.可见

[
{
"bindingName": "send-kafka-out",
"name": "test.topic",
"group": null,
"pausable": false,
"state": "running",
"extendedInfo": {
"bindingDestination": "test.topic",
"ExtendedProducerProperties": {
"autoStartup": true,
"partitionCount": 1,
"headerMode": "headers",
"extension": {
"bufferSize": 16384,
"compressionType": "none",
"sync": false,
"sendTimeoutExpression": null,
"batchTimeout": 0,
"messageKeyExpression": null,
"headerPatterns": null,
"configuration": {},
"topic": {
"replicationFactor": null,
"replicasAssignments": {},
"properties": {}
},
"useTopicHeader": false,
"recordMetadataChannel": null
},
"validPartitionSelectorProperty": true,
"validPartitionKeyProperty": true
}
},
"input": false
}
]

下面是我的流配置

public interface KafkaStreams {
    String OUTPUT = "send-kafka-out";

    @Output(OUTPUT)
    MessageChannel getOutboundMessageChannel();

}

@SpringBootApplication
@EnableDiscoveryClient
@EnableJms
@EnableBinding(KafkaStreams.class)
@ComponentScan("com.test.connector.*")
public class ConnectorApplication {

    public static void main(String[] args) {
        SpringApplication.run(ConnectorApplication.class, args);
    }

}

下面是我的配置。yml公司

spring:
  jms:
    listener:
      auto-startup: true
      acknowledge-mode: AUTO
      max-concurrency: 10
      concurrency: 10
  cloud:
    stream:
      default-binder: kafka-default
      function:
        definition: 
      kafka:
        binder:
          brokers:
            - localhost:9092
          configuration:
            default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
            default.value.serde: org.apache.kafka.common.serialization.Serdes$BytesSerde
            commit.interval.ms: 1000
            #security:
            #  protocol: SASL_PLAINTEXT

      bindings:
        send-kafka-out:
          destination: test.topic
          contentType: text/plain
          group: output-group-1
          binder: kafka-default
          producer:
            header-mode: headers

      binders:
        kafka-default:
         type: kafka
         environment:
           spring:
             cloud:
               stream:
                kafka:
                  binder:
                    brokers: localhost:9092

我使用的是SpringBoot 2.2.2。发布，SpringCloud Hoxton。SR4和spring cloud stream活页夹kafka 3.0.4。释放。任何关于我缺少什么的建议都会有所帮助。

Answer 1

您需要有输入绑定。Kafka活页夹指标仅支持消费者。看起来代码中只有一个输出绑定。我刚刚尝试了一个有消费者的快速应用程序，我能够检索到指标：

执行机构/指标/Spring。云流动粘合剂Kafka。偏移量

{"name"："spring.cloud.stream.binder.kafka.offset"，"描述"："特定组和主题的未消耗消息"，"base Unit"： null，"测量"：[{"统计"："值"，"值"： 0.0}]，"可用标签"：[{"tag"："主题"，"值"：["uppercase-in-0"]}，{"标签"："组"，"值"：["匿名.06730d87-0871-4b82-bd82-ae4958dbe2a3"]}]}