关于 Apache Pulsar
Apache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性。GitHub 地址：http://github.com/apache/pulsar/
本文翻译自：《Using Apache Pulsar With Kotlin》，作者 Gilles Barbier。原文链接：https://gillesbarbier.medium.com/using-apache-pulsar-with-kotlin-3b0ab398cf52
译者简介
宋博，就职于北京百观科技有限公司，高级开发工程师，专注于微服务，云计算，大数据领域。
Apache Pulsar 通常被描述为下一代 Kafka，是开发人员工具集中一颗冉冉升起的新星。Pulsar 是用于 server-to-server 消息传递的多租户、高性能解决方案，通常用作可扩展应用程序的核心。
Pulsar 可以与 Kotlin 一起使用，因为它是用 Java 编写的。不过，它的 API 并没有考虑 Kotlin 带来的强大功能，例如数据类、协程或无反射序列化。
在这篇文章中，我将讨论如何通过 Kotlin 来使用 Pulsar。
为消息体使用原生序列化
在 Kotlin 中定义消息的一种默认方式是使用数据类，这些类的主要目的是保存数据。对于此类数据类，Kotlin 会自动提供 equals()、toString()、copy()等方法 ，从而缩短代码长度并降低出现错误的风险。
使用 Java 创建一个 Pulsar 生产者:
Producer<MyAvro> avroProducer = client .newProducer(Schema.AVRO(MyAvro.class)) .topic(“some-avro-topic”) .create();
复制代码
该 Schema.AVRO(MyAvro.class) 指令将内省 MyAvro Java 类并从中推断出一个 Schema。这需要校验新的生产者是否会产生与现有消费者实际兼容的消息。然而 Kotlin 数据类的 Java 实现不能很好地与 Pulsar 使用的默认序列化器配合使用。但幸运的是，从 2.7.0 版本开始，Pulsar 允许您对生产者和消费者使用自定义序列化程序。
首先，您需要安装官方 Kotlin 序列化插件。使用它可以创建一个如下的消息类：
@Serializable data class RunTask( val taskName: TaskName, val taskId: TaskId, val taskInput: TaskInput, val taskOptions: TaskOptions, val taskMeta: TaskMeta )
复制代码
注意 @Serializable 注解。有了它，你就可以使用 RunTask.serialiser() 让序列化器在不内省的情况下工作，这将使效率大大提升！
目前，序列化插件仅支持 JSON（以及一些其他在 beta 内的格式 例如 protobuf）。所以我们还需要 avro4k 库来扩展它并支持 Avro 格式。
使用这些工具，我们可以创建一个像下面这样的 Producer 任务：
import com.github.avrokotlin.avro4k.Avroimport com.github.avrokotlin.avro4k.io.AvroEncodeFormatimport io.infinitic.common.tasks.executors.messages.RunTaskimport kotlinx.serialization.KSerializerimport org.apache.avro.file.SeekableByteArrayInputimport org.apache.avro.generic.GenericDatumReaderimport org.apache.avro.generic.GenericRecordimport org.apache.avro.io.DecoderFactoryimport org.apache.pulsar.client.api.Consumerimport org.apache.pulsar.client.api.Producerimport org.apache.pulsar.client.api.PulsarClientimport org.apache.pulsar.client.api.Schemaimport org.apache.pulsar.client.api.schema.SchemaDefinitionimport org.apache.pulsar.client.api.schema.SchemaReaderimport org.apache.pulsar.client.api.schema.SchemaWriterimport java.io.ByteArrayOutputStreamimport java.io.InputStream
// Convert T instance to Avro schemaless binary formatfun <T : Any> writeBinary(t: T, serializer: KSerializer<T>): ByteArray { val out = ByteArrayOutputStream() Avro.default.openOutputStream(serializer) { encodeFormat = AvroEncodeFormat.Binary schema = Avro.default.schema(serializer) }.to(out).write(t).close()
return out.toByteArray()}
// Convert Avro schemaless byte array to T instancefun <T> readBinary(bytes: ByteArray, serializer: KSerializer<T>): T { val datumReader = GenericDatumReader<GenericRecord>(Avro.default.schema(serializer)) val decoder = DecoderFactory.get().binaryDecoder(SeekableByteArrayInput(bytes), null)
return Avro.default.fromRecord(serializer, datumReader.read(null, decoder))}
// custom Pulsar SchemaReaderclass RunTaskSchemaReader: SchemaReader<RunTask> { override fun read(bytes: ByteArray, offset: Int, length: Int) = read(bytes.inputStream(offset, length))
override fun read(inputStream: InputStream) = readBinary(inputStream.readBytes(), RunTask.serializer())}
// custom Pulsar SchemaWriterclass RunTaskSchemaWriter : SchemaWriter<RunTask> { override fun write(message: RunTask) = writeBinary(message, RunTask.serializer())}
// custom Pulsar SchemaDefinition<RunTask>fun runTaskSchemaDefinition(): SchemaDefinition<RunTask> = SchemaDefinition.builder<RunTask>() .withJsonDef(Avro.default.schema(RunTask.serializer()).toString()) .withSchemaReader(RunTaskSchemaReader()) .withSchemaWriter(RunTaskSchemaWriter()) .withSupportSchemaVersioning(true) .build()
// Create an instance of Producer<RunTask>fun runTaskProducer(client: PulsarClient): Producer<RunTask> = client .newProducer(Schema.AVRO(runTaskSchemaDefinition())) .topic("some-avro-topic") .create();
// Create an instance of Consumer<RunTask>fun runTaskConsumer(client: PulsarClient): Consumer<RunTask> = client .newConsumer(Schema.AVRO(runTaskSchemaDefinition())) .topic("some-avro-topic") .subscribe();密封类消息和每个 Topic 一个封装Pulsar 每个 Topic 只允许一种类型的消息。在某些特殊情况下，这并不能满足全部需求。但这个问题可以通过使用封装模式来变通。首先，使用密封类从一个 Topic 创建所有类型消息：@Serializablesealed class TaskEngineMessage() { abstract val taskId: TaskId}
@Serializabledata class DispatchTask( override val taskId: TaskId, val taskName: TaskName, val methodName: MethodName, val methodParameterTypes: MethodParameterTypes?, val methodInput: MethodInput, val workflowId: WorkflowId?, val methodRunId: MethodRunId?, val taskMeta: TaskMeta, val taskOptions: TaskOptions = TaskOptions()) : TaskEngineMessage()
@Serializabledata class CancelTask( override val taskId: TaskId, val taskOutput: MethodOutput) : TaskEngineMessage()
@Serializabledata class TaskCanceled( override val taskId: TaskId, val taskOutput: MethodOutput, val taskMeta: TaskMeta) : TaskEngineMessage()
@Serializabledata class TaskCompleted( override val taskId: TaskId, val taskName: TaskName, val taskOutput: MethodOutput, val taskMeta: TaskMeta) : TaskEngineMessage()
复制代码
然后，再为这些消息创建一个封装：
Note @Serializabledata class TaskEngineEnvelope( val taskId: TaskId, val type: TaskEngineMessageType, val dispatchTask: DispatchTask? = null, val cancelTask: CancelTask? = null, val taskCanceled: TaskCanceled? = null, val taskCompleted: TaskCompleted? = null,) { init { val noNull = listOfNotNull( dispatchTask, cancelTask, taskCanceled, taskCompleted )
require(noNull.size == 1) require(noNull.first() == message()) require(noNull.first().taskId == taskId) }
companion object { fun from(msg: TaskEngineMessage) = when (msg) { is DispatchTask -> TaskEngineEnvelope( msg.taskId, TaskEngineMessageType.DISPATCH_TASK, dispatchTask = msg ) is CancelTask -> TaskEngineEnvelope( msg.taskId, TaskEngineMessageType.CANCEL_TASK, cancelTask = msg ) is TaskCanceled -> TaskEngineEnvelope( msg.taskId, TaskEngineMessageType.TASK_CANCELED, taskCanceled = msg ) is TaskCompleted -> TaskEngineEnvelope( msg.taskId, TaskEngineMessageType.TASK_COMPLETED, taskCompleted = msg ) } }
fun message(): TaskEngineMessage = when (type) { TaskEngineMessageType.DISPATCH_TASK -> dispatchTask!! TaskEngineMessageType.CANCEL_TASK -> cancelTask!! TaskEngineMessageType.TASK_CANCELED -> taskCanceled!! TaskEngineMessageType.TASK_COMPLETED -> taskCompleted!! }}
enum class TaskEngineMessageType { CANCEL_TASK, DISPATCH_TASK, TASK_CANCELED, TASK_COMPLETED}
复制代码
请注意 Kotlin 如何优雅地检查 init! 可以借助 TaskEngineEnvelope.from(msg) 很容易创建一个封装，并通过 envelope.message() 返回原始消息。
为什么这里添加了一个显式 taskId 值，而非使用一个全局字段 message:TaskEngineMessage，并且针对每种消息类型使用一个字段呢？是因为通过这种方式，我就可以借助 taskId 或 type，亦或者两者相结合的方式使用PulsarSQL 来获取这个 Topic 的信息。
通过协程来构建 Worker
在普通 Java 中使用 Thread 很复杂且容易出错。好在 Koltin 提供了 coroutines——一种更简单的异步处理抽象——和 channels——一种在协程之间传输数据的便捷方式。
我可以通过以下方式创建一个 Worker：
单个 ("task-engine-message-puller") 专用于从 Pulsar 拉取消息的协程
N 个协程 ( "task-engine-$i") 并行处理消息
单个 ("task-engine-message-acknoldeger") 处理后确认 Pulsar 消息的协程
有很多个类似于这样的进程后我已经添加了一个 logChannel 用来采集日志。请注意，为了能够在与接收它的协程不同的协程中确认 Pulsar 消息，我需要将TaskEngineMessage封装到包含Pulsar messageId的MessageToProcess<TaskEngineMessage>中:
typealias TaskEngineMessageToProcess = MessageToProcess<TaskEngineMessage>
fun CoroutineScope.startPulsarTaskEngineWorker( taskEngineConsumer: Consumer<TaskEngineEnvelope>, taskEngine: TaskEngine, logChannel: SendChannel<TaskEngineMessageToProcess>?, enginesNumber: Int) = launch(Dispatchers.IO) {
val taskInputChannel = Channel<TaskEngineMessageToProcess>() val taskResultsChannel = Channel<TaskEngineMessageToProcess>()
// coroutine dedicated to pulsar message pulling launch(CoroutineName("task-engine-message-puller")) { while (isActive) { val message: Message<TaskEngineEnvelope> = taskEngineConsumer.receiveAsync().await()
try { val envelope = readBinary(message.data, TaskEngineEnvelope.serializer()) taskInputChannel.send(MessageToProcess(envelope.message(), message.messageId)) } catch (e: Exception) { taskEngineConsumer.negativeAcknowledge(message.messageId) throw e } } }
// coroutines dedicated to Task Engine repeat(enginesNumber) { launch(CoroutineName("task-engine-$it")) { for (messageToProcess in taskInputChannel) { try { messageToProcess.output = taskEngine.handle(messageToProcess.message) } catch (e: Exception) { messageToProcess.exception = e } taskResultsChannel.send(messageToProcess) } } }
// coroutine dedicated to pulsar message acknowledging launch(CoroutineName("task-engine-message-acknowledger")) { for (messageToProcess in taskResultsChannel) { if (messageToProcess.exception == null) { taskEngineConsumer.acknowledgeAsync(messageToProcess.messageId).await() } else { taskEngineConsumer.negativeAcknowledge(messageToProcess.messageId) } logChannel?.send(messageToProcess) } }}
data class MessageToProcess<T> ( val message: T, val messageId: MessageId, var exception: Exception? = null, var output: Any? = null)
复制代码
总结
在本文中，我们介绍了如何在 Kotlin 中实现的 Pulsar 使用方法：
关注公众号「Apache Pulsar」，获取更多技术干货
加入 Apache Pulsar 中文交流群 👇🏻
评论