Apache Dubbo的@SPI接口应用

记录：472

场景：使用Apache Dubbo的@SPI接口加载实现类搭建框架。

版本：JDK 1.8,dubbo-common-3.0.0。

SPI全称Service Provider Interface。

1.基础

1.1引用依赖

<dependency>
  <groupId>org.apache.dubbogroupId>
  <artifactId>dubbo-commonartifactId>
  <version>3.0.0version>
dependency>

1.2应用

(1)使用@SPI注解作用在自定义接口com.hub.example.pf.adapter.IPfDataAdapter。

(2)业务类TCityAdapterImpl、TProvinceAdapterImpl等实现接口IPfDataAdapter。

(3)在..\src\main\resources目录下，新建\META-INF\services目录

(4)在\META-INF\services目录新建配置文件：com.hub.example.pf.adapter.IPfDataAdapter，文件名称就是@SPI注解作用的接口全路径名称。

(5)在\META-INF\services\com.hub.example.pf.adapter.IPfDataAdapter配置文件中添加入下内容(key=value键值对)：

t_province=com.hub.example.process.adapter.TProvinceAdapterImpl
t_city=com.hub.example.process.adapter.TCityAdapterImpl

(6)根据配置文件中key名称和接口名称获取实现类，如下：

IPfDataAdapter var= ExtensionLoader.getExtensionLoader(IPfDataAdapter.class).getExtension(tableName);

1.3逻辑

Apache Dubbo框架对使用@SPI注解接口，会在..\src\main\resources\META-INF\services目录下扫描配置并加载。

2.示例场景

(1)从某数据源(比如数据库)中读取数据存放在：Map data。

(2)组装成Tuple3> in数据格式，其中Tuple3第一个参数数：表名称，第二参数：指定的字段名称，第三个参数就是数据。

(3)使用计算框架com.hub.example.pf.function.PfDataMapFunction计算业务

输入：Tuple3> in；

输出：Tuple3> out；

(4)把Tuple3> out传递给下游处理。比如数据写入到HBase等目的端。

3.示例代码

3.1接口和抽象类

(1)接口IPfDataAdapter

全称：com.hub.example.pf.adapter.IPfDataAdapter

代码：

@SPI
public interface IPfDataAdapter {
    Tuple3> getPfData(Map value);
}

(2)抽象类

全称：com.hub.example.pf.adapter.PfDataAdapterAbstract

代码：

public abstract class PfDataAdapterAbstract implements IPfDataAdapter {
    public PfDataAdapterAbstract() {
    }
    public Tuple3> getPfData(Map data) {
        try {
            Map result = new HashMap(data.size());
            for (Map.Entry entry : data.entrySet()) {
                if (entry.getValue() != null) {
                    result.put(entry.getKey(), entry.getValue().toString());
                }
            }
            return new Tuple3(this.getTableName(), this.getRowKeyColumns(), result);
        } catch (Exception e) {
            return new Tuple3(null, null, new HashMap<>());
        }
    }
    public abstract String getDataType();
    public String getTableName() {
        return TableConfig.getTableName(this.getDataType());
    }
    public String getRowKeyColumns() {
        return TableConfig.getRowKeyColumns(this.getDataType());
    }
}

3.2计算框架

全称：com.hub.example.pf.function.PfDataMapFunction

代码：

public class PfDataMapFunction {
    public PfDataMapFunction() {
    }
    public Tuple3> map(Tuple3> data) {
        try {
            String tableName = (String) data.f0;
            IPfDataAdapter dataAdapter = ExtensionLoader.getExtensionLoader(IPfDataAdapter.class).getExtension(tableName);
            if (dataAdapter == null) {
                return new Tuple3();
            } else {
                return dataAdapter.getPfData((Map) data.f2);
            }
        } catch (Exception e) {
            e.printStackTrace();
            return new Tuple3();
        }
    }
}

3.3业务类

(1)业务类TProvinceAdapterImpl对应表：t_province

全称：com.hub.example.process.adapter.TProvinceAdapterImpl

代码：

public class TProvinceAdapterImpl extends PfDataAdapterAbstract {
    public TProvinceAdapterImpl() {
    }
    @Override
    public String getDataType() {
        return "T_PROVINCE";
    }
    public Tuple3> getPfData(Map data) {
        if (data.get("PROVINCE_ID") == null) {
            return new Tuple3();
        }
        return super.getPfData(data);
    }
}

