diff --git a/start/src/main/resources/bootstrap.yml b/start/src/main/resources/bootstrap.yml index 1f3cdb8..dfff433 100644 --- a/start/src/main/resources/bootstrap.yml +++ b/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 diff --git a/start/src/main/resources/kafka.yml b/start/src/main/resources/kafka.yml new file mode 100644 index 0000000..36b8439 --- /dev/null +++ b/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} diff --git a/web-app/pom.xml b/web-app/pom.xml index 8b74104..951ce69 100644 --- a/web-app/pom.xml +++ b/web-app/pom.xml @@ -21,6 +21,10 @@ com.zcloud.personnel.positioning web-infrastructure + + org.springframework.kafka + spring-kafka + org.junit.jupiter junit-jupiter diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/KafkaConsumerLifecycleLogger.java b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/KafkaConsumerLifecycleLogger.java new file mode 100644 index 0000000..bb8ff08 --- /dev/null +++ b/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 partitions) { + LOGGER.info( + "Kafka定位消息订阅成功，broker={}, groupId={}, partitions={}", + bootstrapServers, + consumer.groupMetadata().groupId(), + partitions + ); + } + + @Override + public void onPartitionsRevokedBeforeCommit(Consumer consumer, Collection partitions) { + LOGGER.warn( + "Kafka定位消息分区已回收，groupId={}, partitions={}", + consumer.groupMetadata().groupId(), + partitions + ); + } + + @Override + public void onPartitionsLost(Consumer consumer, Collection 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 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); + } +} diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaPayloadPreview.java b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaPayloadPreview.java new file mode 100644 index 0000000..db91420 --- /dev/null +++ b/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 ""; + } + 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"); + } +} diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaProbeListener.java b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaProbeListener.java new file mode 100644 index 0000000..2cdf589 --- /dev/null +++ b/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 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 record) { + StringJoiner joiner = new StringJoiner(", ", "[", "]"); + for (Header header : record.headers()) { + joiner.add(header.key() + "=" + LocationKafkaPayloadPreview.render(header.value(), maxLogBytes)); + } + return joiner.toString(); + } +}