{"id":2145,"date":"2022-02-23T00:09:09","date_gmt":"2022-02-22T16:09:09","guid":{"rendered":"https:\/\/www.zeroheart.xyz\/wordpress\/?p=2145"},"modified":"2022-02-23T15:25:20","modified_gmt":"2022-02-23T07:25:20","slug":"spring-kafka%e7%ae%80%e5%8d%95%e4%bd%bf%e7%94%a8","status":"publish","type":"post","link":"https:\/\/www.zeroheart.xyz\/wordpress\/?p=2145","title":{"rendered":"spring-kafka\u7b80\u5355\u4f7f\u7528"},"content":{"rendered":"\n

1.\u542f\u52a8kafka\u670d\u52a1\uff0c\u6216\u8005\u4f7f\u7528spring-kafka-test\u6765\u6a21\u62df\u542f\u52a8<\/p>\n\n\n\n

<dependency>\n            <groupId>org.springframework.kafka<\/groupId>\n            <artifactId>spring-kafka-test<\/artifactId>\n            <version>2.8.2<\/version>\n            <scope>compile<\/scope>\n        <\/dependency>        \n\n@RunWith(SpringRunner.class)\n@SpringBootTest(classes = DemoApplicationTests.class)\n@EmbeddedKafka(count = 4, ports = {9092,9093,9094,9095})\npublic class DemoApplicationTests {\n\n    @Test\n    public void contextLoads()throws IOException {\n        System.in.read();\n    }\n\n}\n<\/pre>\n\n\n\n
2.\u53ef\u4ee5\u5148\u521b\u5efa\u6216\u8005\u53d1\u9001\u65f6\u5019\u521b\u5efatopic<\/p>\n\n\n\n
@Configuration
public class KafkaInitialConfiguration {
    \/\/ \u521b\u5efa\u4e00\u4e2a\u540d\u4e3atesttopic\u7684Topic\u5e76\u8bbe\u7f6e\u5206\u533a\u6570\u4e3a8\uff0c\u5206\u533a\u526f\u672c\u6570\u4e3a2
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic(\"topic1\",8, (short) 2 );
    }
    \/\/ \u5982\u679c\u8981\u4fee\u6539\u5206\u533a\u6570\uff0c\u53ea\u9700\u4fee\u6539\u914d\u7f6e\u503c\u91cd\u542f\u9879\u76ee\u5373\u53ef
    \/\/ \u4fee\u6539\u5206\u533a\u6570\u5e76\u4e0d\u4f1a\u5bfc\u81f4\u6570\u636e\u7684\u4e22\u5931\uff0c\u4f46\u662f\u5206\u533a\u6570\u53ea\u80fd\u589e\u5927\u4e0d\u80fd\u51cf\u5c0f
    @Bean
    public NewTopic updateTopic() {
        return new NewTopic(\"topic1\",10, (short) 2 );
    }
}<\/pre>\n\n\n\n
3.KafkaProducer<\/p>\n\n\n\n
@RestController\npublic class KafkaProducer {\n    @Autowired\n    private KafkaTemplate<String, Object> kafkaTemplate;\n\n    \/\/ \u53d1\u9001\u6d88\u606f\n    @GetMapping(\"\/kafka\/normal\/{message}\")\n    public void sendMessage1(@PathVariable(\"message\") String normalMessage) {\n        kafkaTemplate.send(\"topic1\", normalMessage);\n    }\n\n    \/**\n     * \u4f1a\u56de\u6eda, \u589e\u52a0\u914d\u7f6e<\/em>spring.kafka.producer.transaction-id-prefix=kafka_tx.<\/mark>\n     * <\/em>@param <\/em><\/strong>input\n     *\/\n    <\/em>@GetMapping(\"\/kafka\/send\/{input}\")\n    @Transactional(rollbackFor = RuntimeException.class)\n    public void sendFoo(@PathVariable String input) {\n        kafkaTemplate.send(\"topic1\", \"kl\");\n        if (\"error\".equals(input)) {\n            throw new RuntimeException(\"failed\");\n        }\n        kafkaTemplate.send(\"topic1\", \"ckl\");\n    }\n}<\/pre>\n\n\n\n
4.KafkaConsumer<\/p>\n\n\n\n
@Component
public class KafkaConsumer {
    \/\/ \u6d88\u8d39\u76d1\u542c
    @KafkaListener(id = \"webGroup\", topics = {\"topic1\"})
    public void onMessage1(ConsumerRecord<?, ?> record){
        \/\/ \u6d88\u8d39\u7684\u54ea\u4e2atopic\u3001partition\u7684\u6d88\u606f,\u6253\u5370\u51fa\u6d88\u606f\u5185\u5bb9
        System.out<\/em>.println(\"\u7b80\u5355\u6d88\u8d39\uff1a\"+record.topic()+\"-\"+record.partition()+\"-\"+record.value());
    }
}<\/pre>\n\n\n\n
\u4ee5\u4e0a\u662f\u57fa\u7840\u7684\u64cd\u4f5c\uff0c\u8fd8\u53ef\u4ee5\u8bbe\u7f6e\u56de\u8c03\uff0c\u8bbe\u7f6e\u5b9a\u65f6\u5668\u6765\u542f\u52a8\u6216\u8005\u505c\u6b62consumer\u7684\u76d1\u542c\u7b49\uff0c\u53c2\u8003\uff1a(25\u6761\u6d88\u606f) SpringBoot\u96c6\u6210kafka\u5168\u9762\u5b9e\u6218_Felix-CSDN\u535a\u5ba2_springboot\u96c6\u6210kafka<\/a><\/p>\n\n\n\n
\u5305\u62ec\u4ee5\u4e0b\u5185\u5bb9\uff1a<\/p>\n\n\n\n
5.\u4e00\u4e9b\u914d\u7f6e\u53c2\u6570<\/p>\n\n\n\n
###########\u3010Kafka\u96c6\u7fa4\u3011###########\nspring.kafka.bootstrap-servers=112.126.74.249:9092,112.126.74.249:9093\n###########\u3010\u521d\u59cb\u5316\u751f\u4ea7\u8005\u914d\u7f6e\u3011###########\n# \u91cd\u8bd5\u6b21\u6570\nspring.kafka.producer.retries=0\n# \u5e94\u7b54\u7ea7\u522b:\u591a\u5c11\u4e2a\u5206\u533a\u526f\u672c\u5907\u4efd\u5b8c\u6210\u65f6\u5411\u751f\u4ea7\u8005\u53d1\u9001ack\u786e\u8ba4(\u53ef\u90090\u30011\u3001all\/-1)\nspring.kafka.producer.acks=1\n# \u6279\u91cf\u5927\u5c0f\nspring.kafka.producer.batch-size=16384\n# \u63d0\u4ea4\u5ef6\u65f6\nspring.kafka.producer.properties.linger.ms=0\n# \u5f53\u751f\u4ea7\u7aef\u79ef\u7d2f\u7684\u6d88\u606f\u8fbe\u5230batch-size\u6216\u63a5\u6536\u5230\u6d88\u606flinger.ms\u540e,\u751f\u4ea7\u8005\u5c31\u4f1a\u5c06\u6d88\u606f\u63d0\u4ea4\u7ed9kafka\n# linger.ms\u4e3a0\u8868\u793a\u6bcf\u63a5\u6536\u5230\u4e00\u6761\u6d88\u606f\u5c31\u63d0\u4ea4\u7ed9kafka,\u8fd9\u65f6\u5019batch-size\u5176\u5b9e\u5c31\u6ca1\u7528\u4e86\n\u200b\n# \u751f\u4ea7\u7aef\u7f13\u51b2\u533a\u5927\u5c0f\nspring.kafka.producer.buffer-memory = 33554432\n# Kafka\u63d0\u4f9b\u7684\u5e8f\u5217\u5316\u548c\u53cd\u5e8f\u5217\u5316\u7c7b\nspring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer\nspring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer\n# \u81ea\u5b9a\u4e49\u5206\u533a\u5668\n# spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner\n\u200b\n###########\u3010\u521d\u59cb\u5316\u6d88\u8d39\u8005\u914d\u7f6e\u3011###########\n# \u9ed8\u8ba4\u7684\u6d88\u8d39\u7ec4ID\nspring.kafka.consumer.properties.group.id=defaultConsumerGroup\n# \u662f\u5426\u81ea\u52a8\u63d0\u4ea4offset\nspring.kafka.consumer.enable-auto-commit=true\n# \u63d0\u4ea4offset\u5ef6\u65f6(\u63a5\u6536\u5230\u6d88\u606f\u540e\u591a\u4e45\u63d0\u4ea4offset)\nspring.kafka.consumer.auto.commit.interval.ms=1000\n# \u5f53kafka\u4e2d\u6ca1\u6709\u521d\u59cboffset\u6216offset\u8d85\u51fa\u8303\u56f4\u65f6\u5c06\u81ea\u52a8\u91cd\u7f6eoffset\n# earliest:\u91cd\u7f6e\u4e3a\u5206\u533a\u4e2d\u6700\u5c0f\u7684offset;\n# latest:\u91cd\u7f6e\u4e3a\u5206\u533a\u4e2d\u6700\u65b0\u7684offset(\u6d88\u8d39\u5206\u533a\u4e2d\u65b0\u4ea7\u751f\u7684\u6570\u636e);\n# none:\u53ea\u8981\u6709\u4e00\u4e2a\u5206\u533a\u4e0d\u5b58\u5728\u5df2\u63d0\u4ea4\u7684offset,\u5c31\u629b\u51fa\u5f02\u5e38;\nspring.kafka.consumer.auto-offset-reset=latest\n# \u6d88\u8d39\u4f1a\u8bdd\u8d85\u65f6\u65f6\u95f4(\u8d85\u8fc7\u8fd9\u4e2a\u65f6\u95f4consumer\u6ca1\u6709\u53d1\u9001\u5fc3\u8df3,\u5c31\u4f1a\u89e6\u53d1rebalance\u64cd\u4f5c)\nspring.kafka.consumer.properties.session.timeout.ms=120000\n# \u6d88\u8d39\u8bf7\u6c42\u8d85\u65f6\u65f6\u95f4\nspring.kafka.consumer.properties.request.timeout.ms=180000\n# Kafka\u63d0\u4f9b\u7684\u5e8f\u5217\u5316\u548c\u53cd\u5e8f\u5217\u5316\u7c7b\nspring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer\nspring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer\n# \u6d88\u8d39\u7aef\u76d1\u542c\u7684topic\u4e0d\u5b58\u5728\u65f6\uff0c\u9879\u76ee\u542f\u52a8\u4f1a\u62a5\u9519(\u5173\u6389)\nspring.kafka.listener.missing-topics-fatal=false\n# \u8bbe\u7f6e\u6279\u91cf\u6d88\u8d39\n# spring.kafka.listener.type=batch\n# \u6279\u91cf\u6d88\u8d39\u6bcf\u6b21\u6700\u591a\u6d88\u8d39\u591a\u5c11\u6761\u6d88\u606f\n# spring.kafka.consumer.max-poll-records=50<\/pre>\n\n\n\n
6.\u5e26\u56de\u8c03\u7684\u751f\u4ea7\u8005<\/p>\n\n\n\n
kafkaTemplate\u63d0\u4f9b\u4e86\u4e00\u4e2a\u56de\u8c03\u65b9\u6cd5addCallback\uff0c\u6211\u4eec\u53ef\u4ee5\u5728\u56de\u8c03\u65b9\u6cd5\u4e2d\u76d1\u63a7\u6d88\u606f\u662f\u5426\u53d1\u9001\u6210\u529f \u6216 \u5931\u8d25\u65f6\u505a\u8865\u507f\u5904\u7406\uff0c\u6709\u4e24\u79cd\u5199\u6cd5\uff0c\n\n@GetMapping(\"\/kafka\/callbackOne\/{message}\")\npublic void sendMessage2(@PathVariable(\"message\") String callbackMessage) {\n    kafkaTemplate.send(\"topic1\", callbackMessage).addCallback(success -> {\n        \/\/ \u6d88\u606f\u53d1\u9001\u5230\u7684topic\n        String topic = success.getRecordMetadata().topic();\n        \/\/ \u6d88\u606f\u53d1\u9001\u5230\u7684\u5206\u533a\n        int partition = success.getRecordMetadata().partition();\n        \/\/ \u6d88\u606f\u5728\u5206\u533a\u5185\u7684offset\n        long offset = success.getRecordMetadata().offset();\n        System.out.println(\"\u53d1\u9001\u6d88\u606f\u6210\u529f:\" + topic + \"-\" + partition + \"-\" + offset);\n    }, failure -> {\n        System.out.println(\"\u53d1\u9001\u6d88\u606f\u5931\u8d25:\" + failure.getMessage());\n    });\n}\n\n\n\n@GetMapping(\"\/kafka\/callbackTwo\/{message}\")\npublic void sendMessage3(@PathVariable(\"message\") String callbackMessage) {\n    kafkaTemplate.send(\"topic1\", callbackMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {\n        @Override\n        public void onFailure(Throwable ex) {\n            System.out.println(\"\u53d1\u9001\u6d88\u606f\u5931\u8d25\uff1a\"+ex.getMessage());\n        }\n \n        @Override\n        public void onSuccess(SendResult<String, Object> result) {\n            System.out.println(\"\u53d1\u9001\u6d88\u606f\u6210\u529f\uff1a\" + result.getRecordMetadata().topic() + \"-\"\n                    + result.getRecordMetadata().partition() + \"-\" + result.getRecordMetadata().offset());\n        }\n    });\n}\n\n<\/pre>\n\n\n\n
7.\u81ea\u5b9a\u4e49\u5206\u533a\u5668<\/p>\n\n\n\n
\u6211\u4eec\u77e5\u9053\uff0ckafka\u4e2d\u6bcf\u4e2atopic\u88ab\u5212\u5206\u4e3a\u591a\u4e2a\u5206\u533a\uff0c\u90a3\u4e48\u751f\u4ea7\u8005\u5c06\u6d88\u606f\u53d1\u9001\u5230topic\u65f6\uff0c\u5177\u4f53\u8ffd\u52a0\u5230\u54ea\u4e2a\u5206\u533a\u5462\uff1f\u8fd9\u5c31\u662f\u6240\u8c13\u7684\u5206\u533a\u7b56\u7565\uff0cKafka \u4e3a\u6211\u4eec\u63d0\u4f9b\u4e86\u9ed8\u8ba4\u7684\u5206\u533a\u7b56\u7565\uff0c\u540c\u65f6\u5b83\u4e5f\u652f\u6301\u81ea\u5b9a\u4e49\u5206\u533a\u7b56\u7565\u3002\u5176\u8def\u7531\u673a\u5236\u4e3a\uff1a\n\n\u2460 \u82e5\u53d1\u9001\u6d88\u606f\u65f6\u6307\u5b9a\u4e86\u5206\u533a\uff08\u5373\u81ea\u5b9a\u4e49\u5206\u533a\u7b56\u7565\uff09\uff0c\u5219\u76f4\u63a5\u5c06\u6d88\u606fappend\u5230\u6307\u5b9a\u5206\u533a\uff1b\n\n\u2461 \u82e5\u53d1\u9001\u6d88\u606f\u65f6\u672a\u6307\u5b9a patition\uff0c\u4f46\u6307\u5b9a\u4e86 key\uff08kafka\u5141\u8bb8\u4e3a\u6bcf\u6761\u6d88\u606f\u8bbe\u7f6e\u4e00\u4e2akey\uff09\uff0c\u5219\u5bf9key\u503c\u8fdb\u884chash\u8ba1\u7b97\uff0c\u6839\u636e\u8ba1\u7b97\u7ed3\u679c\u8def\u7531\u5230\u6307\u5b9a\u5206\u533a\uff0c\u8fd9\u79cd\u60c5\u51b5\u4e0b\u53ef\u4ee5\u4fdd\u8bc1\u540c\u4e00\u4e2a Key \u7684\u6240\u6709\u6d88\u606f\u90fd\u8fdb\u5165\u5230\u76f8\u540c\u7684\u5206\u533a\uff1b\n\n\u2462  patition \u548c key \u90fd\u672a\u6307\u5b9a\uff0c\u5219\u4f7f\u7528kafka\u9ed8\u8ba4\u7684\u5206\u533a\u7b56\u7565\uff0c\u8f6e\u8be2\u9009\u51fa\u4e00\u4e2a patition\uff1b\n\n\u203b \u6211\u4eec\u6765\u81ea\u5b9a\u4e49\u4e00\u4e2a\u5206\u533a\u7b56\u7565\uff0c\u5c06\u6d88\u606f\u53d1\u9001\u5230\u6211\u4eec\u6307\u5b9a\u7684partition\uff0c\u9996\u5148\u65b0\u5efa\u4e00\u4e2a\u5206\u533a\u5668\u7c7b\u5b9e\u73b0Partitioner\u63a5\u53e3\uff0c\u91cd\u5199\u65b9\u6cd5\uff0c\u5176\u4e2dpartition\u65b9\u6cd5\u7684\u8fd4\u56de\u503c\u5c31\u8868\u793a\u5c06\u6d88\u606f\u53d1\u9001\u5230\u51e0\u53f7\u5206\u533a\uff0c\n\n\n\n\n\n\npublic class CustomizePartitioner implements Partitioner {\n    @Override\n    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {\n        \/\/ \u81ea\u5b9a\u4e49\u5206\u533a\u89c4\u5219(\u8fd9\u91cc\u5047\u8bbe\u5168\u90e8\u53d1\u52300\u53f7\u5206\u533a)\n        \/\/ ......\n        return 0;\n    }\n\u200b\n    @Override\n    public void close() {\n\u200b\n    }\n\u200b\n    @Override\n    public void configure(Map<String, ?> configs) {\n\u200b\n    }\n}\n\n\n\u5728application.propertise\u4e2d\u914d\u7f6e\u81ea\u5b9a\u4e49\u5206\u533a\u5668\uff0c\u914d\u7f6e\u7684\u503c\u5c31\u662f\u5206\u533a\u5668\u7c7b\u7684\u5168\u8def\u5f84\u540d\uff0c\n\n# \u81ea\u5b9a\u4e49\u5206\u533a\u5668\nspring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner\n\n<\/pre>\n\n\n\n
8.kafka\u4e8b\u52a1\u63d0\u4ea4<\/p>\n\n\n\n
\u5982\u679c\u5728\u53d1\u9001\u6d88\u606f\u65f6\u9700\u8981\u521b\u5efa\u4e8b\u52a1\uff0c\u53ef\u4ee5\u4f7f\u7528 KafkaTemplate \u7684 executeInTransaction \u65b9\u6cd5\u6765\u58f0\u660e\u4e8b\u52a1\uff0c\n\n@GetMapping(\"\/kafka\/transaction\")\npublic void sendMessage7(){\n    \/\/ \u58f0\u660e\u4e8b\u52a1\uff1a\u540e\u9762\u62a5\u9519\u6d88\u606f\u4e0d\u4f1a\u53d1\u51fa\u53bb\n    kafkaTemplate.executeInTransaction(operations -> {\n        operations.send(\"topic1\",\"test executeInTransaction\");\n        throw new RuntimeException(\"fail\");\n    });\n\u200b\n    \/\/ \u4e0d\u58f0\u660e\u4e8b\u52a1\uff1a\u540e\u9762\u62a5\u9519\u4f46\u524d\u9762\u6d88\u606f\u5df2\u7ecf\u53d1\u9001\u6210\u529f\u4e86\n   kafkaTemplate.send(\"topic1\",\"test executeInTransaction\");\n   throw new RuntimeException(\"fail\");\n}<\/pre>\n\n\n\n
9.\u6307\u5b9atopic\u3001partition\u3001offset\u6d88\u8d39<\/p>\n\n\n\n
\/**\n * @Title \u6307\u5b9atopic\u3001partition\u3001offset\u6d88\u8d39\n * @Description \u540c\u65f6\u76d1\u542ctopic1\u548ctopic2\uff0c\u76d1\u542ctopic1\u76840\u53f7\u5206\u533a\u3001topic2\u7684 \"0\u53f7\u548c1\u53f7\" \u5206\u533a\uff0c\u6307\u54111\u53f7\u5206\u533a\u7684offset\u521d\u59cb\u503c\u4e3a8\n * @Author long.yuan\n * @Date 2020\/3\/22 13:38\n * @Param [record]\n * @return void\n **\/\n@KafkaListener(id = \"consumer1\",groupId = \"felix-group\",topicPartitions = {\n        @TopicPartition(topic = \"topic1\", partitions = { \"0\" }),\n        @TopicPartition(topic = \"topic2\", partitions = \"0\", partitionOffsets = @PartitionOffset(partition = \"1\", initialOffset = \"8\"))\n})\npublic void onMessage2(ConsumerRecord<?, ?> record) {\n    System.out.println(\"topic:\"+record.topic()+\"|partition:\"+record.partition()+\"|offset:\"+record.offset()+\"|value:\"+record.value());\n}\n\ntopics\u548ctopicPartitions\u4e0d\u80fd\u540c\u65f6\u4f7f\u7528\uff1b<\/mark><\/pre>\n\n\n\n
10.\u6279\u91cf\u6d88\u8d39<\/p>\n\n\n\n
# \u8bbe\u7f6e\u6279\u91cf\u6d88\u8d39\nspring.kafka.listener.type=batch\n# \u6279\u91cf\u6d88\u8d39\u6bcf\u6b21\u6700\u591a\u6d88\u8d39\u591a\u5c11\u6761\u6d88\u606f\nspring.kafka.consumer.max-poll-records=50\n\n@KafkaListener(id = \"consumer2\",groupId = \"felix-group\", topics = \"topic1\")\npublic void onMessage3(List<ConsumerRecord<?, ?>> records) {\n    System.out.println(\">>>\u6279\u91cf\u6d88\u8d39\u4e00\u6b21\uff0crecords.size()=\"+records.size());\n    for (ConsumerRecord<?, ?> record : records) {\n        System.out.println(record.value());\n    }\n}\n\n<\/pre>\n\n\n\n
11.\u81ea\u5b9a\u4e49\u5f02\u5e38\u5904\u7406<\/p>\n\n\n\n
\/\/ \u65b0\u5efa\u4e00\u4e2a\u5f02\u5e38\u5904\u7406\u5668\uff0c\u7528@Bean\u6ce8\u5165\n@Bean\npublic ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {\n    return (message, exception, consumer) -> {\n        System.out.println(\"\u6d88\u8d39\u5f02\u5e38\uff1a\"+message.getPayload());\n        return null;\n    };\n}\n\u200b\n\/\/ \u5c06\u8fd9\u4e2a\u5f02\u5e38\u5904\u7406\u5668\u7684BeanName\u653e\u5230@KafkaListener\u6ce8\u89e3\u7684errorHandler\u5c5e\u6027\u91cc\u9762\n@KafkaListener(topics = {\"topic1\"},errorHandler = \"consumerAwareErrorHandler\")\npublic void onMessage4(ConsumerRecord<?, ?> record) throws Exception {\n    throw new Exception(\"\u7b80\u5355\u6d88\u8d39-\u6a21\u62df\u5f02\u5e38\");\n}\n\u200b\n\/\/ \u6279\u91cf\u6d88\u8d39\u4e5f\u4e00\u6837\uff0c\u5f02\u5e38\u5904\u7406\u5668\u7684message.getPayload()\u4e5f\u53ef\u4ee5\u62ff\u5230\u5404\u6761\u6d88\u606f\u7684\u4fe1\u606f\n@KafkaListener(topics = \"topic1\",errorHandler=\"consumerAwareErrorHandler\")\npublic void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {\n    System.out.println(\"\u6279\u91cf\u6d88\u8d39\u4e00\u6b21...\");\n    throw new Exception(\"\u6279\u91cf\u6d88\u8d39-\u6a21\u62df\u5f02\u5e38\");\n}\n<\/pre>\n\n\n\n
\"\"\/<\/figure>\n\n\n\n
<\/p>\n\n\n\n
12.\u6d88\u606f\u8fc7\u6ee4\u5668<\/p>\n\n\n\n
@Component\npublic class KafkaConsumer {\n    @Autowired\n    ConsumerFactory consumerFactory;\n\u200b\n    \/\/ \u6d88\u606f\u8fc7\u6ee4\u5668\n    @Bean\n    public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {\n        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();\n        factory.setConsumerFactory(consumerFactory);\n        \/\/ \u88ab\u8fc7\u6ee4\u7684\u6d88\u606f\u5c06\u88ab\u4e22\u5f03\n        factory.setAckDiscarded(true);\n        \/\/ \u6d88\u606f\u8fc7\u6ee4\u7b56\u7565\n        factory.setRecordFilterStrategy(consumerRecord -> {\n            if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {\n                return false;\n            }\n            \/\/\u8fd4\u56detrue\u6d88\u606f\u5219\u88ab\u8fc7\u6ee4\n            return true;\n        });\n        return factory;\n    }\n\u200b\n    \/\/ \u6d88\u606f\u8fc7\u6ee4\u76d1\u542c\n    @KafkaListener(topics = {\"topic1\"},containerFactory = \"filterContainerFactory\")\n    public void onMessage6(ConsumerRecord<?, ?> record) {\n        System.out.println(record.value());\n    }\n}\n\n<\/pre>\n\n\n\n
13.\u6d88\u606f\u8f6c\u53d1<\/p>\n\n\n\n
\/**\n * @Title \u6d88\u606f\u8f6c\u53d1\n * @Description \u4ecetopic1\u63a5\u6536\u5230\u7684\u6d88\u606f\u7ecf\u8fc7\u5904\u7406\u540e\u8f6c\u53d1\u5230topic2\n * @Author long.yuan\n * @Date 2020\/3\/23 22:15\n * @Param [record]\n * @return void\n **\/\n@KafkaListener(topics = {\"topic1\"})\n@SendTo(\"topic2\")<\/mark>\npublic String onMessage7(ConsumerRecord<?, ?> record) {\n    return<\/mark> record.value()+\"-forward message\";\n}<\/pre>\n\n\n\n
14.\u5b9a\u65f6\u542f\u52a8\u3001\u505c\u6b62\u76d1\u542c\u5668<\/p>\n\n\n\n
\u9ed8\u8ba4\u60c5\u51b5\u4e0b\uff0c\u5f53\u6d88\u8d39\u8005\u9879\u76ee\u542f\u52a8\u7684\u65f6\u5019\uff0c\u76d1\u542c\u5668\u5c31\u5f00\u59cb\u5de5\u4f5c\uff0c\u76d1\u542c\u6d88\u8d39\u53d1\u9001\u5230\u6307\u5b9atopic\u7684\u6d88\u606f\uff0c\u90a3\u5982\u679c\u6211\u4eec\u4e0d\u60f3\u8ba9\u76d1\u542c\u5668\u7acb\u5373\u5de5\u4f5c\uff0c\u60f3\u8ba9\u5b83\u5728\u6211\u4eec\u6307\u5b9a\u7684\u65f6\u95f4\u70b9\u5f00\u59cb\u5de5\u4f5c\uff0c\u6216\u8005\u5728\u6211\u4eec\u6307\u5b9a\u7684\u65f6\u95f4\u70b9\u505c\u6b62\u5de5\u4f5c\uff0c\u8be5\u600e\u4e48\u5904\u7406\u5462\u2014\u2014\u4f7f\u7528KafkaListenerEndpointRegistry\uff0c\u4e0b\u9762\u6211\u4eec\u5c31\u6765\u5b9e\u73b0\uff1a\n\n\u2460 \u7981\u6b62\u76d1\u542c\u5668\u81ea\u542f\u52a8\uff1b\n\n\u2461 \u521b\u5efa\u4e24\u4e2a\u5b9a\u65f6\u4efb\u52a1\uff0c\u4e00\u4e2a\u7528\u6765\u5728\u6307\u5b9a\u65f6\u95f4\u70b9\u542f\u52a8\u5b9a\u65f6\u5668\uff0c\u53e6\u4e00\u4e2a\u5728\u6307\u5b9a\u65f6\u95f4\u70b9\u505c\u6b62\u5b9a\u65f6\u5668\uff1b\n\n\u65b0\u5efa\u4e00\u4e2a\u5b9a\u65f6\u4efb\u52a1\u7c7b\uff0c\u7528\u6ce8\u89e3@EnableScheduling\u58f0\u660e\uff0cKafkaListenerEndpointRegistry \u5728SpringIO\u4e2d\u5df2\u7ecf\u88ab\u6ce8\u518c\u4e3aBean\uff0c\u76f4\u63a5\u6ce8\u5165\uff0c\u8bbe\u7f6e\u7981\u6b62KafkaListener\u81ea\u542f\u52a8\uff0c\n\n@EnableScheduling\n@Component\npublic class CronTimer {\n\u200b\n    \/**\n     * @KafkaListener\u6ce8\u89e3\u6240\u6807\u6ce8\u7684\u65b9\u6cd5\u5e76\u4e0d\u4f1a\u5728IOC\u5bb9\u5668\u4e2d\u88ab\u6ce8\u518c\u4e3aBean\uff0c\n     * \u800c\u662f\u4f1a\u88ab\u6ce8\u518c\u5728KafkaListenerEndpointRegistry\u4e2d\uff0c\n     * \u800cKafkaListenerEndpointRegistry\u5728SpringIOC\u4e2d\u5df2\u7ecf\u88ab\u6ce8\u518c\u4e3aBean\n     **\/\n    @Autowired\n    private KafkaListenerEndpointRegistry registry;\n\u200b\n    @Autowired\n    private ConsumerFactory consumerFactory;\n\u200b\n    \/\/ \u76d1\u542c\u5668\u5bb9\u5668\u5de5\u5382(\u8bbe\u7f6e\u7981\u6b62KafkaListener\u81ea\u542f\u52a8)\n    @Bean\n    public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {\n        ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();\n        container.setConsumerFactory(consumerFactory);\n        \/\/\u7981\u6b62KafkaListener\u81ea\u542f\u52a8\n        container.setAutoStartup(false);\n        return container;\n    }\n\u200b\n    \/\/ \u76d1\u542c\u5668\n    @KafkaListener(id=\"timingConsumer\",topics = \"topic1\",containerFactory = \"delayContainerFactory\")\n    public void onMessage1(ConsumerRecord<?, ?> record){\n        System.out.println(\"\u6d88\u8d39\u6210\u529f\uff1a\"+record.topic()+\"-\"+record.partition()+\"-\"+record.value());\n    }\n\u200b\n    \/\/ \u5b9a\u65f6\u542f\u52a8\u76d1\u542c\u5668\n    @Scheduled(cron = \"0 42 11 * * ? \")\n    public void startListener() {\n        System.out.println(\"\u542f\u52a8\u76d1\u542c\u5668...\");\n        \/\/ \"timingConsumer\"\u662f@KafkaListener\u6ce8\u89e3\u540e\u9762\u8bbe\u7f6e\u7684\u76d1\u542c\u5668ID,\u6807\u8bc6\u8fd9\u4e2a\u76d1\u542c\u5668\n        if (!registry.getListenerContainer(\"timingConsumer\").isRunning()) {\n            registry.getListenerContainer(\"timingConsumer\").start();\n        }\n        \/\/registry.getListenerContainer(\"timingConsumer\").resume();\n    }\n\u200b\n    \/\/ \u5b9a\u65f6\u505c\u6b62\u76d1\u542c\u5668\n    @Scheduled(cron = \"0 45 11 * * ? \")\n    public void shutDownListener() {\n        System.out.println(\"\u5173\u95ed\u76d1\u542c\u5668...\");\n        registry.getListenerContainer(\"timingConsumer\").pause();\n    }\n}\n\n<\/pre>\n\n\n\n
15.\u540c\u7ec4\u4e0d\u540c\u7ec4\u7684\u6d88\u8d39\u8005\u8bbe\u7f6e<\/p>\n\n\n\n
\/\/ \u6d88\u8d39\u76d1\u542c\n    @KafkaListener(id = \"KafkaConsumer2\", groupId = \"webGroup\", topics = {\"topic1\"})\n    public void onMessage(ConsumerRecord<?, ?> record){\n        \/\/ \u6d88\u8d39\u7684\u54ea\u4e2atopic\u3001partition\u7684\u6d88\u606f,\u6253\u5370\u51fa\u6d88\u606f\u5185\u5bb9\n        log.info(\"KafkaConsumer2_webGroup\uff1a\"+record.topic()+\"-\"+record.partition()+\"_\"+record.offset()+\"-\"+record.value());\n    }\n\n    \/\/ \u6d88\u8d39\u76d1\u542c\n    @KafkaListener(id = \"KafkaConsumer2_1\", groupId = \"webGroup\", topics = {\"topic1\"})\n    public void onMessage1(ConsumerRecord<?, ?> record){\n        \/\/ \u6d88\u8d39\u7684\u54ea\u4e2atopic\u3001partition\u7684\u6d88\u606f,\u6253\u5370\u51fa\u6d88\u606f\u5185\u5bb9\n        log.info(\"KafkaConsumer2_1_webGroup\uff1a\"+record.topic()+\"-\"+record.partition()+\"_\"+record.offset()+\"-\"+record.value());\n    }\n\n\/\/ \u6d88\u8d39\u76d1\u542c\n    @KafkaListener(id = \"KafkaConsumer2_1\", groupId = \"webGroup2\", topics = {\"topic1\"})\n    public void onMessage1(ConsumerRecord<?, ?> record){\n        \/\/ \u6d88\u8d39\u7684\u54ea\u4e2atopic\u3001partition\u7684\u6d88\u606f,\u6253\u5370\u51fa\u6d88\u606f\u5185\u5bb9\n       log.info(\"KafkaConsumer2_1_webGroup2\uff1a\"+record.topic()+\"-\"+record.partition()+\"_\"+record.offset()+\"-\"+record.value());\n    }\n\n\u540c\u4e00\u4e2a\u6d88\u606f\u5728\u4e00\u4e2a\u7ec4\u91cc\u53ea\u80fd\u6709\u4e00\u4e2a\u6d88\u8d39\uff0c\u4e0d\u540c\u7ec4\u53ef\u4ee5\u591a\u6b21\u6d88\u8d39\u3002\u5982\u679c\u9700\u8981\u4fdd\u8bc1\u4e00\u4e2a\u7ec4\u6d88\u8d39\u987a\u5e8f\uff0c\u90a3\u5c31\u8981\u53d1\u9001\u5230\u540c\u4e00\u4e2a\u5206\u533a\u3002<\/mark><\/pre>\n\n\n\n
\u914d\u7f6e\u9519\u8bef\uff1a\n\n\u4e0d\u8fc7\u5982\u679c\u8981\u786e\u4fdd\u5e42\u7b49\u6027\u529f\u80fd\u6b63\u5e38\uff0c\u8fd8\u9700\u8981\u786e\u4fdd\u751f\u4ea7\u8005\u5ba2\u6237\u7aef\u7684 retries\u3001acks\u3001max.in.flight.requests.per.connection \u8fd9\u51e0\u4e2a\u53c2\u6570\u4e0d\u88ab\u914d\u7f6e\u9519\u3002\u5b9e\u9645\u4e0a\u5728\u4f7f\u7528\u5e42\u7b49\u6027\u529f\u80fd\u7684\u65f6\u5019\uff0c\u7528\u6237\u5b8c\u5168\u53ef\u4ee5\u4e0d\u7528\u914d\u7f6e\uff08\u4e5f\u4e0d\u5efa\u8bae\u914d\u7f6e\uff09\u8fd9\u51e0\u4e2a\u53c2\u6570\u3002\n\n\u5982\u679c\u7528\u6237\u663e\u5f0f\u5730\u6307\u5b9a\u4e86 retries \u53c2\u6570\uff0c\u90a3\u4e48\u8fd9\u4e2a\u53c2\u6570\u7684\u503c\u5fc5\u987b\u5927\u4e8e0\uff0c\u5426\u5219\u4f1a\u62a5\u51fa ConfigException\uff1a\n\norg.apache.kafka.common.config.ConfigException: Must set retries to non-zero when using the idempotent producer.\n\u5982\u679c\u7528\u6237\u6ca1\u6709\u663e\u5f0f\u5730\u6307\u5b9a retries \u53c2\u6570\uff0c\u90a3\u4e48 KafkaProducer \u4f1a\u5c06\u5b83\u7f6e\u4e3a Integer.MAX_VALUE\u3002\u540c\u65f6\u8fd8\u9700\u8981\u4fdd\u8bc1 max.in.flight.requests.per.connection \u53c2\u6570\u7684\u503c\u4e0d\u80fd\u5927\u4e8e5\uff08\u8fd9\u4e2a\u53c2\u6570\u7684\u503c\u9ed8\u8ba4\u4e3a5\uff0c\u5728 2.2.1 \u8282\u4e2d\u6709\u76f8\u5173\u7684\u4ecb\u7ecd\uff09\uff0c\u5426\u5219\u4e5f\u4f1a\u62a5\u51fa ConfigException\uff1a\n\norg.apache.kafka.common.config.ConfigException: Must set max.in.flight. requests.per.connection to at most 5 to use the idempotent producer.\n\u5982\u679c\u7528\u6237\u8fd8\u663e\u5f0f\u5730\u6307\u5b9a\u4e86 acks \u53c2\u6570\uff0c\u90a3\u4e48\u8fd8\u9700\u8981\u4fdd\u8bc1\u8fd9\u4e2a\u53c2\u6570\u7684\u503c\u4e3a -1\uff08all\uff09\uff0c\u5982\u679c\u4e0d\u4e3a -1\uff08\u8fd9\u4e2a\u53c2\u6570\u7684\u503c\u9ed8\u8ba4\u4e3a1\uff09\uff0c\u90a3\u4e48\u4e5f\u4f1a\u62a5\u51fa ConfigException\uff1a\n\norg.apache.kafka.common.config.ConfigException: Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.\n\u5982\u679c\u7528\u6237\u6ca1\u6709\u663e\u5f0f\u5730\u6307\u5b9a\u8fd9\u4e2a\u53c2\u6570\uff0c\u90a3\u4e48 KafkaProducer \u4f1a\u5c06\u5b83\u7f6e\u4e3a-1\u3002\u5f00\u542f\u5e42\u7b49\u6027\u529f\u80fd\u4e4b\u540e\uff0c\u751f\u4ea7\u8005\u5c31\u53ef\u4ee5\u5982\u540c\u672a\u5f00\u542f\u5e42\u7b49\u65f6\u4e00\u6837\u53d1\u9001\u6d88\u606f\u4e86\u3002\n<\/pre>\n\n\n\n
\u8fd8\u6709\u4f7f\u7528\u4ee3\u7801\u8bbe\u7f6e\u6279\u91cf\u6d88\u8d39\u7684\uff0c\u53c2\u8003\uff1ahttps:\/\/blog.csdn.net\/weixin_43215250\/article\/details\/103241644<\/p>\n\n\n\n
\n\nimport org.apache.kafka.clients.CommonClientConfigs;\nimport org.apache.kafka.clients.consumer.ConsumerConfig;\nimport org.apache.kafka.common.config.SaslConfigs;\nimport org.apache.kafka.common.config.SslConfigs;\nimport org.apache.kafka.common.serialization.StringDeserializer;\nimport org.springframework.beans.factory.annotation.Value;\nimport org.springframework.context.annotation.Bean;\nimport org.springframework.context.annotation.Configuration;\nimport org.springframework.kafka.annotation.EnableKafka;\nimport org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;\nimport org.springframework.kafka.config.KafkaListenerContainerFactory;\nimport org.springframework.kafka.core.ConsumerFactory;\nimport org.springframework.kafka.core.DefaultKafkaConsumerFactory;\nimport org.springframework.kafka.listener.AbstractMessageListenerContainer;\nimport org.springframework.kafka.listener.ConcurrentMessageListenerContainer;\n\nimport java.util.HashMap;\nimport java.util.Map;\n\n@Configuration\n@EnableKafka\npublic class KafkaConsumerConfig {\n\n\t@Value(\"${kafka.bootstrap-servers}\")\n\tprivate String servers;\n\t@Value(\"${kafka.consumer.enable-auto-commit}\")\n\tprivate boolean enableAutoCommit;\n\t@Value(\"${kafka.consumer.session-timeout}\")\n\tprivate String sessionTimeout;\n\t@Value(\"${kafka.consumer.auto-commit-interval}\")\n\tprivate String autoCommitInterval;\n\t@Value(\"${kafka.consumer.group-id}\")\n\tprivate String groupId;\n\t@Value(\"${kafka.consumer.auto-offset-reset}\")\n\tprivate String autoOffsetReset;\n\t@Value(\"${kafka.consumer.concurrency}\")\n\tprivate int concurrency;\n\t@Value(\"${kafka.consumer.max-poll-records}\")\n\tprivate int maxPollRecords;\n\t@Value(\"${kafka.sasl-mechanism}\")\n\tprivate String saslMechanism;\n\t@Value(\"${kafka.ssl.truststore.location}\")\n\tprivate String truststoreLocation;\n\t@Value(\"${kafka.ssl.truststore.password}\")\n\tprivate String truststorePassword;\n\t@Value(\"${kafka.ssl.security-protocol}\")\n\tprivate String securityProtocol;\n\t@Value(\"${java.security.auth.login.config}\")\n\tprivate String authLoginConfig;\n\n\n\t@Bean\n\tpublic KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {\n\t\tConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();\n\t\tfactory.setConsumerFactory(consumerFactory());\n\t\t\/\/ \u5e76\u53d1\u521b\u5efa\u7684\u6d88\u8d39\u8005\u6570\u91cf\n\t\tfactory.setConcurrency(concurrency);\n\t\t\/\/ \u5f00\u542f\u6279\u5904\u7406\n\t\tfactory.setBatchListener(true);\n\t\tfactory.getContainerProperties().setPollTimeout(1500);\n\t\tif (!enableAutoCommit) {\n\t\t\tfactory.getContainerProperties().setAckMode((AbstractMessageListenerContainer.AckMode.MANUAL));\n\t\t}\n\t\treturn factory;\n\t}\n\n\t@Bean\n\tpublic ConsumerFactory<String, String> consumerFactory() {\n\t\treturn new DefaultKafkaConsumerFactory<>(getCommonPropertis(groupId));\n\t}\n\n\tprivate Map<String, Object> getCommonPropertis(String groupId) {\n\t\tMap<String, Object> props = new HashMap<>();\n\t\tprops.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);\n\t\tprops.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);\n\t\tprops.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);\n\t\tprops.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);\n\t\tprops.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\n\t\tprops.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\n\t\tprops.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);\n\t\tprops.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);\n\t\tprops.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);\n\t\tprops.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 10485760);\n\t\tprops.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, 10485760);\n\t\tprops.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);\n\t\tprops.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststoreLocation);\n\t\tprops.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, truststorePassword);\n\t\tprops.put(SaslConfigs.SASL_MECHANISM, saslMechanism);\n\t\treturn props;\n\t}\n}\n\n\n\n\n@KafkaListener(id = \"test\", topics = \"#{'${kafka.consumer.topics}'.split(',')}\", containerFactory = \"kafkaListenerContainerFactory\")\npublic void consumeList(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {\n    logger.info(\"=============== Total \" + records.size() + \" events in this batch ..\");\n    try {\n        List<String> list = new ArrayList<String>();\n        for (ConsumerRecord<String, String> record : records) {\n            Optional<?> kafkaMessage = Optional.ofNullable(record.value());\n            if (kafkaMessage.isPresent()) {\n                String message = record.value();\n                list.add(message);\n                String topic = record.topic();\n                logger.info(\"message = {}\",  message);\n            }\n        }\n        boolean handle = this.batchHandle(list);\n        if (handle) {\n            \/\/ \u76f4\u63a5\u63d0\u4ea4 offset\n            ack.acknowledge();\n        }\n    } catch (Exception e) {\n        logger.error(e.getMessage(), e);\n    }\n}\n<\/pre>\n\n\n\n
<\/p>\n","protected":false},"excerpt":{"rendered":"
1.\u542f\u52a8kafka\u670d\u52a1\uff0c\u6216\u8005\u4f7f\u7528spring-kafka-test\u6765\u6a21\u62df\u542f\u52a8 <dependency&g […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[144,52],"_links":{"self":[{"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/2145"}],"collection":[{"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=2145"}],"version-history":[{"count":61,"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/2145\/revisions"}],"predecessor-version":[{"id":2208,"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/2145\/revisions\/2208"}],"wp:attachment":[{"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=2145"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=2145"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.zeroheart.xyz\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=2145"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}