alpakka-kafka(1)-producer

alpakka项目是一个基于akka-streams流处理编程工具的scala/java开源项目,通过提供connector连接各种数据源并在akka-streams里进行数据处理。alpakka-kafka就是alpakka项目里的kafka-connector。对于我们来说:可以用alpakka-kafka来对接kafka,使用kafka提供的功能。或者从另外一个角度讲:alpakka-kafka就是一个用akka-streams实现kafka功能的scala开发工具。

alpakka-kafka提供了kafka的核心功能:producer、consumer,分别负责把akka-streams里的数据写入kafka及从kafka中读出数据并输入到akka-streams里。用akka-streams集成kafka的应用场景通常出现在业务集成方面:在一项业务A中产生一些业务操作指令写入kafka,然后通过kafka把指令传送给另一项业务B,业务B从kafka中获取操作指令并进行相应的业务操作。如:有两个业务模块:收货管理和库存管理,一方面收货管理向kafka写入收货记录。另一头库存管理从kafka中读取收货记录并更新相关库存数量记录。注意,这两项业务是分别操作的。在alpakka中,实际的业务操作基本就是在akka-streams里的数据处理(transform),其实是典型的CQRS模式:读写两方互不关联,写时不管受众是谁,如何使用、读者不关心谁是写方。这里的写和读两方分别代表kafka里的producer和consumer。

本篇我们先介绍alpakka-kafka的producer功能及其使用方法。如前所述:alpakka是用akka-streams实现了kafka-producer功能。alpakka提供的producer也就是akka-streams的一种组件,可以与其它的akka-streams组件组合形成更大的akka-streams个体。构建一个producer需要先完成几个配件类构成:

1、producer-settings配置:alpakka-kafka在reference.conf里的akka.kafka.producer配置段落提供了足够支持基本运作的默认producer配置。用户可以通过typesafe config配置文件操作工具来灵活调整配置

2、de/serializer序列化工具:alpakka-kafka提供了String类型的序列化/反序列化函数,可以直接使用

4、bootstrap-server:一个以逗号分隔的kafka-cluster节点ip清单文本

下面是一个具体的例子:

implicit val system = ActorSystem("kafka_sys")
val bootstrapServers = "localhost:9092"
val config = system.settings.config.getConfig("akka.kafka.producer")
val producerSettings =
ProducerSettings(config, new StringSerializer, new StringSerializer)
.withBootstrapServers(bootstrapServers)

这里使用ActorSystem只是为了读取.conf文件里的配置,还没有使用任何akka-streams组件。akka.kafka.producer配置段落在alpakka-kafka的reference.conf里提供了默认配置,不需要在application.conf里重新定义。

alpakka-kafka提供了一个最基本的producer,非akka-streams组件,sendProducer。下面我们示范一下sendProducer的使用和效果:

import akka.actor.ActorSystem
import akka.kafka.scaladsl.{Consumer, SendProducer}
import org.apache.kafka.clients.producer.{ProducerRecord, RecordMetadata}
import akka.kafka._
import org.apache.kafka.common.serialization._
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
object SendProducerDemo extends App {
implicit val system = ActorSystem("kafka_sys")
implicit val executionContext = system.dispatcher
val bootstrapServers = "localhost:9092"
val config = system.settings.config.getConfig("akka.kafka.producer")
val producerSettings =
ProducerSettings(config, new StringSerializer, new StringSerializer)
.withBootstrapServers(bootstrapServers)
val producer = SendProducer(producerSettings)
val topic = "greatings"
val lstfut: Seq[Future[RecordMetadata]] =
(100 to 200).reverse
.map(_.toString)
.map(value => new ProducerRecord[String, String](topic, s"hello-$value"))
.map(msg => producer.send(msg))
val futlst = Future.sequence(lstfut)
Await.result(futlst, 2.seconds)
scala.io.StdIn.readLine()
producer.close()
system.terminate()
}

以上示范用sendProducer向kafka写入100条hello消息。使用的是集合遍历,没有使用akka-streams的Source。为了检验具体效果,我们可以使用kafka提供的一些手工指令,如下:

