SpringCloud对使用者透明的数据同步组件

一、背景

云端使用Spring Cloud实现，A服务有一些数据，B和C服务也需要A服务的这些数据，但是系统上面只有A服务有数据操作的入口，B和C服务只能从A服务处同步数据到自己的表里面。

 

解决方案是A服务对数据进行增删改操作之后，将数据操作发送给消息服务，B和C服务从消息服务拉取数据同步消息，然后修改自己的数据。

 

本文以kafka作为消息服务组件。

 

二、需求和目标

开发一个组件，A服务引用此组件，配置kafka连接和topic信息，注入生产者，即可向指定的topic发送数据同步消息。

 

B和C服务引用此组件，配置kafka连接和topic信息，编写核心的数据同步业务代码，即可实现同步数据的消费和入库。

 

有以下几点要求：

• 对使用者透明，使用者不需要了解具体的消息服务组件，只需要使用数据同步组件的API即可实现数据的发送和消费；

• 扩展方便，如果云端的消息服务不使用kafka而是其余的比如rabbitmq或rocketmq之类的消息服务，使用者不需要修改代码，组件库添加支持rabbitmq或rocketmq的生产者和消费者监听，使用者修改一下配置即可。

三、组件概述

将具体的消息组件（如kafka、rabbitmq或rocketmq等）封装在底层，提供给数据同步的生产者和消费者相对友好、与底层消息组件无关的API和接口。

 

在生产者这边，在业务层中注入生产者即可使用producer来发送需要同步的数据消息。

 

在消费者这边，只要编写类实现业务消费者接口，并且将其放入spring中管理即可消费到同步数据消息做业务处理。

 

四、组件开发 - 基础

1. 消息实体类

生产者接口的实现类需要将数据同步消息封装成DataSyncMessage的对象，序列化之后进行发送。

消费者监听会将消息反序列化为DataSyncMessage对象，从里面获取到操作类型和数据类型，以及同步的真实数据，然后调用具体的业务消费者实现的相应的方法来做业务处理。

/**
 * 数据同步消息封装
 */
@Data
public class DataSyncMessage {

  private Integer operationType;

  private Integer dataType;

  private Object data;
}

2. 操作类型枚举

OperationType枚举封装的操作类型，增、删、改等。

/**
 * 操作类型枚举
 */
public enum OperationType {

  /**
   * 新增
   */
  INSERT(1, "insert"),

  /**
   * 修改
   */
  UPDATE(2, "update"),

  /**
   * 删除
   */
  DELETE(3, "delete");

  private Integer value;
  private String key;

  OperationType(Integer value, String key) {
    this.value = value;
    this.key = key;
  }

  public String getKey() {
    return key;
  }

  public Integer getValue() {
    return value;
  }

  public static OperationType getEnum(Integer value) {
    return Stream.of(OperationType.values())
        .filter(e -> e.getValue().equals(value))
        .findFirst()
        .orElse(null);
  }
}

3. 同步数据类型枚举

/**
 * 数据类型枚举
 */
public enum DataType {

  /**
   * 用户
   */
  USER(1, "USER"),

  /**
   * 订单
   */
  ORDER(2, "ORDER");

  private Integer value;
  private String key;

  DataType(Integer value, String key) {
    this.value = value;
    this.key = key;
  }

  public String getKey() {
    return key;
  }

  public Integer getValue() {
    return value;
  }

  public static DataType getEnum(Integer value) {
    return Stream.of(DataType.values())
        .filter(e -> e.getValue().equals(value))
        .findFirst()
        .orElse(null);
  }
}

五、组件开发 - 生产者

1. 生产者接口

提供给数据同步的生产者使用，对于生产者来说，具体的底层消息组件是透明的。

/**
 * 数据同步生产者接口
 */
public interface DataSyncProducer {

  /**
   * 发送同步消息
   *
   * @param operationType 操作类型，增删改等
   * @param dataType 数据类型，用户、订单等
   * @param data 同步的数据
   */
  void sendSyncData(OperationType operationType, DataType dataType, Object data);
}

组件当前内置了kafka的生产者实现和自动注入配置类，如果将来的技术选型是rocketmq或者其他消息中间件，编写一个相对应的生产者实现类和配置类即可使用，数据同步的生产者不需要修改业务代码。

2. 生产者Kafka实现

内置的kafka生产者实现，应该根据不同的数据同步消息中间件选型编写不同的生产者实现类和配置类。

