被以下文章引用，勿动：

说明

创建工程，实现kafka生产者

创建 maven 工程

略

pom.xml

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.1</version>
</dependency>

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

<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.30</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.30</version>
</dependency>

log4j.properties

在 resources 目录下创建 log4j.properties 文件，内容如下：

# DEBUG 级别，即：所有信息都会输出
log4j.rootLogger=WARN, stdout
# 配置 stdout，输出到控制台
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

java 代码

事件对象

package top.malaoshi.kafka;

import com.fasterxml.jackson.annotation.JsonFormat;

import java.util.Date;

public class Event {
    private String id;
    private double temp;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone = "GMT+8")
    private Date time;

    public Event(String id, double temp, Date time) {
        this.id = id;
        this.temp = temp;
        this.time = time;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public double getTemp() {
        return temp;
    }

    public void setTemp(double temp) {
        this.temp = temp;
    }

    public Date getTime() {
        return time;
    }

    public void setTime(Date time) {
        this.time = time;
    }
}

事件构造器

工具类

package top.malaoshi.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.Date;

public class EventBuilder {

    /**
     *
     * @param id
     * @param temp
     * @param delay 延迟时间，单位秒。最终时间 = 当前时间 - 延迟时间
     * @return
     */
    public static Event build(String id, double temp, int delay){
        Date date = new Date();
        long timeNew = date.getTime() - delay*1000;
        Date dateNew = new Date(timeNew);
        Event ret = new Event(id,temp,dateNew);

        return ret;
    }
    public static Event build(String id, double temp){
        Date date = new Date();
        Event ret = new Event(id,temp,date);

        return ret;
    }

    public static String buildJson(String id, double temp, int delay){
        Event event = build(id,temp,delay);

        ObjectMapper om = new ObjectMapper();
        try {
            String json = om.writeValueAsString(event);
            return json;
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    public static String buildJson(String id, double temp){
        Event event = build(id,temp);

        ObjectMapper om = new ObjectMapper();
        try {
            String json = om.writeValueAsString(event);
            return json;
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }
}

原文出处：/show_1IX47WZKETxV.html