BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 93dadc14f667bc96d727c971b0a9cd00288a0a06

Authored by YunaiV
1 parent 6d3d0973

feat: 增加 IoT Kafka 消息总线实现 

- 新增 IotKafkaMessageBus，支持 IoT gateway 与 yudao-server 通过 Kafka 通信
- 增加 type=kafka 自动配置，并补充消息总线类型说明
- 为 iot-gateway 补充 spring-kafka 依赖和 Kafka 连接配置
- 将 iot-gateway 与 yudao-server 的 IoT message-bus 默认切换为 kafka
Inline Side-by-side
Showing 1 changed file with 127 additions and 0 deletions
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/kafka/IotKafkaMessageBus.java 0 → 100644
  View file @93dadc1
  1 +package cn.iocoder.yudao.module.iot.core.messagebus.core.kafka;
  2 +
  3 +import cn.hutool.core.util.TypeUtil;
  4 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  5 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
  6 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
  7 +import jakarta.annotation.PreDestroy;
  8 +import lombok.Getter;
  9 +import lombok.extern.slf4j.Slf4j;
  10 +import org.apache.kafka.clients.consumer.ConsumerConfig;
  11 +import org.apache.kafka.clients.producer.ProducerConfig;
  12 +import org.apache.kafka.common.serialization.StringDeserializer;
  13 +import org.apache.kafka.common.serialization.StringSerializer;
  14 +import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
  15 +import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  16 +import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  17 +import org.springframework.kafka.core.KafkaTemplate;
  18 +import org.springframework.kafka.listener.AcknowledgingMessageListener;
  19 +import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
  20 +import org.springframework.kafka.listener.ContainerProperties;
  21 +
  22 +import java.lang.reflect.Type;
  23 +import java.util.ArrayList;
  24 +import java.util.HashMap;
  25 +import java.util.List;
  26 +import java.util.Map;
  27 +import java.util.concurrent.ExecutionException;
  28 +
  29 +/**
  30 + * 基于 Kafka 的 {@link IotMessageBus} 实现类
  31 + *
  32 + * @author 芋道源码
  33 + */
  34 +@Slf4j
  35 +public class IotKafkaMessageBus implements IotMessageBus {
  36 +
  37 + private final KafkaTemplate<String, String> kafkaTemplate;
  38 +
  39 + private final KafkaProperties kafkaProperties;
  40 +
  41 + @Getter
  42 + private final List<IotMessageSubscriber<?>> subscribers = new ArrayList<>();
  43 +
  44 + private final List<ConcurrentMessageListenerContainer<String, String>> containers = new ArrayList<>();
  45 +
  46 + public IotKafkaMessageBus(KafkaProperties kafkaProperties) {
  47 + this.kafkaProperties = kafkaProperties;
  48 + this.kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(buildProducerProperties(kafkaProperties)));
  49 + }
  50 +
  51 + @Override
  52 + public void post(String topic, Object message) {
  53 + String messageJson = JsonUtils.toJsonString(message);
  54 + try {
  55 + kafkaTemplate.send(topic, messageJson).get();
  56 + log.info("[post][topic({}) 发送消息({})]", topic, message);
  57 + } catch (InterruptedException e) {
  58 + Thread.currentThread().interrupt();
  59 + throw new IllegalStateException(String.format("发送 Kafka 消息失败，topic(%s) message(%s)", topic, message), e);
  60 + } catch (ExecutionException e) {
  61 + throw new IllegalStateException(String.format("发送 Kafka 消息失败，topic(%s) message(%s)", topic, message), e);
  62 + }
  63 + }
  64 +
  65 + @Override
  66 + public void register(IotMessageSubscriber<?> subscriber) {
  67 + Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0);
  68 + if (type == null) {
  69 + throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
  70 + }
  71 +
  72 + // 1. 创建消费容器
  73 + ContainerProperties containerProperties = new ContainerProperties(subscriber.getTopic());
  74 + containerProperties.setGroupId(subscriber.getGroup());
  75 + containerProperties.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
  76 + containerProperties.setMissingTopicsFatal(false);
  77 + containerProperties.setMessageListener((AcknowledgingMessageListener<String, String>) (message, acknowledgment) -> {
  78 + try {
  79 + subscriber.onMessage(JsonUtils.parseObject(message.value(), type));
  80 + acknowledgment.acknowledge();
  81 + } catch (Exception ex) {
  82 + log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]",
  83 + subscriber.getTopic(), subscriber.getGroup(), message, subscriber.getClass().getName(), ex);
  84 + throw ex;
  85 + }
  86 + });
  87 + ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(
  88 + new DefaultKafkaConsumerFactory<>(buildConsumerProperties(kafkaProperties, subscriber.getGroup())),
  89 + containerProperties);
  90 + container.start();
  91 +
  92 + // 2. 保存消费者引用
  93 + containers.add(container);
  94 + subscribers.add(subscriber);
  95 + }
  96 +
  97 + @PreDestroy
  98 + public void destroy() {
  99 + for (ConcurrentMessageListenerContainer<String, String> container : containers) {
  100 + try {
  101 + container.stop();
  102 + log.info("[destroy][关闭 Kafka 消费者容器成功]");
  103 + } catch (Exception e) {
  104 + log.error("[destroy][关闭 Kafka 消费者容器异常]", e);
  105 + }
  106 + }
  107 + kafkaTemplate.destroy();
  108 + }
  109 +
  110 + private static Map<String, Object> buildProducerProperties(KafkaProperties kafkaProperties) {
  111 + Map<String, Object> properties = new HashMap<>(kafkaProperties.buildProducerProperties());
  112 + properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  113 + properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  114 + return properties;
  115 + }
  116 +
  117 + private static Map<String, Object> buildConsumerProperties(KafkaProperties kafkaProperties, String group) {
  118 + Map<String, Object> properties = new HashMap<>(kafkaProperties.buildConsumerProperties());
  119 + properties.put(ConsumerConfig.GROUP_ID_CONFIG, group);
  120 + properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  121 + properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  122 + properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  123 + properties.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  124 + return properties;
  125 + }
  126 +
  127 +}