<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>

 <groupId>com.bolingcavalry</groupId>
 <artifactId>flinksql</artifactId>
 <version>1.0-SNAPSHOT</version>

 <properties>
  <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  <flink.version>1.10.0</flink.version>
  <kafka.version>2.2.0</kafka.version>
  <java.version>1.8</java.version>
  <scala.binary.version>2.11</scala.binary.version>
  <maven.compiler.source>${java.version}</maven.compiler.source>
  <maven.compiler.target>${java.version}</maven.compiler.target>
 </properties>

 <dependencies>
  <dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>${kafka.version}</version>
  </dependency>

  <dependency>
   <groupId>com.fasterxml.jackson.core</groupId>
   <artifactId>jackson-databind</artifactId>
   <version>2.9.10.1</version>
  </dependency>

  <!-- Logging dependencies -->
  <dependency>
   <groupId>org.slf4j</groupId>
   <artifactId>slf4j-log4j12</artifactId>
   <version>1.7.7</version>
   <scope>runtime</scope>
  </dependency>
  <dependency>
   <groupId>log4j</groupId>
   <artifactId>log4j</artifactId>
   <version>1.2.17</version>
   <scope>runtime</scope>
  </dependency>
  <dependency>
   <groupId>net.sourceforge.javacsv</groupId>
   <artifactId>javacsv</artifactId>
   <version>2.0</version>
  </dependency>

 </dependencies>

 <build>
  <plugins>
   <!-- Java Compiler -->
   <plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-compiler-plugin</artifactId>
    <version>3.1</version>
    <configuration>
     <source>${java.version}</source>
     <target>${java.version}</target>
    </configuration>
   </plugin>

   <!-- Shade plugin to include all dependencies -->
   <plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-shade-plugin</artifactId>
    <version>3.0.0</version>
    <executions>
     <!-- Run shade goal on package phase -->
     <execution>
      <phase>package</phase>
      <goals>
       <goal>shade</goal>
      </goals>
      <configuration>
       <artifactSet>
        <excludes>
        </excludes>
       </artifactSet>
       <filters>
        <filter>
         <!-- Do not copy the signatures in the META-INF folder.
         Otherwise, this might cause SecurityExceptions when using the JAR. -->
         <artifact>*:*</artifact>
         <excludes>
          <exclude>META-INF/*.SF</exclude>
          <exclude>META-INF/*.DSA</exclude>
          <exclude>META-INF/*.RSA</exclude>
         </excludes>
        </filter>
       </filters>
      </configuration>
     </execution>
    </executions>
   </plugin>
  </plugins>
 </build>
</project>

public class UserBehaviorCsvFileReader implements Supplier<UserBehavior> {

 private final String filePath;
 private CsvReader csvReader;

 public UserBehaviorCsvFileReader(String filePath) throws IOException {

  this.filePath = filePath;
  try {
   csvReader = new CsvReader(filePath);
   csvReader.readHeaders();
  } catch (IOException e) {
   throw new IOException("Error reading TaxiRecords from file: " + filePath, e);
  }
 }

 @Override
 public UserBehavior get() {
  UserBehavior userBehavior = null;
  try{
   if(csvReader.readRecord()) {
    csvReader.getRawRecord();
    userBehavior = new UserBehavior(
      Long.valueOf(csvReader.get(0)),
      Long.valueOf(csvReader.get(1)),
      Long.valueOf(csvReader.get(2)),
      csvReader.get(3),
      new Date(Long.valueOf(csvReader.get(4))*1000L));
   }
  } catch (IOException e) {
   throw new NoSuchElementException("IOException from " + filePath);
  }

  if (null==userBehavior) {
   throw new NoSuchElementException("All records read from " + filePath);
  }

  return userBehavior;
 }
}

public class UserBehavior {

 @JsonFormat
 private long user_id;

 @JsonFormat
 private long item_id;

 @JsonFormat
 private long category_id;

 @JsonFormat
 private String behavior;

 @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss'Z'")
 private Date ts;

 public UserBehavior() {
 }

 public UserBehavior(long user_id, long item_id, long category_id, String behavior, Date ts) {
  this.user_id = user_id;
  this.item_id = item_id;
  this.category_id = category_id;
  this.behavior = behavior;
  this.ts = ts;
 }
}

public class JsonSerializer<T> {

 private final ObjectMapper jsonMapper = new ObjectMapper();

 public String toJSONString(T r) {
  try {
   return jsonMapper.writeValueAsString(r);
  } catch (JsonProcessingException e) {
   throw new IllegalArgumentException("Could not serialize record: " + r, e);
  }
 }

 public byte[] toJSONBytes(T r) {
  try {
   return jsonMapper.writeValueAsBytes(r);
  } catch (JsonProcessingException e) {
   throw new IllegalArgumentException("Could not serialize record: " + r, e);
  }
 }
}

public class KafkaProducer implements Consumer<UserBehavior> {

 private final String topic;
 private final org.apache.kafka.clients.producer.KafkaProducer<byte[], byte[]> producer;
 private final JsonSerializer<UserBehavior> serializer;

 public KafkaProducer(String kafkaTopic, String kafkaBrokers) {
  this.topic = kafkaTopic;
  this.producer = new org.apache.kafka.clients.producer.KafkaProducer<>(createKafkaProperties(kafkaBrokers));
  this.serializer = new JsonSerializer<>();
 }

 @Override
 public void accept(UserBehavior record) {
  // 將對(duì)象序列化成byte數(shù)組
  byte[] data = serializer.toJSONBytes(record);
  // 封裝
  ProducerRecord<byte[], byte[]> kafkaRecord = new ProducerRecord<>(topic, data);
  // 發(fā)送
  producer.send(kafkaRecord);

  // 通過sleep控制消息的速度，請(qǐng)依據(jù)自身kafka配置以及flink服務(wù)器配置來調(diào)整
  try {
   Thread.sleep(500);
  }catch(InterruptedException e){
   e.printStackTrace();
  }
 }

 /**
  * kafka配置
  * @param brokers The brokers to connect to.
  * @return A Kafka producer configuration.
  */
 private static Properties createKafkaProperties(String brokers) {
  Properties kafkaProps = new Properties();
  kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
  kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getCanonicalName());
  kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getCanonicalName());
  return kafkaProps;
 }
}

public class SendMessageApplication {

 public static void main(String[] args) throws Exception {
  // 文件地址
  String filePath = "D:\\temp\\202005\\02\\UserBehavior.csv";
  // kafka topic
  String topic = "user_behavior";
  // kafka borker地址
  String broker = "192.168.50.43:9092";

  Stream.generate(new UserBehaviorCsvFileReader(filePath))
    .sequential()
    .forEachOrdered(new KafkaProducer(topic, broker));
 }
}

./kafka-console-consumer.sh \
--bootstrap-server 127.0.0.1:9092 \
--topic user_behavior \
--consumer-property group.id=old-consumer-test \
--consumer-property consumer.id=old-consumer-cl \
--from-beginning