2. 在控制台消费test006的消息，参考命令：
```shell
./kafka-console-consumer.sh
--bootstrap-server 127.0.0.1:9092
--topic test006

2. 根据提示，groupid输入<font color="blue">com.bolingcavalry</font>，artifactid输入<font color="blue">flinksinkdemo</font>，即可创建一个maven工程；
3. 在pom.xml中增加kafka依赖库：
```xml
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.11</artifactId><version>1.9.0</version>
</dependency>

private String topic;public ProducerStringSerializationSchema(String topic) {super();this.topic = topic;
}@Override
public ProducerRecord<byte[], byte[]> serialize(String element, Long timestamp) {return new ProducerRecord<byte[], byte[]>(topic, element.getBytes(StandardCharsets.UTF_8));
}

2. 创建任务类KafkaStrSink，请注意FlinkKafkaProducer对象的参数，FlinkKafkaProducer.Semantic.EXACTLY_ONCE表示严格一次：
```java
package com.bolingcavalry.addsink;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;public class KafkaStrSink {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//并行度为1env.setParallelism(1);Properties properties = new Properties();properties.setProperty("bootstrap.servers", "192.168.50.43:9092");String topic = "test006";FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(topic,new ProducerStringSerializationSchema(topic),properties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE);//创建一个List，里面有两个Tuple2元素List<String> list = new ArrayList<>();list.add("aaa");list.add("bbb");list.add("ccc");list.add("ddd");list.add("eee");list.add("fff");list.add("aaa");//统计每个单词的数量env.fromCollection(list).addSink(producer).setParallelism(4);env.execute("sink demo : kafka str");}
}

package com.bolingcavalry.addsink;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import javax.annotation.Nullable;public class ObjSerializationSchema implements KafkaSerializationSchema<Tuple2<String, Integer>> {private String topic;private ObjectMapper mapper;public ObjSerializationSchema(String topic) {super();this.topic = topic;}@Overridepublic ProducerRecord<byte[], byte[]> serialize(Tuple2<String, Integer> stringIntegerTuple2, @Nullable Long timestamp) {byte[] b = null;if (mapper == null) {mapper = new ObjectMapper();}try {b= mapper.writeValueAsBytes(stringIntegerTuple2);} catch (JsonProcessingException e) {// 注意，在生产环境这是个非常危险的操作，// 过多的错误打印会严重影响系统性能，请根据生产环境情况做调整e.printStackTrace();}return new ProducerRecord<byte[], byte[]>(topic, b);}
}

package com.bolingcavalry.addsink;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;public class KafkaObjSink {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//并行度为1env.setParallelism(1);Properties properties = new Properties();//kafka的broker地址properties.setProperty("bootstrap.servers", "192.168.50.43:9092");String topic = "test006";FlinkKafkaProducer<Tuple2<String, Integer>> producer = new FlinkKafkaProducer<>(topic,new ObjSerializationSchema(topic),properties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE);//创建一个List，里面有两个Tuple2元素List<Tuple2<String, Integer>> list = new ArrayList<>();list.add(new Tuple2("aaa", 1));list.add(new Tuple2("bbb", 1));list.add(new Tuple2("ccc", 1));list.add(new Tuple2("ddd", 1));list.add(new Tuple2("eee", 1));list.add(new Tuple2("fff", 1));list.add(new Tuple2("aaa", 1));//统计每个单词的数量env.fromCollection(list).keyBy(0).sum(1).addSink(producer).setParallelism(4);env.execute("sink demo : kafka obj");}
}