Scala Serializer类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Scala›Scala教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Scala中org.apache.kafka.common.serialization.Serializer类的典型用法代码示例。如果您正苦于以下问题：Scala Serializer类的具体用法？Scala Serializer怎么用？Scala Serializer使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了Serializer类的5个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Scala代码示例。

示例1: configure

//设置package包名称以及导入依赖的类
package edu.uw.at.iroberts.wirefugue.kafka.serdes

import java.util

import org.apache.kafka.common.serialization.{Deserializer, Serde, Serializer}


trait StatelessSerde[T] { self: Serde[T] =>
  override def configure(configs: util.Map[String, _], isKey: Boolean) = ()
  override def close() = ()
}

trait StatelessSerializer[T] { self: Serializer[T] =>
  override def configure(configs: util.Map[String, _], isKey: Boolean) = ()
  override def close() = ()
}

trait StatelessDeserializer[T] { self: Deserializer[T] =>
  override def configure(configs: util.Map[String, _], isKey: Boolean) = ()
  override def close() = ()
}

开发者ID:robertson-tech，项目名称:wirefugue，代码行数:22，代码来源:Stateless.scala

示例2: SeqSerialiser

//设置package包名称以及导入依赖的类
package serialisation

import java.io.{ByteArrayOutputStream, ObjectOutputStream}
import java.util

import org.apache.kafka.common.serialization.Serializer

class SeqSerialiser[ELEMENT] extends Serializer[Seq[ELEMENT]] {

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = {
  }

  override def serialize(topic: String, data: Seq[ELEMENT]): Array[Byte] = {
    val byteStream = new ByteArrayOutputStream()
    val objectStream = new ObjectOutputStream(byteStream)
    data.foreach(objectStream.writeObject(_))
    objectStream.close()
    byteStream.toByteArray
  }

  override def close(): Unit = {
  }
}

开发者ID:benwheeler，项目名称:kafka-streams-poc，代码行数:24，代码来源:SeqSerialiser.scala

示例3: EventSerialiser

//设置package包名称以及导入依赖的类
package serialisation

import java.util

import model.Event
import org.apache.kafka.common.serialization.{Serializer, StringSerializer}
import org.json4s.native.Serialization
import org.json4s.native.Serialization.write
import org.json4s.{Formats, NoTypeHints}

class EventSerialiser extends Serializer[Event] {

  implicit val formats: Formats = Serialization.formats(NoTypeHints)

  val stringSerialiser = new StringSerializer

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = {
    stringSerialiser.configure(configs, isKey)
  }

  override def serialize(topic: String, data: Event): Array[Byte] = {
    val stringValue = write(data)
    stringSerialiser.serialize(topic, stringValue)
  }

  override def close(): Unit = {
    stringSerialiser.close()
  }
}

开发者ID:benwheeler，项目名称:kafka-streams-poc，代码行数:30，代码来源:EventSerialiser.scala

示例4: produceRecord

//设置package包名称以及导入依赖的类
package de.choffmeister.microserviceutils.kafka.testkit

import java.util.UUID

import akka.Done
import akka.actor.ActorSystem
import akka.kafka.scaladsl.{Consumer, Producer}
import akka.kafka.{ConsumerSettings, ProducerSettings, Subscriptions}
import akka.stream.Materializer
import akka.stream.scaladsl.{Keep, Sink, Source}
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.{Deserializer, Serializer}

import scala.concurrent.Future

trait KafkaTest {
  def produceRecord[K, V](topic: String, keySerializer: Serializer[K], valueSerializer: Serializer[V], key: K, value: V)(implicit system: ActorSystem, mat: Materializer): Future[Done] = {
    val producerSettings = ProducerSettings(system, keySerializer, valueSerializer)
      .withBootstrapServers(system.settings.config.getString("kafka.bootstrap-servers"))

    Source.single(new ProducerRecord("mail.command.send", key, value))
      .toMat(Producer.plainSink(producerSettings))(Keep.right)
      .run()
  }

  def consumeRecordPF[K, V, Out](topic: String, keyDeserializer: Deserializer[K], valueDeserializer: Deserializer[V])(pf: PartialFunction[ConsumerRecord[K, V], Out])(implicit system: ActorSystem, mat: Materializer): Future[Out] = {
    val consumerSettings = ConsumerSettings(system, keyDeserializer, valueDeserializer)
      .withBootstrapServers(system.settings.config.getString("kafka.bootstrap-servers"))
      .withGroupId(UUID.randomUUID.toString)
      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

    Consumer.plainSource(consumerSettings, Subscriptions.topics(topic))
      .filter(pf.isDefinedAt)
      .map(pf)
      .toMat(Sink.head)(Keep.right)
      .run()
  }
}

开发者ID:choffmeister，项目名称:microservice-utils，代码行数:40，代码来源:KafkaTest.scala

示例5: UUIDSerializer

//设置package包名称以及导入依赖的类
package de.choffmeister.microserviceutils.kafka

import java.nio.ByteBuffer
import java.util
import java.util.UUID

import org.apache.kafka.common.serialization.{Deserializer, Serializer}

class UUIDSerializer extends Serializer[UUID] {
  override def serialize(topic: String, data: UUID): Array[Byte] = {
    val bb = ByteBuffer.wrap(new Array[Byte](16))
    bb.putLong(data.getMostSignificantBits)
    bb.putLong(data.getLeastSignificantBits)
    bb.array
  }

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = {}

  override def close(): Unit = {}
}

class UUIDDeserializer extends Deserializer[UUID] {
  override def deserialize(topic: String, data: Array[Byte]): UUID = {
    val bb = ByteBuffer.wrap(data)
    new UUID(bb.getLong, bb.getLong)
  }

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = {}

  override def close(): Unit = {}
}

开发者ID:choffmeister，项目名称:microservice-utils，代码行数:32，代码来源:UUIDSerializer.scala

注：本文中的org.apache.kafka.common.serialization.Serializer类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。