Windows平台下kafka+flink开发环境搭建

1.准备工作
安装JDK1.8(注意最好设置环境变量JAVA_HOME)
安装SCALA 2.11
下载kafka
下载flink1.7
2.kafka安装
由于kafka自带zookeeper所以下载下来就能使用。
1.进入当前目录的DOS界面：在 kafka 目录，按住shift+鼠标右键->在此处打开命令窗口(W)

2.开启zookeeper服务
bin\windows\zookeeper-server-start.bat config\zookeeper.properties


3.启动第二个DOS界面开启kafka服务
bin\windows\kafka-server-start.bat config\server.properties

4.启动第三个DOS界面开启kafka生产者（produce）

• 创建一个主题test
  bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
  
• 查看主题test
  bin\windows\kafka-topics.bat --list --zookeeper localhost:2181
  
• 启动生产者(consumer)
  bin\windows\kafka-console-producer.bat --broker-list {本机ip}:9092 --topic test
  由于是本地{本机ip}=localhost
  
  5.启动第4个DOS界面创建消费者
  bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning
  
  6.验证生产者和消费者之间的联通性
  
  7.idea开发的pom文件添加下面的依赖
  
  8.编写代码
  public static void main(String[] args) throws Exception {
  // create execution environment
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  Properties properties = new Properties();
  properties.setProperty(“bootstrap.servers”, “localhost:9092”);
  properties.setProperty(“group.id”, “flink_consumer”);
  DataStream stream = env.addSource(new FlinkKafkaConsumer<>(
  “demo”, new SimpleStringSchema(), properties) );
  stream.map(new MapFunction<String, String>() {
  private static final long serialVersionUID = -6867736771747690202L;
  @Override
  public String map(String value) throws Exception {
  return "Stream Value: " + value;
  }}).print();
  env.execute();
  }
  9.IDEA执行结果
  
  10.启动flink进入bin目录下执行start-cluster.bat
  此时出现两个Java的dos界面，同时可以进入http://localhost:8081查看
  
  
  11提交程序查看结果