\w> ./kafka-topics --create --topic greatings --bootstrap-server localhost:9092
Created topic greatings.
\w> ./kafka-console-consumer --topic greatings  --bootstrap-server localhost:9092
hello-100
hello-101
hello-102
hello-103
hello-104
hello-105
hello-106
...

既然producer代表写入功能,那么在akka-streams里就是Sink或Flow组件的功能了。下面这个例子是producer Sink组件plainSink的示范:

import akka.Done
import akka.actor.ActorSystem
import akka.kafka.scaladsl._
import akka.kafka._
import akka.stream.scaladsl._
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization._
import scala.concurrent._
import scala.concurrent.duration._
object plain_sink extends App {
implicit val system = ActorSystem("kafka_sys")
val bootstrapServers = "localhost:9092"
val config = system.settings.config.getConfig("akka.kafka.producer")
val producerSettings =
ProducerSettings(config, new StringSerializer, new StringSerializer)
.withBootstrapServers(bootstrapServers)
implicit val executionContext = system.dispatcher
val topic = "greatings"
val done: Future[Done] =
Source(1 to 100)
.map(_.toString)
.map(value => new ProducerRecord[String, String](topic, s"hello-$value"))
.runWith(Producer.plainSink(producerSettings))
Await.ready(done,3.seconds)
scala.io.StdIn.readLine()
system.terminate()
}

这是一个典型的akka-streams应用实例,其中Producer.plainSink就是一个akka-streams Sink组件。

以上两个示范都涉及到构建一个ProducerRecord类型并将之写入kafka。ProducerRecord是一个基本的kafka消息类型:

public ProducerRecord(String topic, K key, V value) {
this(topic, null, null, key, value, null);
}

topic是String类型,key, value 是 Any 类型的。 alpakka-kafka在ProducerRecord之上又拓展了一个复杂点的消息类型ProducerMessage.Envelope类型:

sealed trait Envelope[K, V, +PassThrough] {
def passThrough: PassThrough
def withPassThrough[PassThrough2](value: PassThrough2): Envelope[K, V, PassThrough2]
}
final case class Message[K, V, +PassThrough](
record: ProducerRecord[K, V],
passThrough: PassThrough
) extends Envelope[K, V, PassThrough] {
override def withPassThrough[PassThrough2](value: PassThrough2): Message[K, V, PassThrough2] =
copy(passThrough = value)
}

ProducerMessage.Envelope增加了个PassThrough参数,用来与消息一道传递额外的元数据。alpakka-kafka streams组件使用这个消息类型作为流元素,最终把它转换成一或多条ProducerRecord写入kafka。如下: 

object EventMessages {
//一对一条ProducerRecord
def createMessage[KeyType,ValueType,PassThroughType](
topic: String,
key: KeyType,
value: ValueType,
passThrough: PassThroughType): ProducerMessage.Envelope[KeyType,ValueType,PassThroughType] = {
val single = ProducerMessage.single(
new ProducerRecord[KeyType,ValueType](topic,key,value),
passThrough
)
single
}
//一对多条ProducerRecord
def createMultiMessage[KeyType,ValueType,PassThroughType] (
topics: List[String],
key: KeyType,
value: ValueType,
passThrough: PassThroughType): ProducerMessage.Envelope[KeyType,ValueType,PassThroughType] = {
import scala.collection.immutable
val msgs = topics.map { topic =>
new ProducerRecord(topic,key,value)
}.toSeq
val multi = ProducerMessage.multi(
msgs,
passThrough
)
multi
}
//只传递通过型元数据
def createPassThroughMessage[KeyType,ValueType,PassThroughType](
topic: String,
key: KeyType,
value: ValueType,
passThrough: PassThroughType): ProducerMessage.Envelope[KeyType,ValueType,PassThroughType] = {
ProducerMessage.passThrough(passThrough)
}
}

flexiFlow是一个alpakka-kafka Flow组件,流入ProducerMessage.Evelope,流出Results类型: 

def flexiFlow[K, V, PassThrough](
settings: ProducerSettings[K, V]
): Flow[Envelope[K, V, PassThrough], Results[K, V, PassThrough], NotUsed] = { ... }

Results类型定义如下:

