Kafka 生产者解析

public class InterceptorOne<Key, Value> implements ProducerInterceptor<Key, Value> {
private static final Logger LOGGER = LoggerFactory.getLogger(InterceptorOne.class);


@Override
public ProducerRecord<Key, Value> onSend(ProducerRecord<Key, Value> record) {
System.out.println("拦截器1---go");
// 此处根据业务需要对相关的数据作修改
String topic = record.topic();
Integer partition = record.partition();
Long timestamp = record.timestamp();
Key key = record.key();
Value value = record.value();
Headers headers = record.headers();
// 添加消息头
headers.add("interceptor", "interceptorOne".getBytes());
ProducerRecord<Key, Value> newRecord = new ProducerRecord<Key, Value>(topic,
partition, timestamp, key, value, headers);
return newRecord;
}


@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
System.out.println("拦截器1---back");
if (exception != null) {
// 如果发⽣异常，记录⽇志中
LOGGER.error(exception.getMessage());
}
}


@Override
public void close() {


}


@Override
public void configure(Map<String, ?> configs) {


}
}

public class MyProducer1 {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
Map<String, Object> configs = new HashMap<>();
// 设置连接Kafka的初始连接⽤到的服务器地址
// 如果是集群，则可以通过此初始连接发现集群中的其他broker
configs.put("bootstrap.servers", "192.168.0.102:9092");
// 设置key的序列化器
configs.put("key.serializer", IntegerSerializer.class);
// 设置⾃定义的序列化类
configs.put("value.serializer", UserSerializer.class);
// 设置自定义分区器
configs.put("partitioner.class", "com.mfc.config.MyPartitioner");
// 设置拦截器
configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
"com.mfc.interceptor.InterceptorOne,"
+ "com.mfc.interceptor.InterceptorTwo,"
+ "com.mfc.interceptor.InterceptorThree");


KafkaProducer<Integer, User> producer = new KafkaProducer<>(configs);
User user = new User();
user.setUserId(1001);
user.setUsername("阿彪");


// ⽤于封装Producer的消息
ProducerRecord<Integer, User> record = new ProducerRecord<>(
"topic_1", // 主题名称
0, // 分区编号
user.getUserId(), // 数字作为key
user // user 对象作为value
);
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception e) {
if (e == null) {
System.out.println("消息发送成功：" + metadata.topic() + "\t"
+ metadata.partition() + "\t"
+ metadata.offset());
} else {
System.out.println("消息发送异常");
}
}
});


// 关闭⽣产者
producer.close();
}
}

package org.apache.kafka.common.serialization;


import java.io.Closeable;
import java.util.Map;


/**
将对象转换为byte数组的接⼝
该接⼝的实现类需要提供⽆参构造器
@param <T> 从哪个类型转换
*/
public interface Serializer<T> extends Closeable {
/*
    类的配置信息
    @param configs key/value pairs
    @param isKey key的序列化还是value的序列化
    */
void configure(Map<String, ?> var1, boolean var2);


/*
    将对象转换为字节数组
     @param topic 主题名称
     @param data 需要转换的对象
     @return 序列化的字节数组
    */
byte[] serialize(String var1, T var2);


/*
    关闭序列化器
    该⽅法需要提供幂等性，因为可能调⽤多次。
    */
void close();
}

public class User {
private Integer userId;
private String username;
// set、get方法省略
}

public class UserSerializer implements Serializer<User> {
@Override
public void configure(Map<String, ?> map, boolean b) {
// do Nothing
}


@Override
public byte[] serialize(String topic, User user) {
try {
// 如果数据是null，则返回null
if (user == null) return null;
Integer userId = user.getUserId();
String username = user.getUsername();
int length = 0;
byte[] bytes = null;
if (null != username) {
bytes = username.getBytes("utf-8");
length = bytes.length;
}
ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + length);
buffer.putInt(userId);
buffer.putInt(length);
buffer.put(bytes);
return buffer.array();
} catch (UnsupportedEncodingException e) {
throw new SerializationException("序列化数据异常");
}
}


@Override
public void close() {
// do Nothing
}
}

public class MyProducer1 {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
Map<String, Object> configs = new HashMap<>();
// 设置连接Kafka的初始连接⽤到的服务器地址
// 如果是集群，则可以通过此初始连接发现集群中的其他broker
configs.put("bootstrap.servers", "192.168.0.102:9092");
// 设置key的序列化器
configs.put("key.serializer", IntegerSerializer.class);
// 设置⾃定义的序列化类
configs.put("value.serializer", UserSerializer.class);


KafkaProducer<Integer, User> producer = new KafkaProducer<>(configs);
User user = new User();
user.setUserId(1001);
user.setUsername("阿彪");


// ⽤于封装Producer的消息
ProducerRecord<Integer, User> record = new ProducerRecord<>(
"topic_1", // 主题名称
0, // 分区编号
user.getUserId(), // 数字作为key
user // user 对象作为value
);
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception e) {
if (e == null) {
System.out.println("消息发送成功：" + metadata.topic() + "\t"
+ metadata.partition() + "\t"
+ metadata.offset());
} else {
System.out.println("消息发送异常");
}
}
});


// 关闭⽣产者
producer.close();
}
}

public interface Partitioner extends Configurable, Closeable {


/**
     * 为指定的消息记录计算分区值
     *
     * @param topic 主题名称
     * @param key 根据该key的值进⾏分区计算，如果没有则为null
     * @param keyBytes key的序列化字节数组，根据该数组进⾏分区计算。如果没有key，则为null
     * @param value 根据value值进⾏分区计算，如果没有，则为null
     * @param valueBytes value的序列化字节数组，根据此值进⾏分区计算。如果没有，则为null
     * @param cluster 当前集群的元数据
     */
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);


/**
     * 关闭分区器的时候调⽤该⽅法
     */
public void close();


}

public class MyPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
return 0;
}


@Override
public void close() {


}


@Override
public void configure(Map<String, ?> configs) {


}
}