{"id":1102,"date":"2023-05-28T02:28:45","date_gmt":"2023-05-27T18:28:45","guid":{"rendered":"https:\/\/xinchewhd.com.cn\/?p=1102"},"modified":"2023-05-28T02:45:19","modified_gmt":"2023-05-27T18:45:19","slug":"%e5%9c%a8java%e4%b8%ad%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af","status":"publish","type":"post","link":"https:\/\/xinchewhd.com.cn\/index.php\/java\/%e5%9c%a8java%e4%b8%ad%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/","title":{"rendered":"\u5728Java\u4e2d\u53d1\u9001kafka\u6d88\u606f"},"content":{"rendered":"
\n
<\/section>\n

\u6587\u7ae0\u76ee\u5f55<\/span><\/h3>\n
    \n
  • \n

    \u4e00\u3001kafka API\u7b80\u4ecb<\/p>\n<\/li>\n

  • \n

    \u4e8c\u3001\u5f15\u5165kafka java\u5ba2\u6237\u7aef\u4f9d\u8d56<\/p>\n<\/li>\n

  • \n

    \u4e09\u3001\u53d1\u9001\u6d88\u606f<\/p>\n<\/li>\n

  • \n

    \u56db\u3001\u6d88\u606f\u5206\u533a\u673a\u5236<\/p>\n

      \n
    • \n

      1\u3001\u81ea\u5b9a\u4e49\u5206\u533a\u673a\u5236<\/p>\n<\/li>\n<\/ul>\n<\/li>\n

    • \n

      \u4e94\u3001\u6d88\u606f\u5e8f\u5217\u5316<\/p>\n

        \n
      • \n

        1\u3001\u81ea\u5b9a\u4e49\u5e8f\u5217\u5316\u5668<\/p>\n<\/li>\n<\/ul>\n<\/li>\n

      • \n

        \u516d\u3001producer\u62e6\u622a\u5668<\/p>\n<\/li>\n

      • \n

        \u4e03\u3001\u6d88\u606f\u7684\u53ef\u9760\u53d1\u9001<\/p>\n

          \n
        • \n

          1\u3001\u65e0\u6d88\u606f\u4e22\u5931\u914d\u7f6e<\/p>\n<\/li>\n

        • \n

          2\u3001\u9644\u5f55\uff1a\u6d88\u8d39\u8005\u4ee3\u7801\u6f14\u793a<\/p>\n<\/li>\n

        • \n

          \n<\/li>\n<\/ul>\n<\/li>\n<\/ul>\n

          kafka\u5305\u542b5\u4e2a\u6838\u5fc3api\uff1a<\/p>\n

            \n
          1. \n

            Producer API\u7528\u6765\u8ba9\u5e94\u7528\u7a0b\u5e8f\u53d1\u9001\u6d41\u6570\u636e\u5230kafka\u96c6\u7fa4\u4e2d\u7684\u4e3b\u9898\u3002<\/p>\n<\/li>\n

          2. \n

            Consumer API\u7528\u6765\u8ba9\u5e94\u7528\u7a0b\u5e8f\u8bfb\u53d6kafka\u96c6\u7fa4\u4e3b\u9898\u4e2d\u7684\u6d41\u6570\u636e\u3002<\/p>\n<\/li>\n

          3. \n

            Streams API \u7528\u6765\u4ece\u8f93\u5165\u4e3b\u9898\u5230\u8f93\u51fa\u4e3b\u9898\u8f6c\u6362\u6d41\u6570\u636e\u3002<\/p>\n<\/li>\n

          4. \n

            Connect API \u7528\u6765\u5b9e\u73b0\u8fde\u63a5\u5668\uff0c\u6301\u7eed\u7684\u4ece\u6570\u636e\u6e90\u6216\u5e94\u7528\u7a0b\u5e8f\u63a8\u9001\u6570\u636e\u5230kafka\uff0c\u6216\u4ecekafka\u4e2d\u62c9\u53d6\u6570\u636e\u3002<\/p>\n<\/li>\n

          5. \n

            AdminClient API\u7528\u6765\u7ba1\u7406\u548c\u68c0\u67e5\u4e3b\u9898\u3001broker\u548c\u5176\u4ed6kafka\u5bf9\u8c61\u3002<\/p>\n<\/li>\n<\/ol>\n

            Kafka \u901a\u8fc7\u72ec\u7acb\u4e8e\u8bed\u8a00\u7684\u534f\u8bae\u516c\u5f00\u5176\u6240\u6709\u529f\u80fd\uff0c\u8be5\u534f\u8bae\u5177\u6709\u591a\u79cd\u7f16\u7a0b\u8bed\u8a00\u7684\u5ba2\u6237\u7aef\u3002\u7136\u800c\uff0c\u53ea\u6709 Java \u5ba2\u6237\u7aef\u4f5c\u4e3a\u4e3b\u8981 Kafka \u9879\u76ee\u7684\u4e00\u90e8\u5206\u8fdb\u884c\u7ef4\u62a4\uff0c\u5176\u4ed6\u5ba2\u6237\u7aef\u53ef\u4f5c\u4e3a\u72ec\u7acb\u7684\u5f00\u6e90\u9879\u76ee\u4f7f\u7528\u3002<\/p>\n

            \u4e8c\u3001\u5f15\u5165kafka java\u5ba2\u6237\u7aef\u4f9d\u8d56<\/span><\/h1>\n

            \u8981\u4f7f\u7528Producer\u3001Consumer\u548cAdminClient\uff0c\u4f60\u53ef\u4ee5\u4f7f\u7528\u4ee5\u4e0b maven \u4f9d\u8d56\u9879\uff1a<\/span><\/p>\n

            \n
                <dependency<\/span>>\n<\/span><\/span><\/code>      <groupId<\/span>><\/span>org.apache.kafka<\/groupId<\/span>>\n<\/span><\/span><\/code>      <artifactId<\/span>><\/span>kafka-clients<\/artifactId<\/span>>\n<\/span><\/span><\/code>      <version<\/span>><\/span>3.4.0<\/version<\/span>>\n<\/span><\/span><\/code>    <\/dependency<\/span>><\/span><\/span><\/code><\/pre>\n<\/section>\n
            \u4e09\u3001\u53d1\u9001\u6d88\u606f<\/span><\/p>\n
            kafka\u5c01\u88c5\u4e86\u4e00\u5957\u4e8c\u8fdb\u5236\u901a\u4fe1\u534f\u8bae<\/span>\uff0c\u7528\u4e8e\u5bf9\u5916<\/span>\u63d0\u4f9b\u5404\u79cd\u5404\u6837\u7684\u670d\u52a1\uff0c\u5bf9\u4e8eproducer\u800c\u8a00\uff0c\u7528\u6237\u53ef\u4ee5\u4f7f\u7528\u4efb\u610f\u7f16\u7a0b\u8bed\u8a00\u6309\u7167\u8be5\u534f\u8bae\u7684\u683c\u5f0f\u8fdb\u884c\u7f16\u7a0b\uff0c\u4ece\u800c\u5b9e\u73b0\u5411kafka\u53d1\u9001\u6d88\u606f\u3002<\/span>\u8fd9\u7ec4\u534f\u8bae\u672c\u8d28\u4e0a\u4e3a\u4e0d\u540c\u7684\u534f\u8bae\u7c7b\u578b\u5206\u522b\u5b9a\u4e49\u4e86\u4e13\u5c5e\u7684\u7d27\u51d1\u4e8c\u8fdb\u5236\u5b57\u8282\u6570\u7ec4\u683c\u5f0f\uff0c\u7136\u540e\u901a\u8fc7socket\u53d1\u9001\u7ed9\u5408\u9002\u7684broker\uff0c\u4e4b\u540e\u7b49\u5f85broker\u5904\u7406\u5b8c\u6210\u540e\u8fd4\u8fd8\u54cd\u5e94\u7ed9producer\u3002<\/span><\/p>\n
            \u6bcf\u4e2aproducer\u90fd\u662f\u72ec\u7acb\u8fdb\u884c\u5de5\u4f5c\u7684\uff0c\u4e0e\u5176\u4ed6producer\u4e4b\u95f4\u6ca1\u6709\u5173\u8054\uff0c\u76ee\u524dproducer\u7684\u9996\u8981\u529f\u80fd\u5c31\u662f\u5411\u67d0\u4e2atopic\u7684\u67d0\u4e2a\u5206\u533a\u53d1\u9001\u4e00\u6761\u6d88\u606f\uff0c\u6240\u4ee5\u5b83\u9996\u5148\u9700\u8981\u786e\u8ba4\u5230\u5e95\u5411topic\u7684\u54ea\u4e2a\u5206\u533a\u5199\u5165\u6d88\u606f\uff0c\u8fd9\u5c31\u662f\u5206\u533a\u5668\uff08partitioner\uff09\u7684\u4e8b\u60c5\u3002kafka producer\u63d0\u4f9b\u4e86\u4e00\u4e2a\u9ed8\u8ba4\u7684\u5206\u533a\u5668\uff0c\u5bf9\u4e8e\u6bcf\u6761\u5f85\u53d1\u9001\u7684\u6d88\u606f\uff0c\u5982\u679c\u8be5\u6d88\u606f\u6307\u5b9a\u4e86key\uff0c\u90a3\u4e48\u8be5partitioner\u4f1a\u6839\u636ekey\u7684\u54c8\u5e0c\u503c\u6765\u9009\u62e9\u76ee\u6807\u5206\u533a\uff0c\u82e5\u6ca1\u6709\uff0c\u4f1a\u4f7f\u7528\u8f6e\u8be2\u7684\u65b9\u5f0f\u786e\u8ba4\u76ee\u6807\u5206\u533a\u3002\u5f53\u7136\uff0cproducer\u7684API\u8d4b\u4e88\u4e86\u7528\u6237\u81ea\u884c\u6307\u5b9a\u76ee\u6807\u5206\u533a\u7684\u6743\u529b\u3002<\/span><\/p>\n
            \u5728\u786e\u8ba4\u4e86\u76ee\u6807\u5206\u533a\u540e\uff0cproducer\u8981\u505a\u7684\u7b2c\u4e8c\u4ef6\u4e8b\u5c31\u662f\u8981\u5bfb\u627e\u8fd9\u4e2a\u5206\u533a\u5bf9\u5e94\u7684leader\uff0c\u53ea\u6709leader\u624d\u80fd\u54cd\u5e94\u5ba2\u6237\u7aef\u53d1\u9001\u8fc7\u6765\u7684\u8bf7\u6c42\uff0c\u800c\u5269\u4e0b\u7684\u4ece\u8282\u70b9\u4e2d\u6709\u4e00\u90e8\u5206\u4f1a\u540c\u6b65\u8be5\u6d88\u606f\u3002\u56e0\u6b64\u5728\u53d1\u9001\u6d88\u606f\u65f6\uff0cproducer\u6709\u4e0d\u7b49\u5f85\u4efb\u4f55\u526f\u672c\u7684\u54cd\u5e94\u4fbf\u8fd4\u56de\u6210\u529f\uff0c\u6216\u8005\u53ea\u7b49\u5f85leader\u54cd\u5e94\u5199\u5165\u64cd\u4f5c\u4e4b\u540e\u518d\u8fd4\u56de\u6210\u529f\u3002<\/span><\/p>\n
            Java\u7248\u672cproducer\u5de5\u4f5c\u6d41\u7a0b\uff1a<\/span><\/p>\n
            \u751f\u4ea7\u8005\u4ee3\u7801\u6f14\u793a\uff1a<\/span><\/p>\n
            \n
            import<\/span>\u00a0java.util.Properties;<\/span><\/code>\n<\/span><\/code>import<\/span> org.apache.kafka.clients.CommonClientConfigs;<\/span><\/code>import<\/span> org.apache.kafka.clients.producer.Callback;<\/span><\/code>import<\/span> org.apache.kafka.clients.producer.KafkaProducer;<\/span><\/code>import<\/span> org.apache.kafka.clients.producer.Producer;<\/span><\/code>import<\/span> org.apache.kafka.clients.producer.ProducerConfig;<\/span><\/code>import<\/span> org.apache.kafka.clients.producer.ProducerRecord;<\/span><\/code>import<\/span> org.apache.kafka.clients.producer.RecordMetadata;<\/span><\/code>import<\/span> org.apache.kafka.common.config.SaslConfigs;<\/span><\/code>import<\/span> org.apache.kafka.common.serialization.StringSerializer;<\/span><\/code>\n<\/span><\/code>\/**<\/span><\/span><\/code> * kafka\u751f\u4ea7<\/span><\/span><\/code> * <\/span><\/span><\/code> * @author dong.zhao<\/span><\/span><\/code> * @date 2023\/02\/09 14:06<\/span><\/span><\/code> **\/\n<\/span><\/span><\/code>public<\/span> class<\/span> ProducerTest {<\/span><\/code>\n<\/span><\/code>    public<\/span> static<\/span> void<\/span> main(String<\/span>[] args) {<\/span><\/code>\n        Properties props = new<\/span> Properties();\n<\/span><\/code>        \/\/ server\n<\/span><\/span><\/code>        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, \"127.0.0.1:9092\"<\/span>);\n<\/span><\/code>        \/\/ \u88ab\u53d1\u9001\u5230broker\u7684\u4efb\u4f55\u6d88\u606f\u7684\u683c\u5f0f\u90fd\u5fc5\u987b\u662f\u5b57\u8282\u6570\u7ec4\n<\/span><\/span><\/code>        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());\n<\/span><\/code>        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());\n<\/span><\/code>        \/\/ \u975e\u5fc5\u987b\u53c2\u6570\u914d\u7f6e\n<\/span><\/span><\/code>        \/\/ acks=0\u8868\u660eproducer\u5b8c\u5168\u4e0d\u7ba1\u53d1\u9001\u7ed3\u679c\uff1b\n<\/span><\/span><\/code>        \/\/ acks=all\u6216-1\u8868\u660eproducer\u4f1a\u7b49\u5f85ISR\u6240\u6709\u8282\u70b9\u5747\u5199\u5165\u540e\u7684\u54cd\u5e94\u7ed3\u679c\uff1b<\/span><\/span><\/code>\n        \/\/ acks=1\uff0c\u8868\u660eproducer\u4f1a\u7b49\u5f85leader\u5199\u5165\u540e\u7684\u54cd\u5e94\u7ed3\u679c<\/span><\/span><\/code>\n        props.put(ProducerConfig.ACKS_CONFIG, \"-1\"<\/span>);<\/span><\/code>\n        \/\/ \u53d1\u751f\u53ef\u91cd\u8bd5\u5f02\u5e38\u65f6\u7684\u91cd\u8bd5\u6b21\u6570<\/span><\/span><\/code>\n        props.put(ProducerConfig.RETRIES_CONFIG, 3<\/span>);<\/span><\/code>\n        \/\/ producer\u4f1a\u5c06\u53d1\u5f80\u540c\u4e00\u5206\u533a\u7684\u591a\u6761\u6d88\u606f\u5c01\u88c5\u8fdb\u4e00\u4e2abatch\u4e2d\uff0c<\/span><\/span><\/code>\n        \/\/ \u5f53batch\u6ee1\u4e86\u7684\u65f6\u5019\uff0c\u53d1\u9001\u5176\u4e2d\u7684\u6240\u6709\u6d88\u606f,\u4e0d\u8fc7\u5e76\u4e0d\u603b\u662f\u7b49\u5f85batch\u6ee1\u4e86\u624d\u53d1\u9001\u6d88\u606f\uff1b<\/span><\/span><\/code>\n        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840<\/span>);<\/span><\/code>\n        \/\/ \u63a7\u5236\u6d88\u606f\u53d1\u9001\u5ef6\u65f6\uff0c\u9ed8\u8ba4\u4e3a0\uff0c\u5373\u7acb\u5373\u53d1\u9001\uff0c\u65e0\u9700\u5173\u5fc3batch\u662f\u5426\u5df2\u88ab\u586b\u6ee1\u3002<\/span><\/span><\/code>\n        props.put(ProducerConfig.LINGER_MS_CONFIG, 1<\/span>);<\/span><\/code>\n        \/\/ \u6307\u5b9a\u4e86producer\u7528\u4e8e\u7f13\u5b58\u6d88\u606f\u7684\u7f13\u51b2\u533a\u5927\u5c0f\uff0c\u5355\u4f4d\u5b57\u8282\uff0c\u9ed8\u8ba432MB<\/span><\/span><\/code>\n        \/\/ producer\u542f\u52a8\u65f6\u4f1a\u9996\u5148\u521b\u5efa\u4e00\u5757\u5185\u5b58\u7f13\u51b2\u533a\u7528\u4e8e\u4fdd\u5b58\u5f85\u53d1\u9001\u7684\u6d88\u606f\uff0c\u7136\u540e\u7531\u53e6\u4e00\u4e2a\u4e13\u5c5e\u7ebf\u7a0b\u8d1f\u8d23\u4ece\u7f13\u51b2\u533a\u4e2d\u8bfb\u53d6\u6d88\u606f\u6267\u884c\u771f\u6b63\u7684\u53d1\u9001<\/span><\/span><\/code>\n        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432<\/span>);<\/span><\/code>\n        \/\/ \u8bbe\u7f6eproducer\u80fd\u53d1\u9001\u7684\u6700\u5927\u6d88\u606f\u5927\u5c0f<\/span><\/span><\/code>\n        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760<\/span>);<\/span><\/code>\n        \/\/ \u8bbe\u7f6e\u662f\u5426\u538b\u7f29\u6d88\u606f\uff0c\u9ed8\u8ba4none<\/span><\/span><\/code>\n        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, \"lz4\"<\/span>);<\/span><\/code>\n        \/\/ \u8bbe\u7f6e\u6d88\u606f\u53d1\u9001\u540e\uff0c\u7b49\u5f85\u54cd\u5e94\u7684\u6700\u5927\u65f6\u95f4,\u9ed8\u8ba430\u79d2<\/span><\/span><\/code>\n        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30<\/span>);<\/span><\/code>\n        \/\/ \u662f\u5426\u5f00\u542f\u8ba4\u8bc1<\/span><\/span><\/code> \n       boolean<\/span> auth = false<\/span>;<\/span><\/code>\n        if<\/span> (auth) {<\/span><\/code>\n            String<\/span> userName = \"admin\"<\/span>;<\/span><\/code>\n            String<\/span> password = \"123456\"<\/span>;<\/span><\/code>\n            \/\/ \u8bbe\u7f6eSASL\u8fde\u63a5<\/span><\/span><\/code>\n            props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, \"SASL_PLAINTEXT\"<\/span>);<\/span><\/code>\n            props.put(SaslConfigs.SASL_MECHANISM, \"PLAIN\"<\/span>);<\/span><\/code>\n            props.put(SaslConfigs.SASL_JAAS_CONFIG,<\/span><\/code>\n                \"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\"<\/span> + userName<\/span><\/code>\n                    + \"\" password=\"\"<\/span> + password + \"\";\"<\/span>);<\/span><\/code>\n        }<\/span><\/code>\n<\/span><\/code>\n        Producer<String<\/span>, String<\/span>> producer = new<\/span> KafkaProducer<>(props);<\/span><\/code>\n        String<\/span> data =<\/span><\/code>\n            \"{\"account\":\"\u8d26\u53f7\",\"taskId\":\"\u4efb\u52a1ID\",\"data\":[{\"service\":\"\u670d\u52a1\",\"type\":\"\u7c7b\u578b\",\"cycle\":\"\u8ba1\u8d39\u5468\u671f\",\"cost\":[{\"cpu\":\"4\",\"gpu\":\"1\",\"memory\":\"32\"},{\"cpu\":\"4\",\"gpu\":\"1\",\"memory\":\"\"}]}]}\"<\/span>;<\/span><\/code>\n        \/\/ \u76f4\u63a5\u53d1\u9001\uff0c\u5373\u53d1\u9001\u4e4b\u540e\u4fbf\u4e0d\u518d\u7406\u4f1a\u53d1\u9001\u7ed3\u679c\uff0c\u8fd9\u79cd\u65b9\u5f0f\u5728\u5b9e\u9645\u4e2d\u662f\u4e0d\u88ab\u63a8\u8350\u4f7f\u7528\u7684\u3002<\/span><\/span><\/code>\n        \/\/ producer.send(new ProducerRecord<>(\"test\", \"key\", data));<\/span><\/span><\/code>\n        \/\/ \u5f02\u6b65\u53d1\u9001<\/span><\/span><\/code>\n        producer.send(new<\/span> ProducerRecord<>(\"test-topic\"<\/span>, uuid, data), new<\/span> Callback() {<\/span><\/code>\n            @Override<\/span><\/span><\/code>\n            public<\/span> void<\/span> onCompletion(RecordMetadata metadata, Exception exception) {<\/span><\/code>\n                if<\/span> (exception == null<\/span>) {<\/span><\/code>\n                    System.out.println(\"\u6d88\u606f\u53d1\u9001\u6210\u529f\"<\/span>);<\/span><\/code>\n                } else<\/span> {<\/span><\/code>\n                    System.out.println(\"\u6d88\u606f\u53d1\u9001\u5931\u8d25\"<\/span> + exception);<\/span><\/code>\n                }<\/span><\/code>\n            }<\/span><\/code>\n        });<\/span><\/code>\n        \/\/ \u540c\u6b65\u53d1\u9001<\/span><\/span><\/code>\n        \/\/ ProducerRecord<String, String> record = new ProducerRecord<>(\"my-replicated-topic\", \"key\", data);<\/span><\/span><\/code>\n        \/\/ RecordMetadata recordMetadata = producer.send(record).get();<\/span><\/span><\/code>\n<\/span><\/code>        producer.close();<\/span><\/code>\n<\/span><\/code>    }<\/span><\/code>}<\/span><\/code><\/pre>\n<\/section>\n
            kaf<\/span>ka producer\u53d1\u9001\u6d88\u606f\u7684\u4e3b\u65b9\u6cd5\u662fsend\u65b9\u6cd5\uff0c\u5728\u5e95\u5c42\u5b8c\u5168\u5730\u5b9e\u73b0\u4e86\u5f02\u6b65\u5316\u53d1\u9001\uff0c\u5e76\u4e14\u901a\u8fc7Java\u63d0\u4f9b\u7684Future\u540c\u65f6\u5b9e\u73b0\u4e86\u540c\u6b65\u53d1\u9001\u548c\u5f02\u6b65\u53d1\u9001+\u56de\u8c03\u4e24\u79cd\u53d1\u9001\u65b9\u5f0f\u3002<\/span>\u800c\u4e0a\u8ff0\u4ee3\u7801\u4f7f\u7528\u7684\u662f\u7b2c\u4e09\u79cd\u65b9\u5f0f\uff0c\u5373\u53d1\u9001\u4e4b\u540e\u4fbf\u4e0d\u518d\u7406\u4f1a\u53d1\u9001\u7ed3\u679c\uff0c\u8fd9\u79cd\u65b9\u5f0f\u5728\u5b9e\u9645\u4e2d\u662f\u4e0d\u88ab\u63a8\u8350\u4f7f\u7528\u7684\u3002<\/span><\/p>\n
            \u5f02\u6b65\u53d1\u9001\uff1a<\/span><\/span><\/p>\n
            \u5b9e\u9645\u4e0a\u6240\u6709\u7684\u5199\u5165\u64cd\u4f5c\u9ed8\u8ba4\u90fd\u662f\u5f02\u6b65\u7684\uff0csend\u65b9\u6cd5\u4f1a\u8fd4\u56de\u4e00\u4e2aJava Future\u5bf9\u8c61\u4f9b\u7528\u6237\u7a0d\u540e\u83b7\u53d6\u53d1\u9001\u7ed3\u679c\uff0c\u8fd9\u5c31\u662f\u6240\u8c13\u7684\u56de\u8c03\u673a\u5236\u3002\u5177\u4f53\u4ee3\u7801\u5982\u4e0b\uff1a<\/span><\/p>\n
            \n
            producer.send(record,new<\/span> Callback(){<\/span><\/code>\n  @Override<\/span><\/span><\/code>\n  public<\/span> void<\/span> onCompletion<\/span>(RecordMetadata metadata,Exception exception<\/span>)<\/span>{<\/span><\/code>\n    if<\/span>(exception == null<\/span>){<\/span><\/code>\n      System.out<\/span>.println(\"\u6d88\u606f\u53d1\u9001\u6210\u529f\"<\/span>);<\/span><\/code>\n    } else<\/span> {<\/span><\/code>\n      System.out<\/span>.println(\"\u6d88\u606f\u53d1\u9001\u5931\u8d25\"<\/span>);<\/span><\/code>\n    }    <\/span><\/code>\n  }<\/span><\/code>\n});\u00a0\u00a0<\/span><\/code><\/pre>\n<\/section>\n
            \u4e0a\u9762\u7684\u4ee3\u7801\u4e2d\uff0cCallback\u5c31\u662f\u53d1\u9001\u6d88\u606f\u540e\u7684\u56de\u8c03\u7c7b\uff0c\u5176onCompletion\u65b9\u6cd5\u7684\u4e24\u4e2a\u8f93\u5165\u53c2\u6570metadata\u548cexception\u4e0d\u4f1a\u540c\u65f6\u975e\u7a7a\uff0c\u5f53\u6d88\u606f\u53d1\u9001\u6210\u529f\u65f6\uff0cexception\u4e3anull\uff0c\u5f53\u6d88\u606f\u53d1\u9001\u5931\u8d25\u65f6\uff0cmetadata\u5c31\u662fnull\u3002Callback\u5b9e\u9645\u4e0a\u662f\u4e00\u4e2aJava\u63a5\u53e3\uff0c\u56e0\u6b64\u53ef\u521b\u5efa\u81ea\u5b9a\u4e49\u7684Callback\u5b9e\u73b0\u7c7b\u6765\u5904\u7406\u6d88\u606f\u53d1\u9001\u540e\u7684\u903b\u8f91\u3002<\/span><\/p>\n
            \u540c\u6b65\u53d1\u9001\uff1a<\/span><\/span><\/p>\n
            \u540c\u6b65\u53d1\u9001\u548c\u5f02\u6b65\u53d1\u9001\u5176\u5b9e\u5c31\u662f\u901a\u8fc7Java\u7684Future\u6765\u533a\u5206\u7684\uff0c\u8c03\u7528Future.get()\u7b49\u5f85\u8fd4\u56de\u7ed3\u679c\uff0c\u5373\u662f\u540c\u6b65\u53d1\u9001\uff0c\u5177\u4f53\u4ee3\u7801\u5982\u4e0b\uff1a<\/span><\/p>\n
            \n
            ProducerRecord<String<\/span>, String<\/span>> record = new<\/span> ProducerRecord<>(\"test-topic\"<\/span>,uuid,data);<\/span><\/code>\nRecordMetadata\u00a0recordMetadata\u00a0=\u00a0producer.send(record).get();<\/span><\/code><\/pre>\n<\/section>\n
            \u4f7f\u7528Future.get()\u65b9\u6cd5\u4f1a\u4e00\u76f4\u7b49\u5f85\u4e0b\u53bb\u76f4\u5230broker\u5c06\u7ed3\u679c\u8fd4\u56de\u7ed9producer\u7a0b\u5e8f\uff0c\u5f53\u7ed3\u679c\u8fd4\u56de\u65f6\uff0cget()\u65b9\u6cd5\u8981\u4e48\u8fd4\u56de\u53d1\u9001\u7ed3\u679c\uff0c\u8981\u4e48\u629b\u51fa\u5f02\u5e38\u4ea4\u7531producer\u81ea\u884c\u5904\u7406\u3002\u5982\u679c\u6ca1\u6709\u9519\u8bef\uff0cget\u5c06\u8fd4\u56de\u5bf9\u5e94\u7684RecordMetadata\u5b9e\u4f8b\u3002<\/span><\/span><\/span><\/p>\n
            kafka\u53d1\u9001\u5f02\u5e38<\/span><\/span><\/span><\/span><\/p>\n
            \u5f53\u524dkafka\u7684\u9519\u8bef\u7c7b\u578b\u5305\u542b\u4e86\u4e24\u7c7b\uff1a\u53ef\u91cd\u8bd5\u5f02\u5e38\u548c\u4e0d\u53ef\u91cd\u8bd5\u5f02\u5e38\uff0c\u5e38\u89c1\u7684\u53ef\u91cd\u8bd5\u5f02\u5e38\u5982\u4e0b\uff1a<\/span><\/span><\/p>\n
              \n
            • \n
              LeaderNotAvailableException\uff1a\u5206\u533a\u5bf9\u5e94\u7684leader\u4e0d\u53ef\u7528\uff0c\u901a\u5e38\u51fa\u73b0\u5728leader\u6362\u5c4a\u9009\u4e3e\u65f6\uff0c\u56e0\u6b64\u662f\u77ac\u65f6\u7684\u5f02\u5e38\uff0c\u91cd\u8bd5\u4e4b\u540e\u53ef\u81ea\u884c\u6062\u590d\u3002<\/p>\n<\/li>\n
            • \n
              NotControllerException\uff1a\u8868\u660econtroller\u5f53\u524d\u4e0d\u53ef\u7528\uff0c\u5728\u7ecf\u5386\u65b0\u4e00\u8f6e\u7684\u9009\u4e3e\uff0c\u91cd\u8bd5\u4e4b\u540e\u53ef\u81ea\u884c\u6062\u590d\u3002<\/p>\n<\/li>\n
            • \n
              NetworkException\uff1a\u7f51\u7edc\u77ac\u65f6\u5f02\u5e38\uff0c\u53ef\u91cd\u8bd5\u3002<\/p>\n<\/li>\n<\/ul>\n
              \u6240\u6709\u53ef\u91cd\u8bd5\u7684\u5f02\u5e38\u90fd\u7ee7\u627f\u81eaor.apache.kafka.common.errors.RetriableException\uff0c\u5bf9\u4e8e\u8fd9\u4e9b\u53ef\u91cd\u8bd5\u7684\u5f02\u5e38\uff0c\u5982\u679c\u5728producer\u7a0b\u5e8f\u4e2d\u914d\u7f6e\u4e86\u91cd\u8bd5\u6b21\u6570\uff0c\u90a3\u4e48\u53ea\u8981\u5728\u89c4\u5b9a\u7684\u91cd\u8bd5\u6b21\u6570\u5185\u81ea\u884c\u6062\u590d\u4e86\uff0c\u4fbf\u4e0d\u4f1a\u51fa\u73b0\u5728onCompletion\u7684exception\u4e2d\u3002\u82e5\u8d85\u8fc7\u4e86\u91cd\u8bd5\u6b21\u6570\u4ecd\u6ca1\u6210\u529f\uff0c\u5c31\u4f1a\u88ab\u5c01\u88c5\u5230exception\u4e2d\uff0c\u6b64\u65f6\u5c31\u9700\u8981producer\u7a0b\u5e8f\u81ea\u884c\u5904\u7406\u8fd9\u79cd\u5f02\u5e38\u3002<\/span><\/p>\n
              \u6ca1\u6709\u7ee7\u627f\u81eaRetriableException\u7684\u5176\u4ed6\u5f02\u5e38\u90fd\u5c5e\u4e8e\u4e0d\u53ef\u91cd\u8bd5\u5f02\u5e38\uff0c\u8fd9\u7c7b\u5f02\u5e38\u8868\u660e\u4e86\u4e00\u4e9b\u975e\u5e38\u4e25\u91cd\u6216kafka\u65e0\u6cd5\u5904\u7406\u7684\u95ee\u9898\u3002<\/span><\/span><\/p>\n
              \u56db\u3001\u6d88\u606f\u5206\u533a\u673a\u5236<\/span><\/h1>\n
              producer\u53d1\u9001\u8fc7\u7a0b\u4e2d\u9700\u8981\u786e\u5b9a\u5c06\u6d88\u606f\u53d1\u9001\u5230topic\u7684\u54ea\u4e00\u4e2a\u5206\u533a\uff0c\u9ed8\u8ba4\u7684\u5206\u533a\u5668\u4f1a\u5c3d\u529b\u786e\u4fdd\u5177\u6709\u76f8\u540ckey\u7684\u6240\u6709\u6d88\u606f\u90fd\u88ab\u53d1\u9001\u5230\u76f8\u540c\u7684\u5206\u533a\u4e0a\uff1b\u82e5\u6ca1\u6709\u6307\u5b9akey\uff0c\u4f1a\u4ee5\u8f6e\u8be2\u7684\u65b9\u5f0f\u6765\u786e\u4fdd\u6d88\u606f\u5728topic\u7684\u5206\u533a\u4e0a\u5747\u5300\u5206\u914d\u3002<\/span><\/p>\n
              1\u3001\u81ea\u5b9a\u4e49\u5206\u533a\u673a\u5236<\/span><\/h2>\n
              \u81ea\u5b9a\u4e49\u5206\u533a\u5668\u9700\u8981\u5b9e\u73b0org.apache.kafka.clients.producer.Partitioner\u63a5\u53e3\uff0c\u5206\u533a\u903b\u8f91\u5199\u5728partition()\u65b9\u6cd5\u4e2d\uff0c\u4f8b\u5982\uff1a<\/span><\/span><\/p>\n
              \n
              public<\/span> class<\/span> AuditPartitioner<\/span> implements<\/span> Partitioner<\/span> <\/span>{<\/span><\/code>\n<\/span><\/code>    private<\/span> Random random;<\/span><\/code>\n<\/span><\/code>    @Override<\/span><\/span><\/code>\n    public<\/span> void<\/span> configure<\/span>(Map<String, ?> map)<\/span> <\/span>{<\/span><\/code>\n        \/\/ \u8be5\u65b9\u6cd5\u5b9e\u73b0\u5fc5\u8981\u8d44\u6e90\u7684\u521d\u59cb\u5316\u5de5\u4f5c<\/span><\/span><\/code>\n        random = new<\/span> Random();<\/span><\/code>\n    }<\/span><\/code>\n<\/span><\/code>    @Override<\/span><\/span><\/code>\n    public<\/span> int<\/span> partition<\/span>(String topic, Object keyObj, byte<\/span>[] keyBytes, Object value, byte<\/span>[] valueBytes, Cluster cluster)<\/span> <\/span>{<\/span><\/code>\n        String key = (String)keyObj;<\/span><\/code>\n        \/\/ \u83b7\u53d6\u8be5topic\u53ef\u7528\u7684\u6240\u6709\u5206\u533a<\/span><\/span><\/code>\n        List<PartitionInfo> partitionInfoList = cluster.availablePartitionsForTopic(topic);<\/span><\/code>\n        int<\/span> partitionCount = partitionInfoList.size();<\/span><\/code>\n        int<\/span> auditPartition = partitionCount -1<\/span>;<\/span><\/code>\n<\/span><\/code>        return<\/span> key == null<\/span> || !key.contains(\"audit\"<\/span>) ? random.nextInt(auditPartition) : auditPartition;<\/span><\/code>\n    }<\/span><\/code>\n<\/span><\/code>    @Override<\/span><\/span><\/code>\n    public<\/span> void<\/span> close<\/span>()<\/span> <\/span>{<\/span><\/code>\n         \/\/ \u8be5\u65b9\u6cd5\u5b9e\u73b0\u5fc5\u8981\u8d44\u6e90\u7684\u6e05\u7406<\/span><\/span><\/code>\n\u00a0\u00a0\u00a0\u00a0}<\/span><\/code>}<\/span><\/code><\/pre>\n<\/section>\n
              \u4f7f\u7528\u81ea\u5b9a\u4e49\u5206\u533a\u5668\uff1a<\/span><\/p>\n
              \n
              props<\/span>.put<\/span>(\"partitioner<\/span>.class<\/span>\",\"xx<\/span>.xx<\/span>.AuditPartitioner<\/span>\");<\/span><\/code><\/pre>\n<\/section>\n
              \u4e94\u3001\u6d88\u606f\u5e8f\u5217\u5316<\/span><\/h1>\n
              \u5728\u7f51\u7edc\u4e2d\u53d1\u9001\u6570\u636e\u90fd\u662f\u4ee5\u5b57\u8282\u7684\u65b9\u5f0f\uff0c\u5e8f\u5217\u5316\u5668\u8d1f\u8d23\u5728producer\u53d1\u9001\u6d88\u606f\u524d\u5c06\u5176\u8f6c\u6362\u6210\u5b57\u8282\u6570\u7ec4\uff0c\u4e0e\u4e4b\u76f8\u53cd\uff0c\u89e3\u5e8f\u5217\u5316\u5668\u8d1f\u8d23\u5c06consumer\u63a5\u6536\u5230\u7684\u5b57\u8282\u6570\u7ec4\u8f6c\u6362\u6210\u76f8\u5e94\u7684\u5bf9\u8c61\u6570\u636e\u3002<\/p>\n
              1\u3001\u81ea\u5b9a\u4e49\u5e8f\u5217\u5316\u5668<\/span><\/h2>\n
              \u81ea\u5b9a\u4e49\u5e8f\u5217\u5316\u5668\u9700\u8981\u5b9e\u73b0org.apache.kafka.common.serialization.Serializer\u63a5\u53e3\uff0c\u5728serializer\u65b9\u6cd5\u4e2d\u5b9e\u73b0\u5e8f\u5217\u5316\u903b\u8f91\uff0c\u4f8b\u5982\uff1a<\/p>\n
              \u9996\u5148\u5b9a\u4e49\u4e00\u4e2aPOJO\u5bf9\u8c61<\/p>\n
              \n
              public<\/span> class<\/span> User{<\/span><\/code>\n  private<\/span> String<\/span> firstName;<\/span><\/code>\n  private<\/span> String<\/span> lastName;<\/span><\/code>\n  private<\/span> int age;<\/span><\/code>\n  private<\/span> String<\/span> address;<\/span><\/code>\n<\/span><\/code>  public<\/span> User(String<\/span> firstName,String<\/span> lastName,int age,String<\/span> address){<\/span><\/code>\n  this<\/span>.firstName=firstName;<\/span><\/code>\n  this<\/span>.lastName=lastName;<\/span><\/code>\n  this<\/span>.age=age;<\/span><\/code>\n  this<\/span>.address=address;<\/span><\/code>\n\u00a0\u00a0}<\/span><\/code>}\u00a0\u00a0<\/span><\/code><\/pre>\n<\/section>\n
              \u7531\u4e8e\u8981\u7528jackson-mapper-asl\u5305\u4e2d\u7684ObjectMapper\u6765\u5c06\u5bf9\u8c61\u8f6c\u6210\u5b57\u8282\u6570\u7ec4\uff0c\u56e0\u6b64\u9700\u8981\u5c06\u5176\u4f9d\u8d56\u5f15\u5165\uff1a<\/span><\/span><\/p>\n
              \n
              <dependency<\/span>><\/span><\/span><\/code>\n  <groupId<\/span>><\/span>org.codehaus.jackson<\/groupId<\/span>><\/span><\/span><\/code>\n  <artifactId<\/span>><\/span>jackson-mapper-asl<\/artifactId<\/span>><\/span><\/span><\/code>\n  <version<\/span>><\/span>1.9.13<\/version<\/span>><\/span><\/span><\/code>\n<\/dependency<\/span>><\/span>\u00a0\u00a0<\/span><\/code><\/pre>\n<\/section>\n
              \u63a5\u4e0b\u6765\u521b\u5efaserializer<\/span><\/p>\n
              \n
              public<\/span> class<\/span> UserSerializer<\/span> implements<\/span> Serializer<\/span><\/span>{<\/span><\/code>\n  private<\/span> ObjectMapper objectMapper;<\/span><\/code>\n  <\/code>  @Override<\/span><\/span><\/code>\n  public<\/span> void<\/span> configure<\/span>(Map config,boolean<\/span> isKey)<\/span><\/span>{<\/span><\/code>\n      objectMapper = new<\/span> ObjectMapper();<\/span><\/code>\n  }<\/span><\/code>\n<\/span><\/code>  @Override<\/span><\/span><\/code>\n  public<\/span> byte<\/span>[] serialize(String topic,Object data){<\/span><\/code>\n    byte<\/span>[] res = null<\/span>;<\/span><\/code>\n    try<\/span>{<\/span><\/code>\n      res = objectMapper.writeValueAsString(data).getBytes(\"utf-8\"<\/span>);<\/span><\/code>\n    }catch<\/span>(Exception e){<\/span><\/code>\n      logger.warn(\"failed to serialize the object: {}\"<\/span>, data, e);<\/span><\/code>\n    }<\/span><\/code>\n    return<\/span> res;<\/span><\/code>\n  }<\/span><\/code>\n<\/span><\/code>   @Override<\/span><\/span><\/code>\n   public<\/span> void<\/span> close<\/span>()<\/span><\/span>{}<\/span><\/code>\n}\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0<\/span><\/code><\/pre>\n<\/section>\n
              \u4f7f\u7528\u81ea\u5b9a\u4e49\u5e8f\u5217\u5316\u5668\uff1a<\/span><\/span><\/p>\n
              \n
              props<\/span>.put<\/span>(\"value<\/span>.serializer<\/span>\",\"xx<\/span>.xx<\/span>.UserSerializer<\/span>\");<\/span><\/code><\/pre>\n<\/section>\n
              \u516d\u3001producer\u62e6\u622a\u5668<\/span><\/h1>\n
              producer\u62e6\u622a\u5668\u4f7f\u5f97\u7528\u6237\u5728\u6d88\u606f\u53d1\u9001\u524d\u548cproducer\u56de\u8c03\u903b\u8f91\u6267\u884c\u524d\u53ef\u5bf9\u6d88\u606f\u505a\u4e00\u4e9b\u5b9a\u5236\u5316\u5904\u7406\uff0c\u5141\u8bb8\u4f7f\u7528\u591a\u4e2a\u62e6\u622a\u5668\u6784\u6210\u62e6\u622a\u5668\u94fe\u3002\u62e6\u622a\u5668\u7684\u5b9e\u73b0\u63a5\u53e3\u662forg.apache.kafka.clients.producer.ProducerInterceptor\u3002interceptor\u53ef\u80fd\u8fd0\u884c\u5728\u591a\u4e2a\u7ebf\u7a0b\u4e2d\uff0c\u56e0\u6b64\u5728\u5177\u4f53\u5b9e\u73b0\u65f6\u7528\u6237\u9700\u8981\u786e\u4fdd\u7ebf\u7a0b\u5b89\u5168\u3002\u4e0b\u9762\u4ee5\u4e00\u4e2a\u7b80\u5355\u7684\u53ccinterceptor\u7ec4\u6210\u7684\u62e6\u622a\u94fe\u4e3a\u4f8b\u3002\u7b2c\u4e00\u4e2ainterceptor\u4f1a\u5728\u6d88\u606f\u53d1\u9001\u524d\u5c06\u65f6\u95f4\u6233\u4fe1\u606f\u52a0\u5230\u6d88\u606fvalue\u7684\u6700\u524d\u90e8\uff0c\u7b2c\u4e8c\u4e2ainterceptor\u4f1a\u5728\u6d88\u606f\u53d1\u9001\u540e\u66f4\u65b0\u6210\u529f\u53d1\u9001\u6d88\u606f\u6216\u5931\u8d25\u53d1\u9001\u6d88\u606f\u6570\u3002<\/span><\/p>\n
              \n
              public<\/span> class<\/span> TimeStampPrependerInterceptor<\/span> implements<\/span> ProducerInterceptor<\/span><String<\/span>,String<\/span>> <\/span>{<\/span><\/code>\n    \/**<\/span><\/span><\/code>\n     * producer\u786e\u4fdd\u5728\u6d88\u606f\u88ab\u5e8f\u5217\u5316\u524d\u8c03\u7528\u8be5\u65b9\u6cd5<\/span><\/span><\/code>\n     * \u53ef\u4ee5\u5728\u8be5\u65b9\u6cd5\u4e2d\u5bf9\u6d88\u606f\u505a\u4efb\u4f55\u64cd\u4f5c<\/span><\/span><\/code>\n     *\/<\/span><\/span><\/code>\n    @Override<\/span><\/span><\/code>\n    public<\/span> ProducerRecord onSend<\/span>(ProducerRecord record)<\/span> <\/span>{<\/span><\/code>\n        return<\/span> new<\/span> ProducerRecord(record.topic(),record.partition(),record.timestamp(),record.key(),System.currentTimeMillis() +\",\"<\/span>+record.value().toString());<\/span><\/code>\n    }<\/span><\/code>\n<\/span><\/code>    \/**<\/span><\/span><\/code>\n     * \u8be5\u65b9\u6cd5\u4f1a\u5728\u6d88\u606f\u88ab\u5e94\u7b54\u4e4b\u524d\u6216\u6d88\u606f\u53d1\u9001\u65f6\u8c03\u7528<\/span><\/span><\/code>\n     *\/<\/span><\/span><\/code>\n    @Override<\/span><\/span><\/code>\n    public<\/span> void<\/span> onAcknowledgement<\/span>(RecordMetadata recordMetadata, Exception e)<\/span> <\/span>{<\/span><\/code>\n<\/span><\/code>    }<\/span><\/code>\n<\/span><\/code>    @Override<\/span><\/span><\/code>\n    public<\/span> void<\/span> close<\/span>()<\/span> <\/span>{<\/span><\/code>\n<\/span><\/code>    }<\/span><\/code>\n<\/span><\/code>    @Override<\/span><\/span><\/code>\n    public<\/span> void<\/span> configure<\/span>(Map<String, ?> map)<\/span> <\/span>{<\/span><\/code>\n<\/span><\/code>    }<\/span><\/code>}<\/span><\/code><\/pre>\n<\/section>\n
              \n
              public<\/span> class<\/span> CounterInterceptor<\/span> implements<\/span> ProducerInterceptor<\/span><String<\/span>,String<\/span>> <\/span>{<\/span><\/code>\n<\/span><\/code>    private<\/span> int<\/span> errorCounter = 0<\/span>;<\/span><\/code>\n    private<\/span> int<\/span> successCounter = 0<\/span>;<\/span><\/code>\n<\/span><\/code>    @Override<\/span><\/span><\/code>\n    public<\/span> ProducerRecord<String, String> onSend<\/span>(ProducerRecord<String, String> record)<\/span> <\/span>{<\/span><\/code>\n        return<\/span> record;<\/span><\/code>\n    }<\/span><\/code>\n<\/span><\/code>    @Override\n<\/span><\/span><\/code>    public<\/span> void<\/span> onAcknowledgement<\/span>(RecordMetadata recordMetadata, Exception e)<\/span> <\/span>{<\/span><\/code>\n        if<\/span> (e == null<\/span>) {<\/span><\/code>\n            successCounter++;<\/span><\/code>\n        }else<\/span>{<\/span><\/code>\n            errorCounter++;<\/span><\/code>\n        }<\/span><\/code>\n    }<\/span><\/code>\n<\/span><\/code>    @Override<\/span><\/span><\/code>\n    public<\/span> void<\/span> close<\/span>()<\/span> <\/span>{<\/span><\/code>\n        System.out.println(\"Successful sent: \"<\/span> + successCounter);<\/span><\/code>\n        System.out.println(\"Failed sent: \"<\/span> + errorCounter);<\/span><\/code>\n    }<\/span><\/code>\n<\/span><\/code>    @Override<\/span><\/span><\/code>\n    public<\/span> void<\/span> configure<\/span>(Map<String, ?> map)<\/span> <\/span>{<\/span><\/code>\n<\/span><\/code>    }<\/span><\/code>}<\/span><\/code><\/pre>\n<\/section>\n
              \u4f7f\u7528\u81ea\u5b9a\u4e49interceptor\uff1a<\/p>\n
              \n
              List<String> interceptors = new<\/span> ArrayList<>();<\/span><\/code>\ninterceptors.add<\/span>(\"xx.xx.TimeStampPrependerInterceptor\"<\/span>);<\/span><\/code>\ninterceptors.add<\/span>(\"xx.xx.CounterInterceptor\"<\/span>);<\/span><\/code>\nprops.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptors);<\/span><\/code><\/pre>\n<\/section>\n
              \u4e03\u3001\u6d88\u606f\u7684\u53ef\u9760\u53d1\u9001<\/span><\/h1>\n
              Java\u7248\u672c\u7684producer\u91c7\u7528\u5f02\u6b65\u53d1\u9001\u673a\u5236\uff0csend\u65b9\u6cd5\u5c06\u6d88\u606f\u653e\u5165\u7f13\u51b2\u533a\uff0c\u7531\u4e00\u4e2a\u4e13\u5c5eI\/O\u7ebf\u7a0b\u8d1f\u8d23\u4ece\u7f13\u51b2\u533a\u4e2d\u63d0\u53d6\u6d88\u606f\u5e76\u5c01\u88c5\u8fdb\u6d88\u606fbatch\u4e2d\uff0c\u7136\u540e\u53d1\u9001\u51fa\u53bb\u3002\u8fd9\u4e2a\u8fc7\u7a0b\u5b58\u5728\u7740\u6570\u636e\u4e22\u5931\u7684\u7a97\u53e3\uff0c\u5373\u82e5I\/O\u7ebf\u7a0b\u53d1\u9001\u4e4b\u524dproducer\u5d29\u6e83\uff0c\u5219\u5b58\u50a8\u7f13\u51b2\u533a\u4e2d\u7684\u6d88\u606f\u4f1a\u5168\u90e8\u4e22\u5931\u3002producer\u7684\u53e6\u4e00\u4e2a\u95ee\u9898\u5c31\u662f\u6d88\u606f\u7684\u4e71\u5e8f\uff0c\u5047\u8bbe\u73b0\u53d1\u9001record1\u548crecord2\u4e24\u6761\u6d88\u606f\uff0c\u7531\u4e8e\u67d0\u4e9b\u539f\u56e0\u5bfc\u81f4record1\u672a\u53d1\u9001\u6210\u529f\uff0c\u540c\u65f6kafka\u53c8\u914d\u7f6e\u4e86\u91cd\u8bd5\u673a\u5236\uff0c\u90a3\u4e48producer\u91cd\u8bd5record1\u6210\u529f\u540e\uff0crecord1\u5728\u65e5\u5fd7\u4e2d\u7684\u4f4d\u7f6e\u53ef\u80fd\u53cd\u800c\u4f4d\u4e8erecord2\u4e4b\u540e\u3002<\/p>\n
              1\u3001\u65e0\u6d88\u606f\u4e22\u5931\u914d\u7f6e<\/span><\/h2>\n
              \/\/ \u8be5\u914d\u7f6e\u63a7\u5236 KafkaProducer.send() \u548c KafkaProducer.partitionsFor() \u5c06\u963b\u585e\u591a\u957f\u65f6\u95f4\u3002\u6b64\u5916\u8fd9\u4e9b\u65b9\u6cd5\u88ab\u963b\u6b62\uff0c\u4e5f\u53ef\u80fd\u662f\u56e0\u4e3a\u7f13\u51b2\u533a\u5df2\u6ee1\u6216\u5143\u6570\u636e\u4e0d\u53ef\u7528\u3002\u5728\u7528\u6237\u63d0\u4f9b\u7684\u5e8f\u5217\u5316\u7a0b\u5e8f\u6216\u5206\u533a\u5668\u4e2d\u7684\u9501\u5b9a\u4e0d\u4f1a\u8ba1\u5165\u6b64\u8d85\u65f6\u3002\u9ed8\u8ba4\u4e3a60000ms\u3002<\/section>\n
              max.block.ms=60000<\/section>\n
              acks=all<\/section>\n
              retries=Integer.MAX_VALUE<\/section>\n
              \/\/ \u8be5\u53c2\u6570\u8bbe\u7f6e\u4e3a1\u4f7fproducer\u5728\u67d0\u4e2abroker\u53d1\u9001\u54cd\u5e94\u4e4b\u524d\u5c06\u65e0\u6cd5\u518d\u7ed9broker\u53d1\u9001\u8bf7\u6c42\uff0c\u53ef\u9632\u6b62topic\u540c\u5206\u533a\u4e0b\u7684\u6d88\u606f\u4e71\u5e8f\u95ee\u9898\uff0c<\/section>\n
              max.in.flight.requests.per.connection=1<\/section>\n
              \/\/ \u8bbe\u7f6e\u4e0d\u5141\u8bb8\u975eISR\u4e2d\u7684\u526f\u672c\u88ab\u9009\u4e3e\u4e3aleader\uff0c\u4ece\u800c\u907f\u514dbroker\u7aef\u56e0\u65e5\u5fd7\u95ee\u9898\u9020\u6210\u6d88\u606f\u7684\u4e22\u5931<\/section>\n
              unclean.leader.electionenable=false<\/section>\n
              replication.factor=3<\/section>\n
              \/\/ \u7528\u4e8e\u63a7\u5236\u67d0\u6761\u6d88\u606f\u81f3\u5c11\u88ab\u5199\u5165\u5230ISR\u4e2d\u7684\u591a\u5c11\u4e2a\u526f\u672c\u624d\u7b97\u6210\u529f<\/p>\n
              min.insync.replicas=2<\/section>\n
              enable.auto.commit=false<\/section>\n
              \u4f7f\u7528\u5e26\u56de\u8c03\u673a\u5236\u7684send\u53d1\u9001\u6d88\u606f<\/p>\n
              Callback\u903b\u8f91\u4e2d\u663e\u5f0f\u7acb\u5373\u5173\u95edproducer<\/p>\n
              2\u3001\u9644\u5f55\uff1a\u6d88\u8d39\u8005\u4ee3\u7801\u6f14\u793a\uff1a<\/span><\/p>\n
              \n
              import<\/span>\u00a0java.time.Duration;<\/span><\/code>import<\/span> java.util.Collections;<\/span><\/code>import<\/span> java.util.Properties;<\/span><\/code>\n<\/span><\/code>import<\/span> org.apache.kafka.clients.consumer.ConsumerConfig;<\/span><\/code>import<\/span> org.apache.kafka.clients.consumer.ConsumerRecord;<\/span><\/code>import<\/span> org.apache.kafka.clients.consumer.ConsumerRecords;<\/span><\/code>import<\/span> org.apache.kafka.clients.consumer.KafkaConsumer;<\/span><\/code>import<\/span> org.apache.kafka.common.serialization.StringDeserializer;<\/span><\/code>\n<\/span><\/code>\/**<\/span><\/span><\/code>\n * kafka\u6d88\u8d39\u8005<\/span><\/span><\/code>\n * <\/span><\/span><\/code>\n * @author dong.zhao<\/span><\/span><\/code>\n * @date 2023\/02\/10 17:53<\/span><\/span><\/code>\n **\/<\/span><\/span><\/code>\npublic<\/span> class<\/span> ConsumerTest<\/span> <\/span>{<\/span><\/code>\n    private<\/span> static<\/span> final<\/span> String<\/span> TOPIC_NAME<\/span> = \"sothisai\"<\/span>;<\/span><\/code>\n    private<\/span> static<\/span> final<\/span> String<\/span> CONSUMER_GROUP_NAME<\/span> = \"testGroup\"<\/span>;<\/span><\/code>\n<\/span><\/code>    public<\/span> static<\/span> void main(String<\/span>[] args) {<\/span><\/code>\n        Properties<\/span> properties = new Properties<\/span>();<\/span><\/code>\n        properties.put(ConsumerConfig<\/span>.BOOTSTRAP_SERVERS_CONFIG<\/span>, \"127.0.0.1:9092\"<\/span>);<\/span><\/code>\n        \/\/ \u6d88\u8d39\u5206\u7ec4\u540d<\/span><\/span><\/code>\n        properties.put(ConsumerConfig<\/span>.GROUP_ID_CONFIG<\/span>, CONSUMER_GROUP_NAME<\/span>);<\/span><\/code>\n        \/\/ earliest:offset\u504f\u79fb\u81f3\u6700\u65e9\u65f6\u5019\u5f00\u59cb\u6d88\u8d39\uff1blatest\uff1a\u504f\u79fb\u5230\u4ece\u6700\u65b0\u5f00\u59cb\u6d88\u8d39\uff08\u9ed8\u8ba4\uff09<\/span><\/span><\/code>\n        properties.put(ConsumerConfig<\/span>.AUTO_OFFSET_RESET_CONFIG<\/span>, \"earliest\"<\/span>);<\/span><\/code>\n        \/\/ \u8bbe\u7f6e\u5e8f\u5217\u5316<\/span><\/span><\/code>\n        properties.put(ConsumerConfig<\/span>.KEY_DESERIALIZER_CLASS_CONFIG<\/span>, StringDeserializer<\/span>.class<\/span>.getName());<\/span><\/code>\n        properties.put(ConsumerConfig<\/span>.VALUE_DESERIALIZER_CLASS_CONFIG<\/span>, StringDeserializer<\/span>.class<\/span>.getName());<\/span><\/code>\n<\/span><\/code>        \/\/ \u521b\u5efa\u4e00\u4e2a\u6d88\u8d39\u8005\u7684\u5ba2\u6237\u7aef<\/span><\/span><\/code>\n        KafkaConsumer<\/span><String<\/span>, String<\/span>> consumer = new KafkaConsumer<\/span><>(properties);<\/span><\/code>\n<\/span><\/code>        \/\/ \u6d88\u8d39\u8005\u8ba2\u9605\u4e3b\u9898\u5217\u8868<\/span><\/span><\/code>\n        \/\/ KafkaConsumer.subscribe()\uff1a\u4e3aconsumer\u81ea\u52a8\u5206\u914dpartition\uff0c\u6709\u5185\u90e8\u7b97\u6cd5\u4fdd\u8bc1topic-partition\u4ee5\u6700\u4f18\u7684\u65b9\u5f0f\u5747\u5300\u5206\u914d\u7ed9\u540cgroup\u4e0b\u7684\u4e0d\u540cconsumer\u3002<\/span><\/span><\/code>\n        \/\/ KafkaConsumer.assign()\uff1a\u4e3aconsumer\u624b\u52a8\u3001\u663e\u793a\u7684\u6307\u5b9a\u9700\u8981\u6d88\u8d39\u7684topic-partitions\uff0c\u4e0d\u53d7group.id\u9650\u5236\u3002<\/span><\/span><\/code>\n        consumer.subscribe(Collections<\/span>.singletonList(TOPIC_NAME<\/span>));<\/span><\/code>\n<\/span><\/code>        \/\/ \/\/ A.\u8bbe\u7f6e\u8981\u8ba2\u9605\u7684topic \u5217\u8868<\/span><\/span><\/code>\n        \/\/ List<TopicPartition> topicPartitions = new ArrayList<>();<\/span><\/span><\/code>\n        \/\/ topicPartitions.add(new TopicPartition(TOPIC_NAME, 0));<\/span><\/span><\/code>\n        \/\/ consumer.assign(topicPartitions);<\/span><\/span><\/code>\n<\/span><\/code>        while<\/span> (true<\/span>) {<\/span><\/code>\n            ConsumerRecords<\/span><String<\/span>, String<\/span>> records = consumer.poll(Duration<\/span>.ofMillis(2000<\/span>));<\/span><\/code>\n            for<\/span> (ConsumerRecord<\/span><String<\/span>, String<\/span>> record : records) {<\/span><\/code>\n                System<\/span>.out.printf(\"\u6536\u5230\u7684\u6d88\u606f\uff1apartition= %d,offset= %d,key= %s,value=%s %n\"<\/span>, record.partition<\/span>(),<\/span><\/code>\n                    record.offset(), record.key(), record.value());<\/span><\/code>\n            }<\/span><\/code>\n        }<\/span><\/code>\n    }<\/span><\/code>}<\/span><\/code><\/pre>\n<\/section>\n
              \u00b7END\u00b7<\/strong><\/p>\n
              \n
              \u672c\u7bc7\u6587\u7ae0\u6765\u6e90\u4e8e\u5fae\u4fe1\u516c\u4f17\u53f7: IT\u725b\u5ba2<\/p><\/blockquote>\n<\/div>\n","raw":"","protected":false},"excerpt":{"rendered":"
              \u6587\u7ae0\u76ee\u5f55 \u4e00\u3001kafka API\u7b80\u4ecb \u4e8c\u3001\u5f15\u5165kafka java\u5ba2\u6237\u7aef\u4f9d\u8d56 \u4e09\u3001\u53d1\u9001\u6d88\u606f \u56db\u3001\u6d88\u606f\u5206\u533a\u673a\u5236 1\u3001\u81ea\u5b9a\u4e49\u5206\u533a\u673a\u5236  …<\/p>\n","protected":false},"author":1,"featured_media":1097,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"emotion":"","emotion_color":"","title_style":"","license":"","footnotes":""},"categories":[2,16,7],"tags":[],"class_list":["post-1102","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-java","category-kafka","category-spring-boot"],"post_thumbnail_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022520471-150x150.jpeg","content_first_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022521778.jpeg","post_medium_image_300":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022520471-300x300.jpeg","post_thumbnail_image_624":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022520471.jpeg","post_frist_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022521778.jpeg","post_medium_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022520471-300x300.jpeg","post_large_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022520471.jpeg","post_full_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022520471.jpeg","post_all_images":[{"imagesurl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022521778.jpeg","id":"image0"},{"imagesurl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022521778.jpeg","id":"image1"},{"imagesurl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022523822.png","id":"image2"},{"imagesurl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230528022523822.png","id":"image3"}],"videoAdId":"","listAd":"0","listAdId":"","listAdEvery":6,"total_comments":0,"category_name":"Java","post_date":"2023-05-27","like_count":"0","praiseWord":"\u9f13\u52b1","copyright_state":"","excitationAd":"0","rewardedVideoAdId":"","detailAdId":"","detailAd":"0","enterpriseMinapp":"0","audios":[],"postImageUrl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230519082947553.jpg","avatarurls":[],"related_posts":null,"pageviews":352,"next_post_id":1231,"next_post_title":"SpringBoot \u7cfb\u5217\u6559\u7a0b\u4e4b\u5b9a\u4e49\u63a5\u53e3\u8fd4\u56de\u7c7b\u578b\u7684\u51e0\u79cd\u65b9\u5f0f","previous_post_id":1176,"previous_post_title":"Kafka \u6838\u5fc3\u5168\u9762\u603b\u7ed3\uff0c\u9ad8\u53ef\u9760\u9ad8\u6027\u80fd\u6838\u5fc3\u539f\u7406\u63a2\u7a76","_links":{"self":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/posts\/1102","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/comments?post=1102"}],"version-history":[{"count":0,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/posts\/1102\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/media\/1097"}],"wp:attachment":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/media?parent=1102"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/categories?post=1102"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/tags?post=1102"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}