/**
 * 数据同步生产者 - kafka实现
 */
@Slf4j
public class KafkaDataSyncProducer implements DataSyncProducer {

  private final KafkaTemplate<String, String> kafkaTemplate;

  private final String topic;

  public KafkaDataSyncProducer(KafkaTemplate<String, String> kafkaTemplate, String topic) {
    this.kafkaTemplate = kafkaTemplate;
    this.topic = topic;
  }

  @Override
  public void sendSyncData(OperationType operationType, DataType dataType, Object data) {

    DataSyncMessage message = new DataSyncMessage();
    message.setOperationType(operationType.getValue());
    message.setDataType(dataType.getValue());
    message.setData(data);

    String jsonMessage = JSON.toJSONString(message);

    kafkaTemplate.send(topic, jsonMessage);
  }
}

3. 生产者kafka配置类

kafka数据同步生产者配置类，只有在data-sync.producer.data-sync-type配置的值为kafka时才自动装配kafka生产者。

@Configuration
@ConditionalOnProperty(prefix = "data-sync.producer", name = "data-sync-type", havingValue = "kafka")
public class KafkaDataSyncConfig {

  @Value("${data-sync.producer.data-sync-topic}")
  private String dataSyncTopic;

  @Resource
  private KafkaTemplate<String, String> kafkaTemplate;

  @Bean
  public DataSyncProducer dataSyncProducer() {
    return new KafkaDataSyncProducer(kafkaTemplate, dataSyncTopic);
  }
}

@ConditionalOnProperty注解表示，当data-sync.producer.data-sync-type配置的值为kafka时才会装配该configuration类。

六、组件开发 - 消费者

1. 业务层需实现的接口

数据同步消费者需要编写DataSyncConsumer接口的实现类并将其注入到spring中，另外实现类需要使用@DataSyncListener标注，并指定数据类型。

/**
 * 数据同步监听器接口
 */
public interface DataSyncConsumer<T> {

  /**
   * 当发生插入操作时触发
   */
  void onInsert(T data);

  /**
   * 当发生修改操作时触发
   */
  void onUpdate(T data);

  /**
   * 当发生删除操作时触发
   */
  void onDelete(T data);
}

2. 消费扫描处理器

DataSyncListenerAnnotationBeanPostProcessor处理器会扫描到容器中被@DataSyncListener标注、实现了DataSyncConsumer接口的所有bean对象，从@DataSyncListener注解获取数据类型，然后将此bean对象交给DataSyncConsumerHolder管理。

DataSyncConsumerHolder管理着所有的数据同步消费者，内部使用Map结构，key是同步消费者所处理的数据类型，value是消费者对象本身。

BeanPostProcessor接口的实现会在bean对象被初始化之后调用，此处不做过多解释。

@Slf4j
@Component
public class DataSyncListenerAnnotationBeanPostProcessor implements BeanPostProcessor {

  @Override
  public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {

    DataSyncListener annotation = bean.getClass().getAnnotation(DataSyncListener.class);

    if (annotation != null) {

      Class<?>[] interfaces = bean.getClass().getInterfaces();

      boolean isDataSyncConsumer = false;

      for (Class<?> i : interfaces) {
        if (i == DataSyncConsumer.class) {
          isDataSyncConsumer = true;
          break;
        }
      }

      if (isDataSyncConsumer) {

        DataSyncConsumer dataSyncConsumer = (DataSyncConsumer) bean;

        DataType dataType = annotation.dataType();

        Class clazz = annotation.dataClass();

        DataSyncDataTypeAndConsumer typeAndConsumer = new DataSyncDataTypeAndConsumer();
        typeAndConsumer.setDataClass(clazz);
        typeAndConsumer.setConsumer(dataSyncConsumer);

        DataSyncConsumerHolder.add(dataType, typeAndConsumer);

        DataSyncConsumerHolder.print();
      }
    }

    return bean;
  }
}

@DataSyncListener注解：

