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。