spring:
kafka:
# 以逗号分隔的地址列表，用于建立与Kafka集群的初始连接(kafka 默认的端口号为9092)
bootstrap-servers: ip1:port1,ip2:port2,ip3:port3
producer:
key-serializer: org.apache.kafka.common.serialization.StringDeserializer
# 值的序列化方式
value-serializer: org.apache.kafka.common.serialization.StringDeserializer
序列化方式换成 value-serializer:：com.xxx.xxx.ObjectSerializer
反序列化方式换成value-serializer:：com.xxx.xxx.ObjectDeserializer
注意：kafka属性配置文件中红色，可以满足只发送字符串数据时候用。然而想发送对象数据，就得用以下自定义序列类来实现。java代码如下：
1.ObjectSerializer.java
public class ObjectSerializer implements Serializer
2.ObjectDeserializer.java
public class ObjectDeserializer implements Deserializer
3.BeanUtils.java
public class BeanUtils {
private BeanUtils(){}
public static byte[] ObjectToBytes(Object obj){
byte[] bytes = null;
ByteArrayOutputStream bo = null;
ObjectOutputStream oo = null;
try {
bo = new ByteArrayOutputStream();
oo = new ObjectOutputStream(bo);
oo.writeObject(obj);
bytes = bo.toByteArray();
} catch (IOException e) {
e.printStackTrace();
}finally {
try {
if(bo!=null){
bo.close();
}
if(oo!=null){
oo.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
return bytes;
}
public static Object BytesToObject(byte[] bytes){
Object obj = null;
ByteArrayInputStream bi = null;
ObjectInputStream oi = null;
try {
bi =new ByteArrayInputStream(bytes);
oi =new ObjectInputStream(bi);
obj = oi.readObject();
} catch (Exception e) {
e.printStackTrace();
}finally {
try {
if(bi!=null){
bi.close();
}
if(oi!=null){
oi.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
return obj;
}
}
4.发送对象代码
public void sendObj(){
try{
log.info("发送给kafka 发对象开始");
Map m = new HashMap();
m.put("name", "jim");
kafkaTemplate.send(topicName, m);//或者这里发字符串也一样可用
log.info("发送给kafka结束");
}catch(Exception e){
e.printStackTrace();
}
}
5.消费对象代码
@Component
public class KafkaConsumer {
// 消费监听
@KafkaListener(topics = {"${config.kafka.send-topic}"},groupId = "${config.kafka.group-id}")
public void onMessage1(ConsumerRecord record){
String a = record.value().toString();
log.info("消费的数据："+ a);
}
}
