# Flink Kafka样例程序（Scala）
#### 功能介绍
在Flink应用中，调用flink-connector-kafka模块的接口，生产并消费数据。
#### 代码样例
用户在开发前需要使用对接安全模式的Kafka，则需要引入FusionInsight的kafka-clients-\*.jar，该jar包可在kafka客户端目录下获取。
下面代码片段仅为演示，完整代码参见FlinkKafkaScalaExample样例工程下的com.huawei.bigdata.flink.examples.WriteIntoKafka和com.huawei.bigdata.flink.examples.ReadFromKafka。
```
//producer代码
object WriteIntoKafka {
     def main(args: Array[String]) {
     // 打印出执行flink run的参考命令
       System.out.println("use command as: ")
       System.out.println("./bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka" +
         " /opt/test.jar --topic topic-test -bootstrap.servers 10.91.8.218:9092")
       System.out.println
("******************************************************************************************")
       System.out.println("<topic> is the kafka topic name")
       System.out.println("<bootstrap.servers> is the ip:port list of brokers")
       System.out.println
("******************************************************************************************")
       // 构造执行环境
       val env = StreamExecutionEnvironment.getExecutionEnvironment
       // 设置并发度
       env.setParallelism(1)
       // 解析运行参数
       val paraTool = ParameterTool.fromArgs(args)
       // 构造流图，将自定义Source生成的数据写入Kafka
       val messageStream: DataStream[String] = env.addSource(new SimpleStringGenerator)
       messageStream.addSink(new FlinkKafkaProducer(
         paraTool.get("topic"), new SimpleStringSchema, paraTool.getProperties))
       // 调用execute触发执行
       env.execute
     }
   }
   // 自定义Source，每隔1s持续产生消息
   class SimpleStringGenerator extends SourceFunction[String] {
     var running = true
     var i = 0
     override def run(ctx: SourceContext[String]) {
       while (running) {
         ctx.collect("element-" + i)
         i += 1
         Thread.sleep(1000)
       }
     }
     override def cancel() {
       running = false
     }
   } 
//consumer代码
object ReadFromKafka {
     def main(args: Array[String]) {
     // 打印出执行flink run的参考命令
       System.out.println("use command as: ")
       System.out.println("./bin/flink run --class com.huawei.bigdata.flink.examples.ReadFromKafka" +
         " /opt/test.jar --topic topic-test -bootstrap.servers 10.91.8.218:9092")
       System.out.println
("******************************************************************************************")
       System.out.println("<topic> is the kafka topic name")
       System.out.println("<bootstrap.servers> is the ip:port list of brokers")
       System.out.println
("******************************************************************************************")
      // 构造执行环境
       val env = StreamExecutionEnvironment.getExecutionEnvironment
      // 设置并发度
       env.setParallelism(1)
      // 解析运行参数
       val paraTool = ParameterTool.fromArgs(args)
      // 构造流图，从Kafka读取数据并换行打印
       val messageStream = env.addSource(new FlinkKafkaConsumer(
         paraTool.get("topic"), new SimpleStringSchema, paraTool.getProperties))
       messageStream
         .map(s => "Flink says " + s + System.getProperty("line.separator")).print()
       // 调用execute触发执行
       env.execute()
     }
   }
```