final case class Result[K, V, PassThrough] private (
metadata: RecordMetadata,
message: Message[K, V, PassThrough]
) extends Results[K, V, PassThrough] {
def offset: Long = metadata.offset()
def passThrough: PassThrough = message.passThrough
}

也就是说flexiFlow可以返回写入kafka后kafka返回的操作状态数据。我们再看看flexiFlow的使用案例: 

import akka.kafka.ProducerMessage._
import akka.actor.ActorSystem
import akka.kafka.scaladsl._
import akka.kafka.{ProducerMessage, ProducerSettings}
import akka.stream.scaladsl.{Sink, Source}
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer
import scala.concurrent._
import scala.concurrent.duration._
object flexi_flow extends App {
implicit val system = ActorSystem("kafka_sys")
val bootstrapServers = "localhost:9092"
val config = system.settings.config.getConfig("akka.kafka.producer")
val producerSettings =
ProducerSettings(config, new StringSerializer, new StringSerializer)
.withBootstrapServers(bootstrapServers)
// needed for the future flatMap/onComplete in the end
implicit val executionContext = system.dispatcher
val topic = "greatings"
val done = Source(1 to 100)
.map { number =>
val value = number.toString
EventMessages.createMessage(topic,"key",value,number)
}
.via(Producer.flexiFlow(producerSettings))
.map {
case ProducerMessage.Result(metadata, ProducerMessage.Message(record, passThrough)) =>
s"${metadata.topic}/${metadata.partition} ${metadata.offset}: ${record.value}"
case ProducerMessage.MultiResult(parts, passThrough) =>
parts
.map {
case MultiResultPart(metadata, record) =>
s"${metadata.topic}/${metadata.partition} ${metadata.offset}: ${record.value}"
}
.mkString(", ")
case ProducerMessage.PassThroughResult(passThrough) =>
s"passed through"
}
.runWith(Sink.foreach(println(_)))
Await.ready(done,3.seconds)
scala.io.StdIn.readLine()
system.terminate()
}
object EventMessages {
def createMessage[KeyType,ValueType,PassThroughType](
topic: String,
key: KeyType,
value: ValueType,
passThrough: PassThroughType): ProducerMessage.Envelope[KeyType,ValueType,PassThroughType] = {
val single = ProducerMessage.single(
new ProducerRecord[KeyType,ValueType](topic,key,value),
passThrough
)
single
}
def createMultiMessage[KeyType,ValueType,PassThroughType] (
topics: List[String],
key: KeyType,
value: ValueType,
passThrough: PassThroughType): ProducerMessage.Envelope[KeyType,ValueType,PassThroughType] = {
import scala.collection.immutable
val msgs = topics.map { topic =>
new ProducerRecord(topic,key,value)
}.toSeq
val multi = ProducerMessage.multi(
msgs,
passThrough
)
multi
}
def createPassThroughMessage[KeyType,ValueType,PassThroughType](
topic: String,
key: KeyType,
value: ValueType,
passThrough: PassThroughType): ProducerMessage.Envelope[KeyType,ValueType,PassThroughType] = {
ProducerMessage.passThrough(passThrough)
}
}

producer除向kafka写入与业务相关的业务事件或业务指令外还会向kafka写入当前消息读取的具体位置offset,所以alpakka-kafka的produce可分成两种类型:上面示范的plainSink, flexiFlow只向kafka写业务数据。还有一类如commitableSink还包括了把消息读取位置offset写入commit的功能。如下:

val control =
Consumer
.committableSource(consumerSettings, Subscriptions.topics(topic1, topic2))
.map { msg =>
ProducerMessage.single(
new ProducerRecord(targetTopic, msg.record.key, msg.record.value),
msg.committableOffset
)
}
.toMat(Producer.committableSink(producerSettings, committerSettings))(DrainingControl.apply)
.run()
control.drainAndShutdown()

如上所示,committableSource从kafka读取业务消息及读取位置committableOffsset,然后Producer.committableSink把业务消息和offset再写入kafka。

下篇讨论我们再具体介绍consumer。

博客园精华区
我还没有学会写个人说明!
上一篇

卸载 PyCharm!这才是 Python 小白的最理想的 IDE

下一篇

淘宝客从业者开年福利,“红包”可以这样拿!

你也可能喜欢

评论已经被关闭。

插入图片