kafka java序列化时间日期按照统一格式以及使用gson来做序列化反序列化配置
直接上代码

public class GsonSerializer<T> implements Serializer<T> {public static final String CONFIG_VALUE_CLASS = "value.deserializer.class";public static final String CONFIG_KEY_CLASS = "key.deserializer.class";private Class<T> cls;private Gson gson =new GsonBuilder().setLongSerializationPolicy(LongSerializationPolicy.DEFAULT).registerTypeAdapter(LocalDateTime.class,new DateTimeSerializer()).registerTypeAdapter(Date.class, new DateTimeSerializer()).create();@Overridepublic void configure(Map<String, ?> config, boolean isKey) {// this is called right after construction// use it for initialisationString configKey = isKey ? CONFIG_KEY_CLASS : CONFIG_VALUE_CLASS;String clsName = String.valueOf(config.get(configKey));try {cls = (Class<T>) Class.forName(clsName);} catch (ClassNotFoundException e) {System.err.printf("Failed to configure GsonDeserializer. " +"Did you forget to specify the '%s' property ?%n",configKey);}}@Overridepublic byte[] serialize(String s, T t) {return gson.toJson(t).getBytes(Charsets.UTF_8);}@Overridepublic void close() {// this is called right before destruction}
}

上面是序列化类

@Slf4j
public class GsonDeserializer<T> implements Deserializer<T> {public static final String CONFIG_VALUE_CLASS = "value.deserializer.class";public static final String CONFIG_KEY_CLASS = "key.deserializer.class";private Class<T> cls;private Gson gson =new GsonBuilder().setLongSerializationPolicy(LongSerializationPolicy.DEFAULT).registerTypeAdapter(LocalDateTime.class,new DateTimeDeserializer()).create();@Overridepublic void configure(Map<String, ?> config, boolean isKey) {String configKey = isKey ? CONFIG_KEY_CLASS : CONFIG_VALUE_CLASS;String clsName = String.valueOf(config.get(configKey));try {cls = (Class<T>) Class.forName(clsName);} catch (ClassNotFoundException e) {System.err.printf("Failed to configure GsonDeserializer. " +"Did you forget to specify the '%s' Stringproperty ?%n",configKey);}}@Overridepublic T deserialize(String topic, byte[] bytes) {try {return (T) gson.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes),"UTF-8"), cls);} catch (UnsupportedEncodingException e) {log.error("Failed to desserialize ", e);e.printStackTrace();return null;}}@Overridepublic void close() {}
}

上面是反序列化类

public class DateTimeSerializer implements JsonSerializer {private DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:SS");private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:SS");@Overridepublic JsonElement serialize(Object o, Type type, JsonSerializationContext jsonSerializationContext) {if((o instanceof LocalDateTime || type == LocalDateTime.class) && o != null){return new JsonPrimitive(((LocalDateTime) o).format(dateTimeFormatter).toString());}if((o instanceof Date || type == LocalDateTime.class) && o != null){return new JsonPrimitive(sdf.format((Date) o));}return null;}
}

上面是时间序列化类

public class DateTimeDeserializer implements JsonDeserializer {String pattern = "yyyy-MM-dd HH:mm:SS";DateFormat df = new SimpleDateFormat(pattern);@Overridepublic Date deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {try {String result = json.getAsJsonPrimitive().getAsString();return df.parse(result);} catch (ParseException e) {e.printStackTrace();}return null;}}

上面是时间反序列化类

将上面的序列化工具类加入的kafka配置里面

@Configuration
public class KafkaConfig {@Autowiredprivate KafkaProperties kafkaProperties;private String MQ_TOPIC_SSO_USERCHANGED = KafkaParams.MQ_TOPIC_SSO_USERCHANGED;// Producer configuration@Beanpublic Map<String, Object> producerConfigs() {Map<String, Object> props =new HashMap<>(kafkaProperties.buildProducerProperties());props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,GsonSerializer.class);return props;}@Beanpublic ProducerFactory<String, Object> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}@Beanpublic KafkaTemplate<String, Object> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}@Beanpublic NewTopic adviceTopic() {return new NewTopic(MQ_TOPIC_SSO_USERCHANGED, 3, (short) 1);}}