卡夫卡的藏书阁【Book16】- Kafka - KafkaJS 生产者 - 4

“It's only because of their stupidity that they're able to be so sure of themselves.”
― Franz Kafka, The Trial
认识自己，方能认识人生

生产者如何传送讯息 ( message ) 到多个主题 ( topic )

生产者要同时传送讯息到多个主题可以使用 sendBatch，这个在做聚合资料时十分有用。

const topicMessages = [
  {
    topic: 'topic-a',
    messages: [{ key: 'key', value: 'hello topic-a' }],
  },
  {
    topic: 'topic-b',
    messages: [{ key: 'key', value: 'hello topic-b' }],
  },
  {
    topic: 'topic-c',
    messages: [
      {
        key: 'key',
        value: 'hello topic-c',
        headers: {
          'correlation-id': '2bfb68bb-893a-423b-a7fa-7b568cad5b67',
        },
      }
    ],
  }
]
await producer.sendBatch({ topicMessages })

sendBatch 使用方式跟 send 一样，只是将 topic 和 message 合并成 topicMessages

await producer.sendBatch({
    topicMessages: <TopicMessages[]>,
    acks: <Number>,
    timeout: <Number>,
    compression: <CompressionTypes>,
})

topicMessages 是一个阵列，包含一个或多个物件，物件内容是 topic 和 messages，messages 的格式跟 send 使用的 messages 是一样的。

客制化属於自己的分区器

生产者是可以接受客制化的分区器，一个分区器是一个函式，它会回传另外一个实际用来写分区逻辑的函式

const MyPartitioner = () => {
    return ({ topic, partitionMetadata, message }) => {
        // select a partition based on some logic
        // return the partition number
        return 0
    }
}

partitionMetadata 是一个阵列里面包含分区和领袖的资讯，它的格式如下：
{ partitionId: , leader: }`

Example:
[
{ partitionId: 1, leader: 1 },
{ partitionId: 2, leader: 2 },
{ partitionId: 0, leader: 0 }
]

使用客制化分区器的方式很简单，只要在新增生产者时，使用可选参数 createPartitioner 即可，范例如下：
kafka.producer({ createPartitioner: MyPartitioner })

预设的分区器
KafkaJS随附有两个预设的分区器：DefaultPartitioner 和 JavaCompatiblePartitioner
JavaCompatiblePartitioner 理论上是可以跟 Java Kafka client 随附的预设分区器相容的，是否相容这一点很重要，会影响到多个主题聚合的时候(Co-partitioning，当 ksqlDB 在 join 串流资料时，必须要能够确认资料串流和表格是分区器是相同逻辑下去分配的）。

使用分区器 JavaCompatiblePartitioner 的方式一样是将它指派给生产者

const { Partitioners } = require('kafkajs')
kafka.producer({ createPartitioner: Partitioners.JavaCompatiblePartitioner })

明天会是生产者的最後一篇，接下来会进入消费者的篇章