如何实现KCache缓存初始化的泛型化，适配不同存储对象类型

Kotlin可以通过实化类型参数（reified） 解决泛型擦除导致无法获取T实际类型的问题，配合工厂方法即可完成初始化逻辑的泛型化，具体实现如下：

方案1：使用实化类型参数的工厂方法（推荐） #

Kotlin的inline函数支持reified修饰的泛型参数，编译期会自动替换为实际调用的类型，运行时可直接读取类型信息，不需要手动传递类型对象。我们可以将构造器私有化，通过伴生对象的inline工厂方法来实例化MyCache：

import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.common.serialization.Serdes
import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig
import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
import org.apache.avro.specific.SpecificRecord
import java.util.*
import kotlin.collections.sortedMapOf

// 构造器私有化，仅通过工厂方法实例化
class MyCache<T> private constructor(
    private val cache: SortedMap<String, T>
) : SortedMap<String, T> by cache { // 委托实现SortedMap接口，对外直接暴露缓存操作能力

    companion object {
        // 带实化泛型参数的工厂方法
        inline fun <reified T> create(name: String, test: Boolean = true, appConfig: AppConfig): MyCache<T> {
            val cache = if (test) {
                sortedMapOf<String, T>()
            } else {
                val cacheProps = Properties()
                cacheProps[KafkaCacheConfig.KAFKACACHE_BOOTSTRAP_SERVERS_CONFIG] =
                    appConfig.propertyOrNull("kafka.bootstrapServers")?.getString()
                cacheProps[KafkaCacheConfig.KAFKACACHE_GROUP_ID_CONFIG] =
                    appConfig.propertyOrNull("applicationId")?.getString()
                cacheProps[KafkaCacheConfig.KAFKACACHE_CLIENT_ID_CONFIG] = "GeneratingUnitsCache"
                cacheProps[KafkaCacheConfig.KAFKACACHE_TOPIC_CONFIG] = "app.$name.kcache"

                val serdeConfig = Collections.singletonMap(
                    AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
                    appConfig.property("kafka.schemaRegistryUrl").getString(),
                )

                // 根据实际类型匹配对应Serde，可扩展更多类型分支
                val valueSerde: Serde<T> = when {
                    T::class == String::class -> Serdes.String() as Serde<T>
                    SpecificRecord::class.java.isAssignableFrom(T::class.java) -> {
                        SpecificAvroSerde<T>().apply {
                            configure(serdeConfig, false)
                        }
                    }
                    else -> throw IllegalArgumentException("不支持的缓存值类型：${T::class.simpleName}")
                }

                KafkaCache(
                    KafkaCacheConfig(cacheProps),
                    Serdes.String(),
                    valueSerde
                ).apply { init() }
            }
            return MyCache(cache)
        }
    }
}

调用方式非常简洁：

// 初始化存储MyAvroObject的缓存
val avroCache = MyCache.create<MyAvroObject>("avro-cache", test = false, appConfig)
// 初始化存储String的缓存
val stringCache = MyCache.create<String>("string-cache", test = false, appConfig)

方案2：构造器传入Class（兼容非inline场景） #

如果业务场景不适合使用inline函数，可以手动在构造时传递类型的Class对象实现同样的逻辑：

class MyCache<T>(
    name: String,
    test: Boolean = true,
    appConfig: AppConfig,
    valueType: Class<T>
) {
    private val cache: SortedMap<String, T>

    init {
        cache = if (test) {
            sortedMapOf()
        } else {
            // 公共配置初始化逻辑和方案1完全一致，此处省略
            val valueSerde: Serde<T> = when {
                valueType == String::class.java -> Serdes.String() as Serde<T>
                SpecificRecord::class.java.isAssignableFrom(valueType) -> {
                    SpecificAvroSerde<T>().apply { configure(serdeConfig, false) }
                }
                else -> throw IllegalArgumentException("不支持的缓存值类型：${valueType.simpleName}")
            }
            KafkaCache(KafkaCacheConfig(cacheProps), Serdes.String(), valueSerde).apply { init() }
        }
    }
}

调用方式：

val avroCache = MyCache("avro-cache", false, appConfig, MyAvroObject::class.java)

内容的提问来源于stack exchange，提问作者hermanjakobsen