模擬數據生成代碼

新建一個employee實體類

package com.zuoan;

public class Employee {

public int id;

public String name;

public String password;

public int age;

public Integer salary;

public String department;

public Employee(int id, String name, String password, int age, Integer salary, String department) {

this.id = id;

this.name = name;

this.password = password;

this.age = age;

this.salary = salary;

this.department = department;

}

public Employee() {

}

@Override

public String toString() {

return "Employee{" +

"id=" + id +

", name='" + name + '\'' +

", password='" + password + '\'' +

", age=" + age +

", salary=" + salary +

", department='" + department + '\'' +

'}';

}

public Integer getSalary() {

return salary;

}

public void setSalary(Integer salary) {

this.salary = salary;

}

public String getDepartment() {

return department;

}

public void setDepartment(String department) {

this.department = department;

}

public int getId() {

return id;

}

public void setId(int id) {

this.id = id;

}

public String getName() {

return name;

}

public void setName(String name) {

this.name = name;

}

public String getPassword() {

return password;

}

public void setPassword(String password) {

this.password = password;

}

public int getAge() {

return age;

}

public void setAge(int age) {

this.age = age;

}

}

模擬生成數據，這里是每秒生成一條數據到kafka

package com.zuoan;

import com.google.gson.Gson;

import org.apache.kafka.clients.producer.ProducerRecord;

import org.apache.kafka.clients.producer.KafkaProducer;

import java.util.Properties;

import java.util.Random;

public class KafkaProducerTest {

public static void main(String[] args) {

Producer();

}

public static void Producer() {

String broker = "localhost:9092";

String topic = "demo";

Properties props = new Properties();

props.put("bootstrap.servers", broker);

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer producer = new KafkaProducer(props);

String[] depLists = new String[5];

depLists[0] = "行政部";

depLists[1] = "賬務部";

depLists[2] = "市場部";

depLists[3] = "技術部";

depLists[4] = "銷售部";

Random rand = new Random(300);

Gson gson = new Gson();

for (int i = 1; i <= 1000; i++) {

Employee employee = new Employee(i, "user" + i, "password" + i, rand.nextInt(40) + 20, (rand.nextInt(300) + 1) * 100, depLists[rand.nextInt(5)]);

String temp = gson.toJson(employee).toString();

ProducerRecord record = new ProducerRecord(topic, null, "user" + i, temp);

producer.send(record);

System.out.println("發送數據: " + temp);

try {

Thread.sleep(1 * 1000); //發送一條數據 sleep

} catch (InterruptedException e) {

e.printStackTrace();

}

}

producer.flush();

}

}

總結

以上是生活随笔為你收集整理的kafka批量写mysql_Flink 读取 Kafka 批量写入到 MySQL 0444的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。