首先定义一个接口

public interface Protobufable {

	//将对象转为字节数组
	public byte[] encode();

}

然后User 实现这个接口，这里可以不实现Serializable接口，User 类代码如下：

public class User implements Serializable,Protobufable{
	private static final long serialVersionUID = 468062760765055608L;

	private Long id;
	
	private String name;
	
	private String email;
	/** {0:男,1:女} **/
	private Integer sex;
	
	public User() {}
	
	public User(Long id, String name, String email, Integer sex) {
		super();
		this.id = id;
		this.name = name;
		this.email = email;
		this.sex = sex;
	}
	
	/***
	 * 将byte解析为User对象
	 * @param bytes
	 */
	public User(byte[] bytes) {
		try {
			//将字节数组转换为UserProto.User对象
			UserProto.User user = UserProto.User.parseFrom(bytes);
			//UserProto.User对象转化为自己的User对象
			this.id = user.getId();
			this.name = user.getName();
			this.email = user.getEmail();
			this.sex = user.getSex();
		} catch (InvalidProtocolBufferException e) {
			e.printStackTrace();
		}
	}

	/** 编码 */
	@Override
	public byte[] encode() {
		UserProto.User.Builder builder = UserProto.User.newBuilder();
		builder.setId(id);
		builder.setName(name);
		builder.setEmail(email);
		builder.setSex(sex);
		return builder.build().toByteArray();
	}
	
	@Override
	public String toString() {
		return "[ID:" + id + ", 姓名：" + name + ", 性别：" + (sex==0?"男":"女") + ", 邮箱：" + email + "]";
	}

	/********************** getter & setter******************************/
	public Long getId() { return id; }
	public void setId(Long id) { this.id = id; }

	public String getName() { return name; }
	public void setName(String name) { this.name = name; }

	public String getEmail() { return email; }
	public void setEmail(String email) { this.email = email; }

	public Integer getSex() { return sex; }
	public void setSex(Integer sex) { this.sex = sex; }
	/********************** getter & setter******************************/
}

再然后，实现一个ProtobufSerializer 的序列化器，调用的实际上就是user.encode()，代码如下：

/**
 * protobuf序列化方式实现kafka消息的的序列化
 */
public class ProtobufSerializer implements Serializer<Protobufable>{
	
	@Override
	public void configure(Map<String, ?> configs, boolean isKey) {}
	
	@Override
	public byte[] serialize(String topic, Protobufable data) {
		return data.encode();
	}

	@Override
	public void close() {}
}

生产者代码：

/**
 * 生产者-使用Protobuf序列化
 */
public class ProtobufSerializerProducer {

	public static final String TOPIC_NAME = "producer-0"; 
	private  static Properties props = new Properties();
	
	static{
		 props.put("bootstrap.servers", "192.168.1.3:9092,192.168.1.128:9092,192.168.1.130:9092");
		 props.put("acks", "all");
		 props.put("retries", 0);
		 props.put("batch.size", 16384);
		 props.put("linger.ms", 1);
		 props.put("buffer.memory", 33554432);
		 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		 props.put("value.serializer", "com.yang.kafka.serialization.ProtobufSerializer");
	}
	
	public static void main(String[] args) {
		 Producer<String, User> producer = new KafkaProducer<>(props);
		 
		 User user = new User(101L,"kafka","serializer@kafka.com",1);
		 producer.send(new ProducerRecord<String, User>(TOPIC_NAME, Long.toString(user.getId()), user));
		 
		 producer.close();
	}
}

消费者代码如下,这里对value.deserializer换成了ByteArrayDeserializer，其实就是返回的字节数组

/**
 * 消费者-使用Protobuf反序列化
 */
public class ProtobufDeserializerConsumer {
	private  static Properties props = new Properties();
	
	private static boolean isClose = false;
	
	static{
	     props.put("bootstrap.servers", "192.168.1.3:9092,192.168.1.128:9092,192.168.1.130:9092");
	     props.put("group.id", "test");
	     props.put("enable.auto.commit", "true");
	     props.put("auto.commit.interval.ms", "1000");
	     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	     props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
	}
	
	public  static void main(String args[]){
	     KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
	     consumer.subscribe(Arrays.asList(JSONSerializerProducer.TOPIC_NAME));
	     
	     while (!isClose) {
	         ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
	         for (ConsumerRecord<String, byte[]> record : records)
	        	 System.out.printf("key = %s, value = %s%n", record.key(), new User(record.value()));
	     }
	     
	     consumer.close();
	}
}