@AllArgsConstructor
@Data
public class User {

    private String name;
    private int age;
}

client.newProducer(Schema.AVRO(User.class))
        .topic("my-topic")
        .create()
        .newMessage()
        .value(new User("xyz", 11))
        .send();

public interface SchemaInfo {

    // schema 名字
    String getName();

    // 对于 AVRO 类型，为 Avro schema 序列化后的字节数组
    byte[] getSchema();

    // schema 类型，比如 AVRO
    SchemaType getType();
  
    // 用户自定义的一组属性
    Map<String, String> getProperties();

public interface Schema<T> extends Cloneable{

    // 将对象按照 schema 定义编码成字节数组
    byte[] encode(T message);

    // 将字节数组按照 schema 定义解码成对象，默认忽略 schema version
    default T decode(byte[] bytes) {
        return decode(bytes, null);
    }
    default T decode(byte[] bytes, byte[] schemaVersion) {
        return decode(bytes);
    }

private SchemaData getSchema(Schema protocolSchema) {
    return SchemaData.builder()
        .data(protocolSchema.getSchemaData())
        .isDeleted(false)
        .timestamp(System.currentTimeMillis())
        .user(Strings.nullToEmpty(originalPrincipal))
        .type(Commands.getSchemaType(protocolSchema.getType()))
        .props(protocolSchema.getPropertiesList().stream().collect(
            Collectors.toMap(
                KeyValue::getKey,
                KeyValue::getValue
            )
        )).build();
}

private CompletableFuture<SchemaVersion> tryAddSchema(Topic topic, SchemaData schema) {
    if (schema != null) {
        return topic.addSchema(schema);
    } else {

if (schema != null) {
    return topic.addSchemaIfIdleOrCheckCompatible(schema)
            .thenCompose(v -> topic.subscribe(option));

public TypedMessageBuilder<T> newMessage() {
    return new TypedMessageBuilderImpl<>(this, schema);
}

public TypedMessageBuilder<T> value(T value) {
    /* ... */
    this.content = ByteBuffer.wrap(schema.encode(value));
    return this;
}

final MessageImpl<V> message = MessageImpl.create(topicName.toString(), batchMessageIdImpl,
        msgMetadata, singleMessageMetadata, payloadBuffer,
        createEncryptionContext(msgMetadata), cnx(), schema, redeliveryCount, poolMessages, consumerEpoch);

public T getValue() {
    SchemaInfo schemaInfo = getSchemaInfo();
    if (schemaInfo != null && SchemaType.KEY_VALUE == schemaInfo.getType()) {
        /* ... */
    } else {
        /* ... */
        return decode(schema.supportSchemaVersioning() ? getSchemaVersion() : null);
    }
}

private T decode(byte[] schemaVersion) {
    // 对于设置了 poolMessage 的消息，直接对 payload 内部的 NIO buffer（可能在堆外）进行解码
    // 否则会对 getData() 进行解码，getData() 会拷贝一份 payload
    T value = poolMessage ? schema.decode(payload.nioBuffer(), schemaVersion) : null;
    if (value != null) {
        return value;
    }
    // 注：下面代码有些啰嗦，直接 schema.decode(getData(), schemaVersion) 就行
    if (null == schemaVersion) {
        return schema.decode(getData());
    } else {
        return schema.decode(getData(), schemaVersion);
    }
}

static <T> Schema<T> AVRO(Class<T> pojo) {
    // 实际上也是将 Class 对象转换成 SchemaDefinition 对象来构造
    return DefaultImplementation.getDefaultImplementation()
            .newAvroSchema(SchemaDefinition.builder().withPojo(pojo).build());
}

static <T> Schema<T> AVRO(SchemaDefinition<T> schemaDefinition) {
    return DefaultImplementation.getDefaultImplementation().newAvroSchema(schemaDefinition);
}

private Class<T> pojo;

public static <T> AvroSchema<T> of(SchemaDefinition<T> schemaDefinition) {
    /* 这里会首先处理 schema reader/writer 用于支持 multi-schema，这里我们暂不关心这个特性 */
    // getPojo() 返回的用户提供的 T 对应的 Class 对象，这里取得它的 ClassLoader
    ClassLoader pojoClassLoader = null;
    if (schemaDefinition.getPojo() != null) {
        pojoClassLoader = schemaDefinition.getPojo().getClassLoader();
    }
    return new AvroSchema<>(parseSchemaInfo(schemaDefinition, SchemaType.AVRO), pojoClassLoader);
}

private AvroSchema(SchemaInfo schemaInfo, ClassLoader pojoClassLoader) {
    super(schemaInfo);
    this.pojoClassLoader = pojoClassLoader;
    /* 设置 schema reader/writer (略) */
}

protected final SchemaInfo schemaInfo;

public AbstractStructSchema(SchemaInfo schemaInfo) {
    this.schemaInfo = schemaInfo;
}

// org.apache.avro.schema
protected final Schema schema;

public AvroBaseStructSchema(SchemaInfo schemaInfo) {
    super(schemaInfo);
    this.schema = parseAvroSchema(new String(schemaInfo.getSchema(), UTF_8));
}

public static <T> SchemaInfo parseSchemaInfo(SchemaDefinition<T> schemaDefinition, SchemaType schemaType) {
    return SchemaInfoImpl.builder()
            .schema(createAvroSchema(schemaDefinition).toString().getBytes(UTF_8))
            .properties(schemaDefinition.getProperties())
            .name("")
            .type(schemaType).build();
}

// 将 SchemaDefinition 转换成 Avro Schema 对象
public static Schema createAvroSchema(SchemaDefinition schemaDefinition) {
    Class pojo = schemaDefinition.getPojo();

    if (StringUtils.isNotBlank(schemaDefinition.getJsonDef())) {
        return parseAvroSchema(schemaDefinition.getJsonDef());
    } else if (pojo != null) { // 在之前通过 withPojo 已经设置过 pojo
        ThreadLocal<Boolean> validateDefaults = null;
        try {
            Field validateDefaultsField = Schema.class.getDeclaredField("VALIDATE_DEFAULTS");
            validateDefaultsField.setAccessible(true);
            validateDefaults = (ThreadLocal<Boolean>) validateDefaultsField.get(null);
        } catch (NoSuchFieldException | IllegalAccessException e) {
            throw new RuntimeException("Cannot disable validation of default values", e);
        }

        final boolean savedValidateDefaults = validateDefaults.get();

        try {
            // Disable validation of default values for compatibility
            validateDefaults.set(false);
            return extractAvroSchema(schemaDefinition, pojo);
        } finally {
            validateDefaults.set(savedValidateDefaults);
        }
    } else {
        throw new RuntimeException("Schema definition must specify pojo class or schema json definition");
    }
}

public static Schema extractAvroSchema(SchemaDefinition schemaDefinition, Class pojo) {
    try {
        // 一般 pojo 是不包含 SCHEMA$ 这个字段的，因此会进入 catch 分支
        return parseAvroSchema(pojo.getDeclaredField("SCHEMA$").get(null).toString());
    } catch (NoSuchFieldException | IllegalAccessException | IllegalArgumentException ignored) {
        // SchemaDefinition 的 alwaysAllowNull 默认为 true，因此回创建 AllowNull 对象
        ReflectData reflectData = schemaDefinition.getAlwaysAllowNull()
                 ? new ReflectData.AllowNull()
                 : new ReflectData();
        // 给 reflectdata 添加一些逻辑类型（主要是日期和时间）的转换
        AvroSchema.addLogicalTypeConversions(reflectData, schemaDefinition.isJsr310ConversionEnabled(), false);
        return reflectData.getSchema(pojo);
    }
}

.schema(createAvroSchema(schemaDefinition).toString().getBytes(UTF_8))

public static Schema parseAvroSchema(String schemaJson) {
    final Schema.Parser parser = new Schema.Parser();
    parser.setValidateDefaults(false);
    return parser.parse(schemaJson);
}

protected SchemaReader<T> reader;
protected SchemaWriter<T> writer;

@Override
public byte[] encode(T message) {
    return writer.write(message);
}

@Override
public T decode(byte[] bytes) {
    return reader.read(bytes);
}

private AvroSchema(SchemaInfo schemaInfo, ClassLoader pojoClassLoader) {
    /* ... */
    // 用于从 SchemaInfo 的属性中取得 __jsr310ConversionEnabled 来判断是否支持 JSR 310
    // 从而兼容旧的 Pulsar 客户端（不支持 JSR 310 转换）
    boolean jsr310ConversionEnabled = getJsr310ConversionEnabledFromSchemaInfo(schemaInfo);
    setReader(new MultiVersionAvroReader<>(schema, pojoClassLoader,
            getJsr310ConversionEnabledFromSchemaInfo(schemaInfo)));
    setWriter(new AvroWriter<>(schema, jsr310ConversionEnabled));
}

public AvroWriter(Schema schema, boolean jsr310ConversionEnabled) {
    this.byteArrayOutputStream = new ByteArrayOutputStream();
    this.encoder = EncoderFactory.get().binaryEncoder(this.byteArrayOutputStream, null);
    ReflectData reflectData = new ReflectData();
    // 这里会决定是否进行 JSR 310 转换
    AvroSchema.addLogicalTypeConversions(reflectData, jsr310ConversionEnabled);
    this.writer = new ReflectDatumWriter<>(schema, reflectData);
}

@Override
public synchronized byte[] write(T message) {
    byte[] outputBytes = null;
    try {
        writer.write(message, this.encoder);
    } catch (Exception e) {
        throw new SchemaSerializationException(e);
    } finally {
        try {
            this.encoder.flush();
            // 从 write 的输出流中取出编码后的字节
            outputBytes = this.byteArrayOutputStream.toByteArray();
        } catch (Exception ex) {
            throw new SchemaSerializationException(ex);
        }
        this.byteArrayOutputStream.reset();
    }
    return outputBytes;
}

public MultiVersionAvroReader(Schema readerSchema, ClassLoader pojoClassLoader, boolean jsr310ConversionEnabled) {
    super(new AvroReader<>(readerSchema, pojoClassLoader, jsr310ConversionEnabled), readerSchema);

public AbstractMultiVersionReader(SchemaReader<T> providerSchemaReader) {
    this.providerSchemaReader = providerSchemaReader;
}

@Override
public T read(byte[] bytes, int offset, int length) {
    return providerSchemaReader.read(bytes);
}

public AvroReader(Schema schema, ClassLoader classLoader, boolean jsr310ConversionEnabled) {
    this.schema = schema;
    if (classLoader != null) {
        ReflectData reflectData = new ReflectData(classLoader);
        AvroSchema.addLogicalTypeConversions(reflectData, jsr310ConversionEnabled);
        this.reader = new ReflectDatumReader<>(schema, schema, reflectData);
    } else {
        this.reader = new ReflectDatumReader<>(schema);
    }
}

public T read(InputStream inputStream) {
    try {
        BinaryDecoder decoderFromCache = decoders.get();
        BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(inputStream, decoderFromCache);
        if (decoderFromCache == null) {
            decoders.set(decoder);
        }
        return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, decoder));

final List<Integer> list1 = new ArrayList<>(Arrays.asList(1, 2, 3));
final List<Integer> list2 = new ArrayList<>(list1.size());
list1.forEach(x -> list2.add(x + 100));
final List<String> list3 = new ArrayList<>(list2.size());
list2.forEach(x -> list3.add("i-" + x));

final List<String> list3 = list1.map(x -> x + 100).map(x -> "i-" + x);

final List<String> list3 = list1.stream().map(x -> x + 100).map(x -> "i-" + x).collect(Collectors.toList());

<R> List<R> map(Function<? super T, ? extends R> mapper)

Arrays.stream(new int[]{0, 1, 2});
Stream.of(1, 2, 3);
IntStream.range(0, 3);
LongStream.range(0L, 3L);

Stream<Integer> stream = Stream.of(1, 2, 3);
stream.map(x -> x + 10);
stream.map(x -> x + 10);

final List<Integer> list = Arrays.asList(1, 2, 3);
int sum = 0;
for (Integer x : list) {
    sum += x;
}

final int sum = Stream.of(1, 2, 3).reduce(0, Integer::sum);

combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)

final List<Integer> list = Arrays.asList(1, 2, 3);
final String result = list.stream().reduce("prefix" , (s, i) -> s + "-" + i, (x, y) -> x);
System.out.println(result)

final String result = list.parallelStream().reduce("prefix" , (s, i) -> s + "-" + i,
        (s1, s2) -> s1 + s2.substring("prefix".length()));

final List<Integer> list1 = Arrays.asList(1, 2, 3);
final List<Integer> list2 = Arrays.asList(4, 5);
final List<Integer> list3 = Arrays.asList(6, 7, 8, 9);

final List<Integer> list = new ArrayList<>(list1);
list.addAll(list2);
list.addAll(list3);

final List<Integer> list = Stream.of(list1, list2, list3).reduce(new ArrayList<>(), (lhs, rhs) -> {
    lhs.addAll(rhs);
    return lhs;
});

<R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);

final List<Integer> list = Stream.of(list1, list2, list3).flatMap(List::stream).collect(Collectors.toList());

final List<List<Integer>> list1 = Arrays.asList(Arrays.asList(1, 2, 3), Arrays.asList(4, 5));
final List<List<Integer>> list2 = Collections.singletonList(Arrays.asList(6, 7, 8, 9));
final List<Integer> list = Stream.of(list1, list2)
        .flatMap(Collection::stream)
        .flatMap(Collection::stream)
        .collect(Collectors.toList());

public static <T, K> Collector<T, ?, Map<K, List<T>>>
groupingBy(Function<? super T, ? extends K> classifier) {
    return groupingBy(classifier, toList());
}

final List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
final Map<Boolean, List<Integer>> map = list.stream().collect(Collectors.groupingBy(i -> i % 2 != 0));
System.out.println(map.get(true)); // 奇数
System.out.println(map.get(false)); // 偶数

final List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
final List<Integer> odds = list.stream().filter(x -> x % 2 != 0).collect(Collectors.toList());
final List<Integer> evens = list.stream().filter(x -> x % 2 == 0).collect(Collectors.toList());
System.out.println(odds);
System.out.println(evens);

final Map<Integer, Integer> map = new HashMap<>();
map.put(1, 100);
map.put(2, 200);
System.out.println(map.entrySet().stream().collect(Collectors.toMap(
        Map.Entry::getKey,
        e -> "value-" + e.getValue())
));
System.out.println(map.entrySet().stream().collect(Collectors.toMap(
        e -> "key-" + e.getKey(),
        e -> "value-" + e.getValue())
));

<R, A> R collect(Collector<? super T, A, R> collector)

public static <T>
Collector<T, ?, List<T>> toList() {
    return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,
                               (left, right) -> { left.addAll(right); return left; },
                               CH_ID);
}

public interface Executor {
    void execute(Runnable command);
}

Executor executor = command -> new Thread(command).start()

final Executor executor = Executors.newSingleThreadExecutor(); // 该线程池仅包含一个可用线程
executor.execute(() -> System.out.println(Thread.currentThread().getName() + " done"))

Runnable r = timed ?
    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
    workQueue.take(); // 卡在这里

private final BlockingQueue<Runnable> workQueue;

final void runWorker(Worker w) {
    /* ... */
    try {
        while (task != null || (task = getTask()) != null) {

void shutdown();

ExecutorService executor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 3; i++) {
    final int index = i;
    executor.execute(() -> {
        println(index + " started...");
        sleep(1000);
        println(index + " stopped.");
    });
}
executor.shutdown();
println("Executor shutdown: " + executor.isShutdown());
try {
    executor.execute(() -> {});
} catch (RejectedExecutionException e) {
    println("Failed to execute task after shutdown: " + e.getMessage());
}

boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException;

println("Executor terminated: " + executor.isTerminated());
while (!executor.awaitTermination(10000, TimeUnit.MILLISECONDS)) {
    // No ops
}
println("Executor terminated: " + executor.isTerminated());

List<Runnable> shutdownNow();

ExecutorService executor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 3; i++) {
    final int index = i;
    executor.execute(() -> {
        println(index + " started...");
        try {
            Thread.sleep(1000);
            println(index + " stopped.");
        } catch (InterruptedException e) {
            println(index + " is cancelled");
        }
    });
}
executor.shutdownNow();
executor.awaitTermination(1, TimeUnit.MINUTES);

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService // 仅仅是个 wrapper，finalize() 方法会调用 shutdown()
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

final ExecutorService executor = new ThreadPoolExecutor(
        1, // corePoolSize
        3, // maximumPoolSize
        1, // keepAliveTime
        TimeUnit.SECONDS, // unit
        new ArrayBlockingQueue<>(1) // workQueue
        );
for (int i = 0; i < 5; i++) {
    final int index = i;
    try {
        executor.execute(() -> {
            println(index + " started...");
            sleep(2000);
            println(index + " stopped.");
        });
    } catch (RejectedExecutionException e) {
        println(index + " rejected: " + e.getMessage());
    }
}
executor.shutdown();
while (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
    // No ops
}

// 必须对 ThreadPoolExecutor 调用 allowCoreThreadTimeOut 方法将同名参数设为 true 才会启用 keepAliveTime 检查
// wc 为工作线程数量，因此只有大于 corePoolSize 才会检查，因为只有超过核心运行线程数量的线程才会被当成多余的线程
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

if ((wc > maximumPoolSize || (timed && timedOut))
    && (wc > 1 || workQueue.isEmpty())) {
    if (compareAndDecrementWorkerCount(c)) // [2] 回收多余线程，减少工作线程数量
        return null;
    continue;
}

try {
    Runnable r = timed ?
        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
        workQueue.take();
    if (r != null)
        return r;
    timedOut = true; // [1] 在 keepAliveTime 内没有任务，则置为 true
} catch (InterruptedException retry) {
    timedOut = false;
}

this.keepAliveTime = unit.toNanos(keepAliveTime);

public interface ThreadFactory {
    Thread newThread(Runnable r);
}

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

DefaultThreadFactory() {
    SecurityManager s = System.getSecurityManager();
    group = (s != null) ? s.getThreadGroup() :
                          Thread.currentThread().getThreadGroup();
    namePrefix = "pool-" +
                  poolNumber.getAndIncrement() +
                 "-thread-";
}

public Thread newThread(Runnable r) {
    Thread t = new Thread(group, r,
                          namePrefix + threadNumber.getAndIncrement(),
                          0);
    if (t.isDaemon())
        t.setDaemon(false); // 对于守护线程，也要将其改成用户线程
    if (t.getPriority() != Thread.NORM_PRIORITY)
        t.setPriority(Thread.NORM_PRIORITY); // 默认线程优先级是 NORM
    return t;
}

ExecutorService executor1 = Executors.newSingleThreadExecutor();
ExecutorService executor2 = Executors.newSingleThreadExecutor();
executor1.execute(() -> println("executor 1 thread 1"));
executor2.execute(() -> println("executor 2 thread 1"));
executor2.execute(() -> println("executor 2 thread 2"));
executor1.shutdown();
executor2.shutdown();

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>(),
                                threadFactory));
}

private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();

public static class AbortPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

private static final List<Integer> DATA = prepareData();

private static List<Integer> prepareData() {
    final List<Integer> data = new ArrayList<>();
    for (int i = 0; i < 100000; i++) {
        data.add(i);
    }
    return data;
}

public static void main(String[] args) throws InterruptedException, ExecutionException {
    final int numThreads = 5;
    final int sizePerThread = DATA.size() / numThreads;
    final List<Future<Long>> futures = new ArrayList<>();
    final ExecutorService executor = Executors.newFixedThreadPool(numThreads);
    for (int i = 0; i < numThreads; i++) {
        final int startIndex = i * sizePerThread;
        final int endIndex = (i < numThreads - 1) ? (startIndex + sizePerThread) : DATA.size();
        futures.add(executor.submit(() -> {
            long sum = 0;
            for (int j = startIndex; j < endIndex; j++) {
                sum += DATA.get(j);
            }
            return sum; // 该 lambda 返回 Long，对应 Callable<Long>，因此 submit 返回的是 Future<Long>
        }));
    }
    long sum = 0;
    for (Future<Long> future : futures) {
        sum += future.get();
    }
    System.out.println(sum);
    executor.shutdown();
}

 for i := 0; i < 100; i++ {
	if i%3 == 0 {
		fmt.Println(i)
	}
}