向kafka发送消息的工具类：KafkaProducer：
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) { // 将对象序列化成byte数组 byte[] data = serializer.toJSONBytes(record); // 封装 ProducerRecord<byte[], byte[]> kafkaRecord = new ProducerRecord<>(topic, data); // 发送 producer.send(kafkaRecord); // 通过sleep控制消息的速度，请依据自身kafka配置以及flink服务器配置来调整 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; } }最后是应用类SendMessageApplication，CSV文件路径、kafka的topic和borker地址都在此设置，另外借助java8的Stream API，只需少量代码即可完成所有工作：
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已经就绪，并且名为user_behavior的topic已经创建；
请将CSV文件准备好；
确认SendMessageApplication.java中的文件地址、kafka topic、kafka broker三个参数准确无误；
运行SendMessageApplication.java；
开启一个 控制台消息kafka消息，参考命令如下：
./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正常情况下可以立即见到消息，如下图：
至此，通过Java应用模拟用户行为消息流的操作就完成了，接下来的flink实战就用这个作为数据源； 欢迎关注公众号：程序员欣宸
微信搜索「程序员欣宸」，我是欣宸，期待与您一同畅游Java世界...
https://github.com/zq2599/blog_demos