编写自定义序列化类

xxxxxxxxxx
package com.itbaizhan.kafka.producer;

import org.apache.kafka.common.serialization.Serializer;
import org.codehaus.jackson.map.ObjectMapper;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class UserSerializer implements Serializer<UserVo> {
    private ObjectMapper objectMapper;
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        objectMapper = new ObjectMapper();
        //Serializer.super.configure(configs, isKey);
    }

    @Override
    public byte[] serialize(String topic, UserVo data) {
        byte[] ret = null;
        try {
            ret = objectMapper.writeValueAsString(data)
                    .getBytes(StandardCharsets.UTF_8);
        } catch (IOException e) {
            throw new SerializationException("Error when serializing UserVo to byte[],exception is " + e.getMessage());
        }
        return ret;
    }

    @Override
    public void close() {
        objectMapper = null;
        //Serializer.super.close();
    }
}

编写生产者程序

xxxxxxxxxx
package com.itbaizhan.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class UserSerProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //TODO 1.声明并实例化Kafka Producer的配置文件对象
        Properties prop = new Properties();
        //TODO 2.为配置文件对象设置参数
        // 2.1 配置bootstrap_servers
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node2:9092,node3:9092,node4:9092");
        // 2.2 配置key和value的序列化类
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserSerializer.class.getName());
        //TODO 3.声明并实例化生产者对象 注意value的泛型类型
        KafkaProducer<String,UserVo> producer = new KafkaProducer<String, UserVo>(prop);
        //TODO 4.发送消息
        UserVo userVo = new UserVo("tuhao",18,"北京");
        producer.send(new ProducerRecord<String,UserVo>("topicA", userVo),
            new Callback() {
                //如下方法在生产者收到acks确认时异步调用
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(e == null){
                        //无异常信息，输出主题和分区信息到控制台
                        System.out.println("topic:"+recordMetadata.topic()
                                +",partition:"+recordMetadata.partition());
                    }else{//打印异常信息
                        System.out.println(e.getMessage());
                    }
                }
            });
        Thread.sleep(50);
        //TODO 5.关闭生产者
        producer.close();
    }
}

在node2上开启Kafka消费者进行消费

xxxxxxxxxx
[root@node2 ~]# kafka-console-consumer.sh --bootstrap-server node2:9092 --topic topicA

运行UserSerProducer类

观察node2上Kafka消费者消费消息的情况

xxxxxxxxxx
[root@node2 ~]# kafka-console-consumer.sh --bootstrap-server node2:9092 --topic topicA
{"name":"tuhao","age":18,"address":"北京"}