Kafka operation

ConsumerTest

package com.palmaplus.amapdata.test.tour

import java.util.{Collections, Properties}

import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
import org.apache.kafka.clients.producer.ProducerConfig

import com.palmaplus.amapdata.common.KafkaUtil
import com.palmaplus.amapdata.common.KafkaUtil.bootstrapServers

/**
  * Created on 2017/5/4.
  *
  * @author shuai.yang
  */
object ConsumerTest {

  def main(args: Array[String]): Unit = {

    val props = new Properties()
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "test")
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000")
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000")
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

    val consumer = new KafkaConsumer[String, String](KafkaUtil.props)
    consumer.subscribe(Collections.singletonList(KafkaUtil.receiveTopic))

    while (true) {
      val records = consumer.poll(1000)
      val iterator = records.iterator()
      while (iterator.hasNext) {
        val record = iterator.next()
        println(record)
        KafkaUtil.send("Hello")
      }
      consumer.commitSync()
    }
  }

}

KafkaUtil

package com.palmaplus.amapdata.common

import java.util.Properties

import com.google.gson.Gson
import com.typesafe.config.{Config, ConfigFactory}
import com.typesafe.scalalogging.slf4j.LazyLogging
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import com.palmaplus.amapdata.template.message.ResultMessage

/**
  * Kafka相关操作
  *
  * Created on 2017/5/4.
  * @author shuai.yang
  */
object KafkaUtil extends LazyLogging {

  val conf: Config = ConfigFactory.load()
  val kafkaConfig: Config = conf.getConfig("kafka")
  val bootstrapServers: String = kafkaConfig.getString(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)
  val group: String = kafkaConfig.getString("group")
  val receiveTopic: String = kafkaConfig.getString("receive")
  val sendTopic: String = kafkaConfig.getString("send")

  val props = new Properties()
  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
  props.put(ConsumerConfig.GROUP_ID_CONFIG, group)
  props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
  props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000")
  props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000")
  props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
  props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
  props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

  /**
    * 发送消息
    *
    * @param message 消息内容
    */
  def send(message: String): Unit = {
    val producer = new KafkaProducer[String, String](props)
    producer.send(new ProducerRecord[String, String](sendTopic, System.currentTimeMillis().toString, message))
    producer.close()
  }

  /**
    * 统计成功
    *
    * @param mapId 地图ID
    * @param version 日期格式的版本号
    */
  def success(mapId: String, version: String): Unit = {
    val message = ResultMessage(Global.CODE_SUCCESS, mapId, version, Array.empty[String], Global.CODE_DESCRIPTION_SUCCESS)
    val content = new Gson().toJson(message)
    send(content)
    logger.info(content)
  }

  /**
    * 传入的版本号不是最新的时发送相应消息
    *
    * @param mapId 地图ID
    * @param version 日期格式的版本号
    */
  def exception(mapId: String, version: String): Unit = {
    val message = ResultMessage(Global.CODE_EXCEPTION, mapId, version, Array.empty[String], Global.CODE_DESCRIPTION_EXCEPTION)
    val content = new Gson().toJson(message)
    send(content)
    logger.error(content)
  }

  /**
    * 传入的版本号不是最新的时发送相应消息
    *
    * @param mapId 地图ID
    * @param version 日期格式的版本号
    */
  def notNewest(mapId: String, version: String): Unit = {
    val message = ResultMessage(Global.CODE_NOT_NEWEST, mapId, version, Array.empty[String], Global.CODE_DESCRIPTION_NOT_NEWEST)
    val content = new Gson().toJson(message)
    send(content)
    logger.error(content)
  }

  /**
    * 数据库中没有与mapId和version对应的数据时发送相应消息
    *
    * @param mapId 地图ID
    * @param version 日期格式的版本号
    */
  def noData(mapId: String, version: String): Unit = {
    val message = ResultMessage(Global.CODE_NO_DATA, mapId, version, Array.empty[String], Global.CODE_DESCRIPTION_NO_DATA)
    val content = new Gson().toJson(message)
    send(content)
    logger.error(content)
  }

  /**
    * region表中没有对应的数据时发送相应消息
    *
    * @param mapId 地图ID
    * @param version 日期格式的版本号
    */
  def noRegion(mapId: String, version: String): Unit = {
    val message = ResultMessage(Global.CODE_NO_REGION, mapId, version, Array.empty[String], Global.CODE_DESCRIPTION_NO_REGION)
    val content = new Gson().toJson(message)
    send(content)
    logger.error(content)
  }

  /**
    * 数据库中不存在对应的店铺名和品牌名的映射关系时发送相应消息
    *
    * @param shops 不存在的店铺名
    * @param mapId 地图ID
    * @param version 日期格式的版本号
    */
  def unknownShops(shops: Array[String], mapId: String, version: String): Unit = {
    val message = ResultMessage(Global.CODE_UNKNOWN_SHOPS, mapId, version, shops, Global.CODE_DESCRIPTION_UNKNOWN_SHOPS)
    val content = new Gson().toJson(message)
    send(content)
    logger.error(content)
  }

}