(2)业务类TCityAdapterImpl对应表：t_city

全称：com.hub.example.process.adapter.TCityAdapterImpl

代码：

public class TCityAdapterImpl extends PfDataAdapterAbstract {
    @Override
    public String getDataType() {
        return "T_CITY";
    }
 
    public Tuple3> getPfData(Map data) {
        if (data.get("CITY_ID") == null) {
            return new Tuple3();
        }
        return super.getPfData(data);
    }
}

3.4配置文件

(1)配置目录..\META-INF\services

目录：..\src\main\resources\META-INF\services

(2)配置文件com.hub.example.pf.adapter.IPfDataAdapter

文件名：com.hub.example.pf.adapter.IPfDataAdapter

文件内容：

t_province=com.hub.example.process.adapter.TProvinceAdapterImpl
t_city=com.hub.example.process.adapter.TCityAdapterImpl

3.5配置类

全称：com.hub.example.config.TableConfig

代码：

public class TableConfig {
    private static ConcurrentHashMap tableConfig = new ConcurrentHashMap<>();
    static {
        tableConfig.put("TABLE:T_PROVINCE", "HUB:T_PROVINCE");
        tableConfig.put("ROW_KEY:T_PROVINCE", "V_DATA_DATE,PROVINCE_ID");
        tableConfig.put("TABLE:T_CITY", "HUB:T_CITY");
        tableConfig.put("ROW_KEY:T_CITY", "V_DATA_DATE,CITY_ID");
    }
    public static String getTableName(String tableName) {
        return tableConfig.get("TABLE:" + tableName.toUpperCase());
    }
    public static String getRowKeyColumns(String tableName) {
        return tableConfig.get("ROW_KEY:" + tableName.toUpperCase());
    }
}

3.6实体类

全称：com.hub.example.api.Tuple3

代码：

public class Tuple3 implements Serializable {
    private static final long serialVersionUID = 1L;
    public T0 f0;
    public T1 f1;
    public T2 f2;
 
    public Tuple3() {
    }
 
    public Tuple3(T0 f0, T1 f1, T2 f2) {
        this.f0 = f0;
        this.f1 = f1;
        this.f2 = f2;
    }
 
    public static String arrayAwareToString(Object o) {
        String arrayString = Arrays.deepToString(new Object[]{o});
        return arrayString.substring(1, arrayString.length() - 1);
    }
 
    public String toString() {
        return "(" + arrayAwareToString(this.f0) + ";" + arrayAwareToString(this.f1) + ";" + arrayAwareToString(this.f2) + ")";
    }
}

4.示例测试

全称：com.hub.example.process.ExampleApplicationSPI

代码：

public class ExampleApplicationSPI {
    public static void main(String[] args) throws Exception {
        PfDataMapFunction dataPfMapFunction = new PfDataMapFunction();
        //操作t_province
        String table01 = "t_province";
        Map data01 = getTProvince();
        Tuple3> in01 = new Tuple3(table01, data01.get("PROVINCE_ID"), data01);
        Tuple3> out01 = dataPfMapFunction.map(in01);
        System.out.println("操作表t_province后数据: " + out01.toString());
        //操作t_city
        String table02 = "t_city";
        Map data02 = getTCity();
        Tuple3> in02 = new Tuple3(table02, data02.get("CITY_ID"), data02);
        Tuple3> out02 = dataPfMapFunction.map(in02);
        System.out.println("操作表t_city后数据: " + out02.toString());
    }
    public static Map getTProvince() {
        Map dataMap = new HashMap<>();
        dataMap.put("PROVINCE_ID", 33001694333578181L);
        dataMap.put("V_DATA_DATE", "20230910");
        dataMap.put("PROVINCE_NAME", "浙江");
        dataMap.put("GROSS", 7.77D);
        return dataMap;
    }
    public static Map getTCity() {
        Map dataMap = new HashMap<>();
        dataMap.put("CITY_ID", 33011694333578181L);
        dataMap.put("V_DATA_DATE", "20230911");
        dataMap.put("CITY_NAME", "杭州");
        dataMap.put("GROSS", 1.88D);
        return dataMap;
    }
}

以上，感谢。

2023年9月10日