集成人员定位Kafka消息订阅

start/src/main/resources/bootstrap.yml

@@ -3,5 +3,6 @@ spring:
     import:
 #      - classpath:nacos-prod.yml
       - classpath:nacos.yml
+      - classpath:kafka.yml
       - classpath:sdk.yml
       - classpath:swagger.yml

start/src/main/resources/kafka.yml

@@ -0,0 +1,25 @@
+spring:
+  kafka:
+    bootstrap-servers: ${PERSONNEL_POSITIONING_KAFKA_BOOTSTRAP_SERVERS:192.168.193.23:9092}
+    consumer:
+      auto-offset-reset: latest
+      enable-auto-commit: false
+      key-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
+      max-poll-records: 20
+      properties:
+        allow.auto.create.topics: false
+        security.protocol: ${PERSONNEL_POSITIONING_KAFKA_SECURITY_PROTOCOL:PLAINTEXT}
+        sasl.mechanism: ${PERSONNEL_POSITIONING_KAFKA_SASL_MECHANISM:PLAIN}
+    listener:
+      ack-mode: manual
+      missing-topics-fatal: false
+      idle-event-interval: ${PERSONNEL_POSITIONING_KAFKA_IDLE_EVENT_INTERVAL:60s}
+
+personnel-positioning:
+  kafka:
+    location:
+      enabled: ${PERSONNEL_POSITIONING_KAFKA_LOCATION_ENABLED:false}
+      topic: ${PERSONNEL_POSITIONING_KAFKA_LOCATION_TOPIC:point_push}
+      group-id: ${PERSONNEL_POSITIONING_KAFKA_LOCATION_GROUP_ID:personnel-position-location-probe}
+      max-log-bytes: ${PERSONNEL_POSITIONING_KAFKA_LOCATION_MAX_LOG_BYTES:16384}

web-app/pom.xml

@@ -21,6 +21,10 @@
             <groupId>com.zcloud.personnel.positioning</groupId>
             <artifactId>web-infrastructure</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>

web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/KafkaConsumerLifecycleLogger.java

