{"id":872,"date":"2023-03-09T21:45:17","date_gmt":"2023-03-09T13:45:17","guid":{"rendered":"https:\/\/www.appblog.cn\/?p=872"},"modified":"2023-04-29T16:43:35","modified_gmt":"2023-04-29T08:43:35","slug":"spring-boot-integrate-kafka-to-implement-producer-and-consumer","status":"publish","type":"post","link":"https:\/\/www.appblog.cn\/index.php\/2023\/03\/09\/spring-boot-integrate-kafka-to-implement-producer-and-consumer\/","title":{"rendered":"Spring Boot\u96c6\u6210Kafka\u5b9e\u73b0producer\u548cconsumer"},"content":{"rendered":"

\u4ecb\u7ecd\u5982\u4f55\u5728Spring Boot\u9879\u76ee\u4e2d\u96c6\u6210Kafka\u6536\u53d1Message<\/p>\n

pom.xml\u4f9d\u8d56<\/h2>\n

<\/p>\n

<!-- https:\/\/mvnrepository.com\/artifact\/org.springframework.kafka\/spring-kafka -->\n<dependency>\n    <groupId>org.springframework.kafka<\/groupId>\n    <artifactId>spring-kafka<\/artifactId>\n    <version>2.2.4.RELEASE<\/version>\n<\/dependency><\/code><\/pre>\n
\u914d\u7f6e\u6587\u4ef6<\/h2>\n
#=================== kafka ===================\nkafka.consumer.zookeeper.connect=127.0.0.1:2181\nkafka.consumer.servers=127.0.0.1:9092\nkafka.consumer.enable.auto.commit=true\nkafka.consumer.session.timeout=6000\nkafka.consumer.auto.commit.interval=100\nkafka.consumer.auto.offset.reset=latest\nkafka.consumer.topic=test\nkafka.consumer.group.id=test\nkafka.consumer.concurrency=10\n\nkafka.producer.servers=127.0.0.1:9092\nkafka.producer.retries=0\nkafka.producer.batch.size=4096\nkafka.producer.linger=1\nkafka.producer.buffer.memory=40960<\/code><\/pre>\n
Kafka producer<\/h2>\n
\uff081\uff09\u901a\u8fc7@Configuration<\/code>\u3001@EnableKafka<\/code>\uff0c\u58f0\u660eConfig\u5e76\u4e14\u6253\u5f00KafkaTemplate\u80fd\u529b\u3002
\n\uff082\uff09\u901a\u8fc7@Value<\/code>\u6ce8\u5165application.properties<\/code>\u914d\u7f6e\u6587\u4ef6\u4e2d\u7684Kafka\u914d\u7f6e\u3002
\n\uff083\uff09\u901a\u8fc7@Bean<\/code>\u751f\u6210bean<\/p>\n
import org.apache.kafka.clients.producer.ProducerConfig;\nimport org.apache.kafka.common.serialization.StringSerializer;\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.core.DefaultKafkaProducerFactory;\nimport org.springframework.kafka.core.KafkaTemplate;\nimport org.springframework.kafka.core.ProducerFactory;\n\nimport java.util.HashMap;\nimport java.util.Map;\n\n@Configuration\n@EnableKafka\npublic class KafkaProducerConfig {\n\n    @Value("${kafka.producer.servers}")\n    private String servers;\n    @Value("${kafka.producer.retries}")\n    private int retries;\n    @Value("${kafka.producer.batch.size}")\n    private int batchSize;\n    @Value("${kafka.producer.linger}")\n    private int linger;\n    @Value("${kafka.producer.buffer.memory}")\n    private int bufferMemory;\n\n    public Map<String, Object> producerConfigs() {\n        Map<String, Object> props = new HashMap<>();\n        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);\n        props.put(ProducerConfig.RETRIES_CONFIG, retries);\n        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);\n        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);\n        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);\n        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\n        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\n        return props;\n    }\n\n    public ProducerFactory<String, String> producerFactory() {\n        return new DefaultKafkaProducerFactory<>(producerConfigs());\n    }\n\n    @Bean\n    public KafkaTemplate<String, String> kafkaTemplate() {\n        return new KafkaTemplate<String, String>(producerFactory());\n    }\n}<\/code><\/pre>\n
\u6d4b\u8bd5\u6211\u4eec\u7684producer\uff0c\u5199\u4e00\u4e2aController\u3002\u5411topic=test<\/code>\uff0ckey=key<\/code>\u53d1\u9001\u6d88\u606fmessage<\/p>\n
\u8bbf\u95ee\uff1ahttp:\/\/localhost:8080\/kafka\/send?message=Joe.Ye<\/a><\/p>\n
@RestController\n@RequestMapping("\/kafka")\npublic class ProducerController {\n\n    protected final Logger logger = LoggerFactory.getLogger(this.getClass());\n\n    @Autowired\n    private KafkaTemplate kafkaTemplate;\n\n    @ResponseBody\n    @RequestMapping(value = "\/send", method = RequestMethod.GET)\n    public String sendKafka(HttpServletRequest request, HttpServletResponse response) {\n        try {\n            String message = request.getParameter("message");\n            logger.info("kafka\u7684\u6d88\u606f={}", message);\n            kafkaTemplate.send("test", "key", message);\n            logger.info("\u53d1\u9001kafka\u6210\u529f");\n            return new Response(ResultCode.SUCCESS, "\u53d1\u9001kafka\u6210\u529f", null).toString();\n        } catch (Exception e) {\n            logger.error("\u53d1\u9001kafka\u5931\u8d25", e);\n            return new Response(ResultCode.EXCEPTION, "\u53d1\u9001kafka\u5931\u8d25", null).toString();\n        }\n    }\n\n}<\/code><\/pre>\n
kafka consumer<\/h2>\n
\uff081\uff09\u901a\u8fc7@Configuration<\/code>\u3001@EnableKafka<\/code>\uff0c\u58f0\u660eConfig\u5e76\u4e14\u6253\u5f00KafkaTemplate\u80fd\u529b
\n\uff082\uff09\u901a\u8fc7@Value<\/code>\u6ce8\u5165application.properties<\/code>\u914d\u7f6e\u6587\u4ef6\u4e2d\u7684Kafka\u914d\u7f6e
\n\uff083\uff09\u901a\u8fc7@Bean<\/code>\u751f\u6210bean<\/p>\n
import me.yezhou.springboot.kafka.listener.Listener;\nimport org.apache.kafka.clients.consumer.ConsumerConfig;\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.ConcurrentMessageListenerContainer;\n\nimport java.util.HashMap;\nimport java.util.Map;\n\n@Configuration\n@EnableKafka\npublic class KafkaConsumerConfig {\n\n    @Value("${kafka.consumer.servers}")\n    private String servers;\n    @Value("${kafka.consumer.enable.auto.commit}")\n    private boolean enableAutoCommit;\n    @Value("${kafka.consumer.session.timeout}")\n    private String sessionTimeout;\n    @Value("${kafka.consumer.auto.commit.interval}")\n    private String autoCommitInterval;\n    @Value("${kafka.consumer.group.id}")\n    private String groupId;\n    @Value("${kafka.consumer.auto.offset.reset}")\n    private String autoOffsetReset;\n    @Value("${kafka.consumer.concurrency}")\n    private int concurrency;\n\n    @Bean\n    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {\n        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();\n        factory.setConsumerFactory(consumerFactory());\n        factory.setConcurrency(concurrency);\n        factory.getContainerProperties().setPollTimeout(1500);\n        return factory;\n    }\n\n    public ConsumerFactory<String, String> consumerFactory() {\n        return new DefaultKafkaConsumerFactory<>(consumerConfigs());\n    }\n\n    public Map<String, Object> consumerConfigs() {\n        Map<String, Object> propsMap = new HashMap<>();\n        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);\n        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);\n        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);\n        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);\n        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\n        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\n        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);\n        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);\n        return propsMap;\n    }\n\n    @Bean\n    public Listener listener() {\n        return new Listener();\n    }\n}<\/code><\/pre>\n
new Listener()<\/code>\u751f\u6210\u4e00\u4e2abean\u7528\u6765\u5904\u7406\u4ecekafka\u8bfb\u53d6\u7684\u6570\u636e\u3002Listener\u7b80\u5355\u7684\u5b9e\u73b0\u5982\u4e0b\uff1a\u53ea\u662f\u7b80\u5355\u7684\u8bfb\u53d6\u5e76\u6253\u5370key\u548cmessage\u503c<\/p>\n
@KafkaListener<\/code>\u4e2dtopics\u5c5e\u6027\u7528\u4e8e\u6307\u5b9akafka topic\u540d\u79f0\uff0ctopic\u540d\u79f0\u7531\u6d88\u606f\u751f\u4ea7\u8005\u6307\u5b9a\uff0c\u4e5f\u5c31\u662f\u7531kafkaTemplate\u5728\u53d1\u9001\u6d88\u606f\u65f6\u6307\u5b9a\u3002<\/p>\n
import org.apache.kafka.clients.consumer.ConsumerRecord;\nimport org.slf4j.Logger;\nimport org.slf4j.LoggerFactory;\nimport org.springframework.kafka.annotation.KafkaListener;\n\npublic class Listener {\n    protected final Logger logger = LoggerFactory.getLogger(this.getClass());\n\n    @KafkaListener(topics = {"test"})\n    public void listen(ConsumerRecord<?, ?> record) {\n        logger.info("Kafka\u7684key: " + record.key());\n        logger.info("Kafka\u7684value: " + record.value().toString());\n    }\n}<\/code><\/pre>\n
Kafka version : 2.0.1\nKafka commitId : fa14705e51bd2ce5\nCluster ID: HhhOI8SET2yHTesb9OQKbg\n\u53d1\u9001kafka\u6210\u529f\nKafka\u7684key: key\nKafka\u7684value: Joe.Ye<\/code><\/pre>\n
\u6ce8\u610f\u4e8b\u9879<\/h2>\n
\uff081\uff09\u914d\u7f6eKafka\u65f6\u6700\u597d\u7528\u5b8c\u5168bind\u7f51\u7edcip\u7684\u65b9\u5f0f\uff0c\u800c\u4e0d\u662flocalhost\u6216\u8005127.0.0.1
\n\uff082\uff09\u6700\u597d\u4e0d\u8981\u4f7f\u7528Kafka\u81ea\u5e26\u7684ZooKeeper\u90e8\u7f72Kafka\uff0c\u53ef\u80fd\u5bfc\u81f4\u8bbf\u95ee\u4e0d\u901a
\n\uff083\uff09\u5b9a\u4e49\u76d1\u542c\u6d88\u606f\u914d\u7f6e\u65f6\uff0cGROUP_ID_CONFIG\u914d\u7f6e\u9879\u7684\u503c\u7528\u4e8e\u6307\u5b9a\u6d88\u8d39\u8005\u7ec4\u7684\u540d\u79f0\uff0c\u5982\u679c\u540c\u7ec4\u4e2d\u5b58\u5728\u591a\u4e2a\u76d1\u542c\u5668\u5bf9\u8c61\u5219\u53ea\u6709\u4e00\u4e2a\u76d1\u542c\u5668\u5bf9\u8c61\u80fd\u6536\u5230\u6d88\u606f<\/p>\n","protected":false},"excerpt":{"rendered":"
\u4ecb\u7ecd\u5982\u4f55\u5728Spring Boot\u9879\u76ee\u4e2d\u96c6\u6210Kafka\u6536\u53d1Message pom.xml\u4f9d\u8d56 <!– h […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[31,41],"tags":[183],"class_list":["post-872","post","type-post","status-publish","format-standard","hentry","category-kafka","category-spring-boot","tag-kafka"],"_links":{"self":[{"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/posts\/872","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/comments?post=872"}],"version-history":[{"count":0,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/posts\/872\/revisions"}],"wp:attachment":[{"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/media?parent=872"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/categories?post=872"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/tags?post=872"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}