scala发送消息到kafka示例

1) build文件需要填写依赖的jar包

"org.slf4j" % "slf4j-api" % "1.7.5",
"org.slf4j" % "slf4j-log4j12" % "1.7.5",
"com.yammer.metrics" % "metrics-core" % "2.2.0",
"com.yammer.metrics" % "metrics-annotation" % "2.2.0",
"org.apache.kafka" % "kafka-clients" % "0.9.0.0",
"org.apache.kafka" % "kafka_2.11" % "0.9.0.0"

2) src/man/resources/

新建一个文件log4j.properties

#log4j.rootLogger=INFO, stdout
log4j.rootLogger=ERROR, FileOut
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

log4j.appender.FileOut=org.apache.log4j.FileAppender
log4j.appender.FileOut.File=./logs.log
log4j.appender.FileOut.Append=true
log4j.appender.FileOut.layout=org.apache.log4j.PatternLayout
log4j.appender.FileOut.layout.ConversionPattern=%-d{yyyy-MM-dd HH\:mm\:ss,SSS} [%C]-[%p] %m%n

3) src/scala目录下新建一个kafka_sendmsg.scala文件

package kafka_sendmsg

import java.util.{Date, Properties}
import kafka.producer.KeyedMessage
import kafka.producer.ProducerConfig
import kafka.producer.Producer

object kafka_sendmsg extends App {
val props = new Properties()
props.put("metadata.broker.list","10.57.137.131:9092")  // broker 如果有多个,中间使用逗号分隔
props.put("serializer.class","kafka.serializer.StringEncoder")
props.put("request.required.acks","1")
val config = new ProducerConfig(props)
val producer = new Producer[String,String](config)
val runtime = new Date().toString
val msg = "message publishing time - " + runtime
val topic="logs" //注意topic
val data = new KeyedMessage[String,String](topic,msg)
producer.send(data)
producer.close()

}

4) consumer的日志

ip说明: 10.57.137.131 是 vmware 自动获取的IP.

5) 文件目录结构