@@ -0,0 +1,187 @@
+package com.zcloud.personnel.positioning.integration.kafka;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.event.EventListener;
+import org.springframework.kafka.event.ConsumerFailedToStartEvent;
+import org.springframework.kafka.event.ConsumerStartedEvent;
+import org.springframework.kafka.event.ConsumerStartingEvent;
+import org.springframework.kafka.event.ConsumerStoppedEvent;
+import org.springframework.kafka.event.KafkaEvent;
+import org.springframework.kafka.event.ListenerContainerIdleEvent;
+import org.springframework.kafka.event.NonResponsiveConsumerEvent;
+import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
+import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.stereotype.Component;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+@Component
+@ConditionalOnProperty(
+        prefix = "personnel-positioning.kafka.location",
+        name = "enabled",
+        havingValue = "true"
+)
+public class KafkaConsumerLifecycleLogger implements ConsumerAwareRebalanceListener {
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumerLifecycleLogger.class);
+    private static final String LOCATION_LISTENER_ID = "personnelPositionLocationProbeListener";
+
+    private final String bootstrapServers;
+    private final String securityProtocol;
+    private final String saslMechanism;
+
+    public KafkaConsumerLifecycleLogger(
+            @Value("${spring.kafka.bootstrap-servers}") String bootstrapServers,
+            @Value("${spring.kafka.consumer.properties.security.protocol:PLAINTEXT}") String securityProtocol,
+            @Value("${spring.kafka.consumer.properties.sasl.mechanism:PLAIN}") String saslMechanism) {
+        this.bootstrapServers = bootstrapServers;
+        this.securityProtocol = securityProtocol;
+        this.saslMechanism = saslMechanism;
+    }
+
+    @EventListener
+    public void onConsumerStarting(ConsumerStartingEvent event) {
+        MessageListenerContainer container = locationContainer(event);
+        if (container == null) {
+            return;
+        }
+        LOGGER.info(
+                "Kafka定位消息消费者开始启动，listenerId={}, broker={}, topics={}, groupId={}, "
+                        + "securityProtocol={}, saslMechanism={}",
+                container.getListenerId(),
+                bootstrapServers,
+                Arrays.toString(container.getContainerProperties().getTopics()),
+                container.getGroupId(),
+                securityProtocol,
+                saslMechanism
+        );
+    }
+
+    @EventListener
+    public void onConsumerStarted(ConsumerStartedEvent event) {
+        MessageListenerContainer container = locationContainer(event);
+        if (container != null) {
+            LOGGER.info(
+                    "Kafka定位消息消费者已启动，等待连接Broker并分配分区，"
+                            + "listenerId={}, groupId={}",
+                    container.getListenerId(),
+                    container.getGroupId()
+            );
+        }
+    }
+
+    @Override
+    public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
+        LOGGER.info(
+                "Kafka定位消息订阅成功，broker={}, groupId={}, partitions={}",
+                bootstrapServers,
+                consumer.groupMetadata().groupId(),
+                partitions
+        );
+    }
+
+    @Override
+    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
+        LOGGER.warn(
+                "Kafka定位消息分区已回收，groupId={}, partitions={}",
+                consumer.groupMetadata().groupId(),
+                partitions
+        );
+    }
+
+    @Override
+    public void onPartitionsLost(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
+        LOGGER.error(
+                "Kafka定位消息分区意外丢失，groupId={}, partitions={}",
+                consumer.groupMetadata().groupId(),
+                partitions
+        );
+    }
+
+    @EventListener
+    public void onConsumerFailedToStart(ConsumerFailedToStartEvent event) {
+        MessageListenerContainer container = locationContainer(event);
+        if (container != null) {
+            LOGGER.error(
+                    "Kafka定位消息消费者启动失败，listenerId={}, broker={}, groupId={}",
+                    container.getListenerId(),
+                    bootstrapServers,
+                    container.getGroupId()
+            );
+        }
+    }
+
+    @EventListener
+    public void onConsumerStopped(ConsumerStoppedEvent event) {
+        MessageListenerContainer container = locationContainer(event);
+        if (container == null) {
+            return;
+        }
+        if (ConsumerStoppedEvent.Reason.NORMAL.equals(event.getReason())) {
+            LOGGER.info(
+                    "Kafka定位消息消费者已正常停止，listenerId={}, groupId={}",
+                    container.getListenerId(),
+                    container.getGroupId()
+            );
+            return;
+        }
+        LOGGER.error(
+                "Kafka定位消息消费者异常停止，listenerId={}, broker={}, groupId={}, reason={}",
+                container.getListenerId(),
+                bootstrapServers,
+                container.getGroupId(),
+                event.getReason()
+        );
+    }
+
+    @EventListener
+    public void onNonResponsiveConsumer(NonResponsiveConsumerEvent event) {
+        if (!isLocationListener(event.getListenerId())) {
+            return;
+        }
+        LOGGER.warn(
+                "Kafka定位消息消费者无响应，listenerId={}, timeSinceLastPollMs={}, partitions={}",
+                event.getListenerId(),
+                event.getTimeSinceLastPoll(),
+                event.getTopicPartitions()
+        );
+    }
+
+    @EventListener
+    public void onIdle(ListenerContainerIdleEvent event) {
+        if (!isLocationListener(event.getListenerId())) {
+            return;
+        }
+        Collection<TopicPartition> partitions = event.getTopicPartitions();
+        if (partitions == null || partitions.isEmpty()) {
+            LOGGER.warn(
+                    "Kafka定位消息订阅尚未分配分区，listenerId={}, broker={}, idleMs={}",
+                    event.getListenerId(),
+                    bootstrapServers,
+                    event.getIdleTime()
+            );
+            return;
+        }
+        LOGGER.info(
+                "Kafka定位消息订阅正常，但暂未收到新消息，listenerId={}, "
+                        + "idleMs={}, partitions={}",
+                event.getListenerId(),
+                event.getIdleTime(),
+                partitions
+        );
+    }
+
+    private MessageListenerContainer locationContainer(KafkaEvent event) {
+        MessageListenerContainer container = event.getContainer(MessageListenerContainer.class);
+        return container != null && isLocationListener(container.getListenerId()) ? container : null;
+    }
+
+    private boolean isLocationListener(String listenerId) {
+        return listenerId != null && listenerId.startsWith(LOCATION_LISTENER_ID);
+    }
+}