/**
 * 标注在数据同步实现类上
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
@Documented
public @interface DataSyncListener {

  DataType dataType();

  Class dataClass();
}

3. Kafka消费者监听

kafka数据同步消费者监听器，只有在data-sync.consumer.data-sync-type配置的值为kafka时才自动装配kafka监听。

@Slf4j
@Service
@ConditionalOnProperty(prefix = "data-sync.consumer", name = "data-sync-type", havingValue = "kafka")
public class KafkaDataSyncService {

  @KafkaListener(topics = {"${data-sync.consumer.data-sync-topic}"})
  public void handleDataSyncMessage(ConsumerRecord<?, ?> record) {
    Optional<?> kafkaMessage = Optional.ofNullable(record.value());
    if (!kafkaMessage.isPresent()) {
      return;
    }
    String message = kafkaMessage.get().toString();

    log.info("topic:{},message:{}", record.topic(), message);

    // 业务处理
    dataSync(message);
  }

  private void dataSync(String message) {

    try {

      DataSyncMessage dataSyncMessage = JSON.parseObject(message, DataSyncMessage.class);

      Integer dataType = dataSyncMessage.getDataType();
      OperationType operationType = OperationType.getEnum(dataSyncMessage.getOperationType());

      DataType type = DataType.getEnum(dataType);

      // debug
      DataSyncConsumerHolder.print();

      DataSyncDataTypeAndConsumer consumer = DataSyncConsumerHolder.get(type);

      if (consumer == null) {
        log.warn("Consumer not found for data type: {}", type);
        return;
      }

      Class clazz = consumer.getDataClass();

      DataSyncConsumer realConsumer = consumer.getConsumer();

      Object data = dataSyncMessage.getData();

      // debug
      log.info("data = " + data);

      Object o;

      if (clazz == Map.class) {
        o = data;
      } else {
        o = clazz.newInstance();
        BeanUtils.populate(o, (Map<String, ? extends Object>) data);
      }

      switch (operationType) {
        case INSERT:
          realConsumer.onInsert(o);
          break;
        case DELETE:
          realConsumer.onDelete(o);
          break;
        case UPDATE:
          realConsumer.onUpdate(o);
          break;
        default:
          log.warn("Error operation: " + operationType);
      }

    } catch (Exception e) {
      log.error(e.getLocalizedMessage(), e);
      log.error("An exception occurred during data synchronization: ", message);
    }
  }
}

七、生产者配置及使用

1. 配置

data-sync:
  producer:
    data-sync-topic: data-sync-topic
    data-sync-type: kafka

2. 注入生产者

@Resource
private DataSyncProducer dataSyncProducer;

3. 发送消息

User user = new User();
user.setUsername("admin");

dataSyncProducer.sendSyncData(OperationType.INSERT, DataType.USER, user);

八、消费者配置及使用

1. 配置

data-sync:
  consumer:
    data-sync-topic: data-sync-topic
    data-sync-type: kafka

2. 实现业务层消费者接口

实现该接口的onInsert、onUpdate、onDelete等方法，如下：

@DataSyncListener(dataType = DataType.USER, dataClass = User.class)
@Service("UserDataSyncConsumer")
public class UserDataSyncConsumer implements DataSyncConsumer<User> {

    @Override
    public void onInsert(User data) {}

    @Override
    public void onUpdate(User data) {}

    @Override
    public void onDelete(User data) {}
}

接口泛型是同步数据的数据类型，如果写Map<String, Object>，组件不会对同步数据做解析和封装，直接将原始Map传递给接口方法。

@DataSyncListener注解的dataType参数指定同步数据的数据类型，具体见上面的DataType枚举介绍。dataClass 参数作用同接口泛型。

@Service注解尽量传一个唯一的bean名称。

 

另一个实现类的例子：

@DataSyncListener(dataType = DataType.USER, dataClass = Map.class)
@Service("UserDataSyncConsumer")
public class UserDataSyncConsumer implements DataSyncConsumer<Map<String, Object>> {

    @Override
    public void onInsert(Map<String, Object> data) {}

    @Override
    public void onUpdate(Map<String, Object> data) {}

    @Override
    public void onDelete(Map<String, Object> data) {}
}

九、扩展说明

1. 扩展同步数据类型

为DataType枚举添加新的类型即可。

暂时这样扩展，后续可以优化为更加友好的扩展方式。

2. 扩展支持新的消息服务

比如使用rabbitmq作为消息服务：

• 编写支持rabbitmq的DataSyncProducer实现类；
• 编写rabbitmq生产者配置类，data-sync.producer.data-sync-type配置的值为rabbitmq时才自动装配rabbitmq生产者；
• 编写支持rabbitmq的消息监听器，data-sync.consumer.data-sync-type配置的值为rabbitmq时才自动装配监听器；
• 数据同步生产者和消费者引入组件库，做相应的配置。

十、 参考源码

https://gitee.com/xuguofeng2020/net5ijy-mall/tree/master/component-datasync