web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaPayloadPreview.java

@@ -0,0 +1,41 @@
+package com.zcloud.personnel.positioning.integration.kafka;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Base64;
+
+final class LocationKafkaPayloadPreview {
+
+    private LocationKafkaPayloadPreview() {
+    }
+
+    static String render(byte[] payload, int maxBytes) {
+        if (payload == null) {
+            return "<null>";
+        }
+        int previewLength = Math.min(payload.length, Math.max(1, maxBytes));
+        byte[] preview = Arrays.copyOf(payload, previewLength);
+        String suffix = payload.length > previewLength
+                ? " (truncated, totalBytes=" + payload.length + ")"
+                : "";
+        try {
+            String text = StandardCharsets.UTF_8.newDecoder()
+                    .onMalformedInput(CodingErrorAction.REPORT)
+                    .onUnmappableCharacter(CodingErrorAction.REPORT)
+                    .decode(ByteBuffer.wrap(preview))
+                    .toString();
+            return "utf8:" + oneLine(text) + suffix;
+        } catch (CharacterCodingException ignored) {
+            return "base64:" + Base64.getEncoder().encodeToString(preview) + suffix;
+        }
+    }
+
+    private static String oneLine(String value) {
+        return value.replace("\r", "\\r")
+                .replace("\n", "\\n")
+                .replace("\t", "\\t");
+    }
+}

web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaProbeListener.java

@@ -0,0 +1,63 @@
+package com.zcloud.personnel.positioning.integration.kafka;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.StringJoiner;
+
+@Component
+@ConditionalOnProperty(
+        prefix = "personnel-positioning.kafka.location",
+        name = "enabled",
+        havingValue = "true"
+)
+public class LocationKafkaProbeListener {
+    private static final Logger LOGGER = LoggerFactory.getLogger(LocationKafkaProbeListener.class);
+
+    private final int maxLogBytes;
+
+    public LocationKafkaProbeListener(
+            @Value("${personnel-positioning.kafka.location.max-log-bytes:16384}") int maxLogBytes) {
+        this.maxLogBytes = Math.max(1, maxLogBytes);
+    }
+
+    @KafkaListener(
+            id = "personnelPositionLocationProbeListener",
+            topics = "${personnel-positioning.kafka.location.topic:point_push}",
+            groupId = "${personnel-positioning.kafka.location.group-id:personnel-position-location-probe}"
+    )
+    public void onMessage(ConsumerRecord<byte[], byte[]> record) {
+        LOGGER.info(
+                "收到Kafka定位原始消息，topic={}, partition={}, offset={}, timestamp={}, "
+                        + "timestampType={}, keyBytes={}, valueBytes={}, key={}, headers={}, value={}",
+                record.topic(),
+                record.partition(),
+                record.offset(),
+                record.timestamp(),
+                record.timestampType(),
+                length(record.key()),
+                length(record.value()),
+                LocationKafkaPayloadPreview.render(record.key(), maxLogBytes),
+                renderHeaders(record),
+                LocationKafkaPayloadPreview.render(record.value(), maxLogBytes)
+        );
+    }
+
+    private int length(byte[] value) {
+        return value == null ? 0 : value.length;
+    }
+
+    private String renderHeaders(ConsumerRecord<byte[], byte[]> record) {
+        StringJoiner joiner = new StringJoiner(", ", "[", "]");
+        for (Header header : record.headers()) {
+            joiner.add(header.key() + "=" + LocationKafkaPayloadPreview.render(header.value(), maxLogBytes));
+        }
+        return joiner.toString();
+    }
+}