{"id":217,"date":"2023-05-25T05:52:00","date_gmt":"2023-05-24T21:52:00","guid":{"rendered":"https:\/\/xinchewhd.com.cn\/index.php\/uncategorized\/2d70c5e4e8\/"},"modified":"2023-05-28T10:45:18","modified_gmt":"2023-05-28T02:45:18","slug":"2d70c5e4e8","status":"publish","type":"post","link":"https:\/\/xinchewhd.com.cn\/index.php\/spring-boot\/2d70c5e4e8\/","title":{"rendered":"\u4e00\u8d77\u6765\u5b66kafka\u4e4b\u6574\u5408SpringBoot\u6df1\u5165\u4f7f\u7528(\u4e00)"},"content":{"rendered":"\n
\n
\n

\u4e00\u8d77\u6765\u5b66kafka\u4e4b\u6574\u5408SpringBoot\u6df1\u5165\u4f7f\u7528(\u4e00)<\/span><\/h1>\n

\u524d\u8a00<\/span><\/h2>\n

\u76ee\u524d\u6b63\u5728\u51fa\u4e00\u4e2aKafka\u4e13\u9898<\/code>\u7cfb\u5217\u6559\u7a0b, \u7bc7\u5e45\u4f1a\u8f83\u591a, \u559c\u6b22\u7684\u8bdd\uff0c\u7ed9\u4e2a\u5173\u6ce8\u2764\ufe0f ~<\/p>\n

\u672c\u8282\u7ed9\u5927\u5bb6\u8bb2\u4e00\u4e0bKafka\u6574\u5408SpringBoot<\/code>\u4e2d\u5982\u4f55\u8fdb\u884c\u6d88\u606f\u5e94\u7b54\u4ee5\u53ca@SendTo \u548c @KafkaListener<\/code>\u7684\u8bb2\u89e3~<\/p>\n

\u597d\u4e86, \u5e9f\u8bdd\u4e0d\u591a\u8bf4\u76f4\u63a5\u5f00\u6574\u5427~<\/p>\n

\n

pom\u5f15\u7528<\/strong><\/p>\n<\/blockquote>\n

<dependency><\/code>
<groupId>org.springframework.kafka<\/groupId><\/code>
<artifactId>spring-kafka<\/artifactId><\/code>
<\/dependency><\/code><\/div>\n
\u00a0<\/div>\n<\/section>\n
\n
\n

application.yml\u914d\u7f6e<\/strong><\/p>\n<\/blockquote>\n

kafka:
bootstrap-servers: kafka\u914d\u7f6e\u5730\u5740
consumer:
group-id: atlas.consumer.group
enable-auto-commit: false
auto-commit-interval: 100ms
properties:
session.timeout.ms: 15000
max-poll-interval-ms: 15000
max-partition-fetch-bytes: 15728640

key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
auto-offset-reset: earliest
max-poll-records: 50

producer:
retries: 3
batch-size: 16384
buffer-memory: 33554432
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
acks: 1
compression-type: gzip
listener:
ack-mode: manual_immediate
concurrency: 5
type: batch
# client-id: atlasKafkaCli<\/code><\/div>\n

\u6d88\u606f\u5e94\u7b54<\/span><\/h2>\n

\u6709\u65f6\u5019\uff0c\u6d88\u8d39\u8005\u6d88\u8d39\u6d88\u606f\u7684\u65f6\u5019\uff0c\u6211\u4eec\u9700\u8981\u77e5\u9053\u5b83\u6709\u6ca1\u6709\u6d88\u8d39\u5b8c\uff0c\u9700\u8981\u5b83\u7ed9\u6211\u4eec\u4e00\u4e2a\u56de\u5e94\uff0c\u8be5\u600e\u4e48\u505a\u5462? \u6211\u4eec\u53ef\u4ee5\u901a\u8fc7\u63d0\u4f9b\u7684ReplyingKafkaTemplate<\/code>, \u4e0b\u9762\u901a\u8fc7\u4e00\u4e2a\u4f8b\u5b50\u6765\u4f53\u9a8c\u4e00\u4e0b,\u65b0\u5efa\u4e00\u4e2aReceiveCustomerController<\/code><\/p>\n

@Slf<\/span>4j\n@RestController<\/span>\npublic<\/span>\u00a0class<\/span>\u00a0ReceiveCustomerController<\/span><\/span>{\n\n\u00a0\u00a0\u00a0\u00a0private<\/span>\u00a0static<\/span>\u00a0final<\/span>\u00a0String\u00a0topic\u00a0=\u00a0\"hello3\"<\/span>;\n\u00a0\u00a0\u00a0\u00a0private<\/span>\u00a0static<\/span>\u00a0final<\/span>\u00a0String\u00a0topicCroup\u00a0=\u00a0\"hello3Group\"<\/span>;\n\n\u00a0\u00a0\u00a0\u00a0@Bean<\/span>\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0ConcurrentMessageListenerContainer<String,\u00a0String>\u00a0repliesContainer<\/span>(ConcurrentKafkaListenerContainerFactory<String,\u00a0String>\u00a0containerFactory)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0ConcurrentMessageListenerContainer<String,\u00a0String>\u00a0repliesContainer\u00a0=\u00a0containerFactory.createContainer(topic\u00a0+\u00a0\"_replies\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0repliesContainer.getContainerProperties().setGroupId(topicCroup);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0repliesContainer.setAutoStartup(false<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0repliesContainer;\n\u00a0\u00a0\u00a0\u00a0}\n\n\u00a0\u00a0\u00a0\u00a0@Bean<\/span>\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0ReplyingKafkaTemplate<String,\u00a0String,\u00a0String>\u00a0replyingTemplate<\/span>(ProducerFactory<String,\u00a0String>\u00a0producerFactory,\u00a0ConcurrentMessageListenerContainer<String,\u00a0String>\u00a0repliesContainer)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0new<\/span>\u00a0ReplyingKafkaTemplate(producerFactory,\u00a0repliesContainer);\n\u00a0\u00a0\u00a0\u00a0}\n\n\u00a0\u00a0\u00a0\u00a0@Bean<\/span>\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0KafkaTemplate\u00a0kafkaTemplate<\/span>(ProducerFactory<String,\u00a0String>\u00a0producerFactory)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0new<\/span>\u00a0KafkaTemplate(producerFactory);\n\u00a0\u00a0\u00a0\u00a0}\n\n\u00a0\u00a0\u00a0\u00a0@Autowired<\/span>\n\u00a0\u00a0\u00a0\u00a0private<\/span>\u00a0ReplyingKafkaTemplate\u00a0kafkaReplyTemplate;\n\n\u00a0\u00a0\u00a0\u00a0@GetMapping<\/span>(\"\/send\/{msg}\"<\/span>)\n\u00a0\u00a0\u00a0\u00a0@Transactional<\/span>(rollbackFor\u00a0=\u00a0RuntimeException.class<\/span>)\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0void<\/span>\u00a0sendMsg<\/span>(@PathVariable<\/span>\u00a0String<\/span>\u00a0msg<\/span>)\u00a0throws<\/span>\u00a0Exception<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0ProducerRecord<String,\u00a0String>\u00a0record\u00a0=\u00a0new<\/span>\u00a0ProducerRecord<>(topic,\u00a0msg);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0RequestReplyFuture<String,\u00a0String,\u00a0String>\u00a0replyFuture\u00a0=\u00a0kafkaReplyTemplate.sendAndReceive(record);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0ConsumerRecord<String,\u00a0String>\u00a0consumerRecord\u00a0=\u00a0replyFuture.get();\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0log.info(\"customer\u00a0reply\u00a0>>>>\u00a0{}:\u00a0\"<\/span>,\u00a0consumerRecord.value());\u00a0\/\/\u00a0customer\u00a0reply\u00a0>>>>listen:\u00a0I\u00a0do\u00a0it\u00a0>>>\u00a01:<\/span>\n\u00a0\u00a0\u00a0\u00a0}\n\n\u00a0\u00a0\u00a0\u00a0@KafkaListener<\/span>(id\u00a0=\u00a0topicCroup,\u00a0topics\u00a0=\u00a0topic)\n\u00a0\u00a0\u00a0\u00a0@SendTo<\/span>\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0String\u00a0listen<\/span>(String\u00a0msg)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0log.info(\"listen\u00a0receive\u00a0msg\u00a0>>>\u00a0\u00a0{}\"<\/span>,\u00a0msg);\u00a0\/\/\u00a0listen\u00a0receive\u00a0msg\u00a0>>>\u00a0\u00a01<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0\"listen:\u00a0I\u00a0do\u00a0it\u00a0>>>\u00a0\"<\/span>\u00a0+\u00a0msg;\n\u00a0\u00a0\u00a0\u00a0}\n}\n<\/code><\/pre>\n
\u542f\u52a8\u5e94\u7528\uff0c\u6d4b\u8bd5\u4e00\u4e0b\uff0c\u89c2\u5bdf\u63a7\u5236\u53f0\u7684\u53d8\u5316~<\/p>\n
\u00a0\/\/\u00a0listen\u00a0receive\u00a0msg\u00a0>>>\u00a0\u00a01\n\u00a0\/\/\u00a0customer\u00a0reply\u00a0>>>>listen:\u00a0I\u00a0do\u00a0it\u00a0>>>\u00a01:\n<\/code><\/pre>\n
@SendTo<\/span><\/h2>\n
\u5728 Spring Kafka<\/code> \u4e2d\uff0c@SendTo<\/code> \u6ce8\u89e3\u53ef\u4ee5\u7528\u4e8e\u6307\u5b9a\u6d88\u606f\u88ab\u53d1\u9001\u5230\u7684\u76ee\u6807 Topic<\/code>\u3002\u5f53\u6d88\u8d39\u8005\u6210\u529f\u6d88\u8d39\u4e00\u4e2a\u6d88\u606f\u540e\uff0c\u53ef\u4ee5\u5c06\u7ed3\u679c\u53d1\u9001\u5230\u6307\u5b9a\u7684\u76ee\u6807 Topic<\/code>\uff0c\u4ee5\u4f9b\u5176\u4ed6\u6d88\u8d39\u8005\u8fdb\u4e00\u6b65\u5904\u7406\u3002<\/p>\n
@SendTo<\/code> \u6ce8\u89e3\u53ef\u4ee5\u5e94\u7528\u4e8e Kafka<\/code> \u6d88\u8d39\u8005\u65b9\u6cd5\u4e0a\uff0c\u4ee5\u6307\u5b9a\u6d88\u606f\u7684\u5904\u7406\u7ed3\u679c\u5c06\u88ab\u53d1\u9001\u5230\u54ea\u4e2a Topic<\/code>\u3002\u4e0b\u9762\u901a\u8fc7\u4e00\u4e2a\u4f8b\u5b50\u6765\u6f14\u793a\u4e00\u4e0b\u5982\u4f55\u8fdb\u884c\u6d88\u606f\u7684\u8f6c\u53d1~<\/p>\n
@Slf<\/span>4j\n@RestController<\/span>\npublic<\/span>\u00a0class<\/span>\u00a0SendToController<\/span><\/span>{\n\n\u00a0\u00a0\u00a0\u00a0@Autowired<\/span>\n\u00a0\u00a0\u00a0\u00a0private<\/span>\u00a0KafkaTemplate<Object,\u00a0Object>\u00a0kafkaTemplate;\n\n\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u63a5\u6536\u6d88\u606f<\/span>\n\u00a0\u00a0\u00a0\u00a0@Transactional<\/span>(rollbackFor\u00a0=\u00a0Exception.class<\/span>)\n\u00a0\u00a0\u00a0\u00a0@KafkaListener<\/span>(id<\/span><\/span>=\u00a0\"input\"<\/span>,\u00a0topics\u00a0=\u00a0\"inputTopic\"<\/span>)\n\u00a0\u00a0\u00a0\u00a0@SendTo<\/span>(\"outputTopic\"<\/span>)\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0String\u00a0processMessage<\/span>(String\u00a0message)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u5904\u7406\u6d88\u606f\u5e76\u8fd4\u56de\u7ed3\u679c<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0log.info(\"inputTopic\u00a0>>>>\u00a0{}\"<\/span>,\u00a0message);\u00a0\/\/\u00a0inputTopic\u00a0>>>>\u00a01<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0\"2\"<\/span>;\n\u00a0\u00a0\u00a0\u00a0}\n\n\u00a0\u00a0\u00a0\u00a0@KafkaListener<\/span>(id\u00a0=\u00a0\"output\"<\/span>,\u00a0topics\u00a0=\u00a0\"outputTopic\"<\/span>)\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0String\u00a0process1Message<\/span>(String\u00a0message)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u5904\u7406\u6d88\u606f\u5e76\u8fd4\u56de\u7ed3\u679c<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0String\u00a0result\u00a0=\u00a0\"Processed\u00a0message:\u00a0\"<\/span>\u00a0+\u00a0message;\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0log.info(\"outputTopic\u00a0>>>>\u00a0{}\"<\/span>,\u00a0result);\u00a0\/\/\u00a0outputTopic\u00a0>>>>\u00a0Processed\u00a0message:\u00a02<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0result;\n\u00a0\u00a0\u00a0\u00a0}\n\n\u00a0\u00a0\u00a0\u00a0@Transactional<\/span>(rollbackFor\u00a0=\u00a0Exception.class<\/span>)\n\u00a0\u00a0\u00a0\u00a0@GetMapping<\/span>(\"\/hello<\/span>\")\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0String<\/span>\u00a0hello<\/span>()\u00a0<\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u53d1\u9001\u6d88\u606f<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0kafkaTemplate.send(\"inputTopic\"<\/span>,\u00a0\"1\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0\"hello\"<\/span>;\n\u00a0\u00a0\u00a0\u00a0}\n}\n<\/code><\/pre>\n
\u89c2\u5bdf\u63a7\u5236\u53f0\u7684\u65e5\u5fd7\u4fe1\u606f<\/p>\n
inputTopic\u00a0>>>>\u00a01\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\noutputTopic\u00a0>>>>\u00a0Processed\u00a0message:\u00a02\n<\/code><\/pre>\n
\u53ef\u4ee5\u770b\u5230\u6d88\u606f\u88ab\u8f6c\u53d1\u5230outputTopic<\/code>\u5e76\u4e14\u88aboutput<\/code>\u6d88\u8d39\u8005\u6210\u529f\u6d88\u8d39<\/p>\n
@KafkaListener<\/span><\/h2>\n
@KafkaListener<\/code> \u662f\u4e00\u4e2a\u6ce8\u89e3<\/code>\uff0c\u7528\u4e8e\u6807\u8bb0\u4e00\u4e2a\u65b9\u6cd5\u4f5c\u4e3a Kafka \u6d88\u8d39\u8005<\/code>\u3002\u5728 Spring Boot<\/code> \u5e94\u7528\u7a0b\u5e8f\u4e2d\uff0c\u4f7f\u7528\u8be5\u6ce8\u89e3\u53ef\u4ee5\u65b9\u4fbf\u5730\u5904\u7406Kafka<\/code> \u6d88\u606f\u3002<\/p>\n
@KafkaListener<\/code> \u6ce8\u89e3\u53ef\u4ee5\u6dfb\u52a0\u5230\u7c7b\u7ea7\u522b\u6216\u65b9\u6cd5\u7ea7\u522b\u3002\u5728\u7c7b\u7ea7\u522b\u6dfb\u52a0\u6ce8\u89e3\uff0c\u5c06\u6307\u5b9a\u9ed8\u8ba4\u7684 Topic<\/code> \u548c\u6d88\u8d39\u8005\u7ec4 ID<\/code>\u3002\u5728\u65b9\u6cd5\u7ea7\u522b\u6dfb\u52a0\u6ce8\u89e3\uff0c\u5219\u53ef\u4ee5\u4f7f\u7528\u4e0d\u540c\u7684 Topic<\/code> \u548c\u6d88\u8d39\u8005\u7ec4 ID<\/code>\u3002<\/p>\n
\u5728\u524d\u9762\u7684\u51e0\u4e2a\u4f8b\u5b50\u4e2d\uff0c\u5e26\u5927\u5bb6\u5df2\u7ecf\u4f53\u9a8c\u8fc7\u4e86\uff0c\u4f46\u90fd\u662f\u76d1\u542c\u4e00\u4e2atopic<\/code>\uff0c\u90a3\u4e48\u5982\u4f55\u53bb\u76d1\u542c\u591a\u4e2atopic<\/code>\u5462? \u5176\u5b9e\u5f88\u7b80\u5355\uff0c\u4e0b\u9762\u901a\u8fc7\u4e00\u4e2a\u4f8b\u5b50\u6765\u4f53\u9a8c\u4e0b<\/p>\n
@Slf<\/span>4j\n@RestController<\/span>\npublic<\/span>\u00a0class<\/span>\u00a0ListenerController<\/span><\/span>{\n\n\u00a0\u00a0\u00a0\u00a0@Autowired<\/span>\n\u00a0\u00a0\u00a0\u00a0private<\/span>\u00a0KafkaTemplate<Object,\u00a0Object>\u00a0kafkaTemplate;\n\n\u00a0\u00a0\u00a0\u00a0@Transactional<\/span>(rollbackFor\u00a0=\u00a0Exception.class<\/span>)\n\u00a0\u00a0\u00a0\u00a0@GetMapping<\/span>(\"\/hello<\/span>\")\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0String<\/span>\u00a0hello<\/span>()\u00a0<\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u53d1\u9001\u6d88\u606f<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0kafkaTemplate.send(\"topic1\"<\/span>,\u00a0\"1\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0kafkaTemplate.send(\"topic2\"<\/span>,\u00a0\"2\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0\"hello\"<\/span>;\n\u00a0\u00a0\u00a0\u00a0}\n\n\u00a0\u00a0\u00a0\u00a0\/**\n\u00a0\u00a0\u00a0\u00a0\u00a0*\u00a0\u76d1\u542c\u591a\u4e2atopic\n\u00a0\u00a0\u00a0\u00a0\u00a0*\u00a0@param<\/span>\u00a0message\n\u00a0\u00a0\u00a0\u00a0\u00a0*\/<\/span>\n\u00a0\u00a0\u00a0\u00a0@KafkaListener<\/span>(topics\u00a0=\u00a0{\"topic1\"<\/span>,\u00a0\"topic2\"<\/span>},\u00a0groupId\u00a0=\u00a0\"group1\"<\/span>)\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0void<\/span>\u00a0listen<\/span>(String\u00a0message)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0log.info(\"Received\u00a0message\u00a0>>>\u00a0{}\"<\/span>,\u00a0message);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/\u00a0Received\u00a0message\u00a0>>>\u00a01<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/\u00a0Received\u00a0message\u00a0>>>\u00a02<\/span>\n\u00a0\u00a0\u00a0\u00a0}\n\n}\n\n<\/code><\/pre>\n
\u5982\u4e0a\uff0c\u6211\u4eec\u53d1\u9001\u4e86\u4e24\u6761\u4e0d\u540c\u7684\u6d88\u606ftopic1<\/code>\u548ctopic2<\/code>\uff0c\u901a\u8fc7\u6307\u5b9atopics = {\"topic1\", \"topic2\"}<\/code>\u6210\u529f\u6d88\u8d39\u4e24\u6761\u6d88\u606f<\/p>\n
\u4e0b\u9762\u4e00\u8d77\u770b\u4e00\u4e0b\uff0c@KafkaListener<\/code>\u8fd8\u652f\u6301\u54ea\u4e9b\u53c2\u6570?<\/p>\n
@KafkaListener<\/code> \u6ce8\u89e3\u652f\u6301\u8bb8\u591a\u53c2\u6570\uff0c\u4ee5\u6ee1\u8db3\u4e0d\u540c\u7684\u4f7f\u7528\u573a\u666f\u3002\u4ee5\u4e0b\u662f\u5e38\u7528\u53c2\u6570\u7684\u5217\u8868\uff1a<\/p>\n
    \n
  • \n
    topics<\/code>\uff1a\u6307\u5b9a\u8981\u6d88\u8d39\u7684 Topic<\/code> \u7684\u540d\u79f0\uff0c\u53ef\u4ee5\u662f\u5b57\u7b26\u4e32\u6216\u5b57\u7b26\u4e32\u6570\u7ec4\u3002\u5fc5\u586b\u53c2\u6570\u3002<\/section>\n<\/li>\n
  • \n
    groupId<\/code>\uff1a\u6307\u5b9a\u6d88\u8d39\u8005\u7ec4 ID<\/code>\u3002\u6d88\u8d39\u8005\u7ec4\u662f\u4e00\u7ec4\u5171\u4eab\u76f8\u540c Topic<\/code> \u7684\u6d88\u8d39\u8005\u7684\u96c6\u5408\u3002\u9ed8\u8ba4\u503c\u4e3a \"\"<\/code>\uff0c\u8868\u793a\u4f7f\u7528\u9ed8\u8ba4\u7684\u6d88\u8d39\u8005\u7ec4 ID<\/code>\u3002<\/section>\n<\/li>\n
  • \n
    containerFactory<\/code>\uff1a\u6307\u5b9a\u8981\u4f7f\u7528\u7684 KafkaListenerContainerFactory<\/code> \u5b9e\u4f8b\u7684\u540d\u79f0\u3002\u5982\u679c\u6ca1\u6709\u6307\u5b9a\uff0c\u5c06\u4f7f\u7528\u9ed8\u8ba4\u7684 KafkaListenerContainerFactory<\/code> \u5b9e\u4f8b\u3002<\/section>\n<\/li>\n
  • \n
    concurrency<\/code>\uff1a\u6307\u5b9a\u8981\u521b\u5efa\u7684\u5e76\u53d1\u6d88\u8d39\u8005\u7684\u6570\u91cf\u3002\u9ed8\u8ba4\u503c\u4e3a 1<\/code>\u3002<\/section>\n<\/li>\n
  • \n
    autoStartup<\/code>\uff1a\u6307\u5b9a\u5bb9\u5668\u662f\u5426\u5728\u5e94\u7528\u542f\u52a8\u65f6\u81ea\u52a8\u542f\u52a8\u3002\u9ed8\u8ba4\u503c\u4e3a true<\/code>\u3002<\/section>\n<\/li>\n
  • \n
    id<\/code>\uff1a\u6307\u5b9a\u76d1\u542c\u5668\u7684\u552f\u4e00\u6807\u8bc6\u7b26\u3002\u9ed8\u8ba4\u503c\u4e3a \"\"<\/code>\u3002<\/section>\n<\/li>\n
  • \n
    errorHandler<\/code>\uff1a\u6307\u5b9a\u5728\u5904\u7406\u6d88\u606f\u65f6\u51fa\u73b0\u5f02\u5e38\u65f6\u8981\u4f7f\u7528\u7684 ErrorHandler<\/code> \u5b9e\u4f8b\u3002<\/section>\n<\/li>\n
  • \n
    properties<\/code>\uff1a\u6307\u5b9a\u4f20\u9012\u7ed9\u6d88\u8d39\u8005\u5de5\u5382\u7684 Kafka<\/code> \u6d88\u8d39\u8005\u914d\u7f6e\u5c5e\u6027\u7684 Map<\/code>\u3002<\/section>\n<\/li>\n
  • \n
    partitionOffsets<\/code>: \u662f\u4e00\u4e2a Map<\/code> \u7c7b\u578b\u7684\u53c2\u6570\uff0c\u8be5\u53c2\u6570\u7528\u4e8e\u6307\u5b9a\u8981\u4ece Topic<\/code> \u7684\u6bcf\u4e2a\u5206\u533a<\/code>\u7684\u54ea\u4e2a\u504f\u79fb\u91cf<\/code>\u5f00\u59cb\u6d88\u8d39\u6d88\u606f<\/section>\n<\/li>\n<\/ul>\n
    \u6709\u51e0\u4e2a\u53c2\u6570\u5f88\u597d\u7406\u89e3\uff0c\u6ca1\u5565\u597d\u8bb2\u7684\uff0c\u6211\u4eec\u4e3b\u8981\u770b\u4e00\u4e0bcontainerFactory<\/code>,errorHandler<\/code>, partitionOffsets<\/code><\/p>\n
    containerFactory<\/span><\/h3>\n
    \u524d\u9762\u6211\u4eec\u4f7f\u7528\u7684\u90fd\u662f\u9ed8\u8ba4<\/code>\u7684\u6d88\u606f\u76d1\u542c\u5668<\/code>,\u5728 Spring Kafka<\/code> \u4e2d\uff0cKafka \u6d88\u8d39\u8005<\/code>\u53ef\u4ee5\u4f7f\u7528\u4e0d\u540c\u7684\u6d88\u606f\u76d1\u542c\u5668\u5bb9\u5668<\/code>\uff0c\u4f8b\u5982 ConcurrentKafkaListenerContainerFactory<\/code>\u3001KafkaMessageListenerContainer<\/code> \u7b49\u3002\u6bcf\u4e2a\u5bb9\u5668\u90fd\u63d0\u4f9b\u4e86\u4e0d\u540c\u7684\u529f\u80fd\u548c\u914d\u7f6e\u9009\u9879\uff0c\u53ef\u4ee5\u6839\u636e\u5b9e\u9645\u9700\u6c42\u8fdb\u884c\u9009\u62e9\u548c\u914d\u7f6e\u3002 \u5982\u679c\u4f60\u9700\u8981\u81ea\u5b9a\u4e49 Kafka \u6d88\u8d39\u8005<\/code>\u7684\u914d\u7f6e\u9009\u9879\uff0c\u53ef\u4ee5\u901a\u8fc7\u5728 Spring Boot<\/code> \u914d\u7f6e\u6587\u4ef6\u4e2d\u8bbe\u7f6e\u5c5e\u6027\u6765\u5b9e\u73b0\u3002\u53e6\u5916\u4e00\u79cd\u65b9\u6cd5\u662f\u901a\u8fc7\u521b\u5efa KafkaListenerContainerFactory bean<\/code> \u5e76\u914d\u7f6e\u5176\u5c5e\u6027\u6765\u5b9e\u73b0\u3002<\/p>\n
    \u4e0b\u9762\u901a\u8fc7\u4e00\u4e2a\u4f8b\u5b50\u6765\u4f53\u9a8c\u4e00\u4e0bConcurrentKafkaListenerContainerFactory<\/code>~<\/p>\n
    @Configuration<\/span>\n@EnableKafka<\/span>\npublic<\/span>\u00a0class<\/span>\u00a0KafkaConfig<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\n\u00a0\u00a0\u00a0\u00a0@Bean<\/span>\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0ConcurrentKafkaListenerContainerFactory<String,\u00a0String>\u00a0kafkaListenerContainerFactory<\/span>()<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0ConcurrentKafkaListenerContainerFactory<String,\u00a0String>\u00a0factory\u00a0=\u00a0new<\/span>\u00a0ConcurrentKafkaListenerContainerFactory<>();\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0factory.setConsumerFactory(consumerFactory());\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0factory.setConcurrency(3<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0factory.getContainerProperties().setPollTimeout(3000<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/\u4e5f\u53ef\u4ee5\u8bbe\u7f6e\u4e3a\u6279\u91cf\u6d88\u8d39\uff0c\u6bcf\u4e2a\u6279\u6b21\u6570\u91cf\u5728Kafka\u914d\u7f6e\u53c2\u6570\u4e2d\u8bbe\u7f6eConsumerConfig.MAX_POLL_RECORDS_CONFIG<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/factory.setBatchListener(true);<\/span>\n\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0factory;\n\u00a0\u00a0\u00a0\u00a0}\n\u00a0\u00a0\u00a0\u00a0\n\u00a0\u00a0\u00a0\u00a0@Bean<\/span>\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0ConsumerFactory<String,\u00a0String>\u00a0consumerFactory<\/span>()<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0new<\/span>\u00a0DefaultKafkaConsumerFactory<>(consumerConfigs());\n\u00a0\u00a0\u00a0\u00a0}\n\u00a0\u00a0\u00a0\u00a0\n\u00a0\u00a0\u00a0\u00a0@Bean<\/span>\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0Map<String,\u00a0Object>\u00a0consumerConfigs<\/span>()<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0Map<String,\u00a0Object>\u00a0props\u00a0=\u00a0new<\/span>\u00a0HashMap<>();\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,\u00a0\"localhost:9092\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0props.put(ConsumerConfig.GROUP_ID_CONFIG,\u00a0\"my-group\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,\u00a0\"earliest\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,\u00a0StringDeserializer.class<\/span>)<\/span>;\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,\u00a0StringDeserializer.class<\/span>)<\/span>;\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0props;\n\u00a0\u00a0\u00a0\u00a0}\n}\n<\/code><\/pre>\n
    \u5728\u4e0a\u9762\u7684\u793a\u4f8b\u4e2d\uff0c\u6211\u4eec\u521b\u5efa\u4e86\u4e00\u4e2a ConcurrentKafkaListenerContainerFactory<\/code> \u7684 bean<\/code>\uff0c\u5e76\u914d\u7f6e\u4e86\u4e00\u4e9b\u5c5e\u6027\uff0c\u4f8b\u5982\uff1a<\/p>\n
      \n
    • \n
      \u4f7f\u7528 DefaultKafkaConsumerFactory<\/code> \u4f5c\u4e3a\u6d88\u8d39\u8005\u5de5\u5382<\/code>\uff1b<\/section>\n<\/li>\n
    • \n
      \u8bbe\u7f6e\u5e76\u53d1\u6d88\u8d39\u8005\u6570\u91cf\u4e3a 3<\/code>\uff1b<\/section>\n<\/li>\n
    • \n
      \u8bbe\u7f6e\u8f6e\u8be2\u8d85\u65f6\u65f6\u95f4\u4e3a 3000 \u6beb\u79d2<\/code>\u3002<\/section>\n<\/li>\n<\/ul>\n
      \u90a3\u4e48\u5982\u4f55\u4f7f\u7528\u5462?\u5f88\u7b80\u5355\uff0c\u53ea\u9700\u4e4b\u6307\u5b9acontainerFactory<\/code>\u5c31\u597d\u4e86<\/p>\n
      @Transactional<\/span>(rollbackFor\u00a0=\u00a0Exception.class<\/span>)\n@GetMapping<\/span>(\"\/hello1<\/span>\")\npublic<\/span>\u00a0String<\/span>\u00a0hello1<\/span>()\u00a0<\/span>{\n\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u53d1\u9001\u6d88\u606f<\/span>\n\u00a0\u00a0\u00a0\u00a0kafkaTemplate.send(\"topic3\"<\/span>,\u00a0\"3\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0\"hello1\"<\/span>;\n}\n\n@KafkaListener<\/span>(topics\u00a0=\u00a0\"topic3\"<\/span>,\u00a0containerFactory\u00a0=\u00a0\"kafkaListenerContainerFactory\"<\/span>)\npublic<\/span>\u00a0void<\/span>\u00a0processMessage<\/span>(String\u00a0message)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u5904\u7406\u6d88\u606f<\/span>\n\u00a0\u00a0\u00a0\u00a0log.info(\"hello1\u00a0>>>>>\u00a0{}\"<\/span>,\u00a0message);\n}\n<\/code><\/pre>\n
      \u6211\u4eec\u53ef\u4ee5\u901a\u8fc7containerFactory <\/code>\u5e94\u7528\u4e8e\u4e0d\u540c\u7684\u6d88\u8d39\u8005\u65b9\u6cd5\u548c\u4e3b\u9898\uff0c\u4ee5\u6ee1\u8db3\u4e0d\u540c\u7684\u9700\u6c42<\/p>\n
      errorHandler<\/span><\/h3>\n
      \u5728 \u00b7Spring Kafka\u00b7 \u4e2d\uff0c\u00b7ErrorHandler\u00b7 \u63a5\u53e3\u5b9a\u4e49\u4e86\u5728\u5904\u7406 Kafka<\/code> \u6d88\u606f\u65f6\u53d1\u751f\u9519\u8bef\u65f6\u5982\u4f55\u5904\u7406\u5f02\u5e38\u3002\u5982\u679c\u4e0d\u914d\u7f6e ErrorHandler<\/code>\uff0c\u5219\u9ed8\u8ba4\u4f7f\u7528 LoggingErrorHandler<\/code> \u5c06\u5f02\u5e38\u8bb0\u5f55\u5230\u65e5\u5fd7\u4e2d\u3002<\/p>\n
      \u5982\u679c\u4f60\u9700\u8981\u81ea\u5b9a\u4e49\u5f02\u5e38\u5904\u7406\u903b\u8f91\uff0c\u53ef\u4ee5\u901a\u8fc7\u5b9e\u73b0 ErrorHandler<\/code> \u63a5\u53e3\u5e76\u914d\u7f6e\u5176 bean<\/code> \u6765\u5b9e\u73b0\u3002<\/p>\n
      \u4e0b\u9762\u901a\u8fc7\u4e00\u4e2a\u4f8b\u5b50\u6765\u4f53\u9a8c\u4e00\u4e0b:<\/p>\n
      @Slf<\/span>4j\n@Component<\/span>\npublic<\/span>\u00a0class<\/span>\u00a0MyErrorHandler<\/span>\u00a0implements<\/span>\u00a0KafkaListenerErrorHandler<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0@Override<\/span>\n\u00a0\u00a0\u00a0\u00a0public<\/span>\u00a0Object\u00a0handleError<\/span>(Message<?>\u00a0message,\u00a0ListenerExecutionFailedException\u00a0e)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u5904\u7406\u5f02\u5e38\uff0c\u4f8b\u5982\u6253\u5370\u9519\u8bef\u65e5\u5fd7\u3001\u53d1\u9001\u9519\u8bef\u6d88\u606f\u7b49,\u81ea\u5b9a\u4e49\u903b\u8f91\u5904\u7406<\/span>\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0log.error(\"Error\u00a0occurred\u00a0while\u00a0processing\u00a0message:\u00a0\u00a0{}\"<\/span>,\u00a0e.getMessage());\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0null<\/span>;\n\u00a0\u00a0\u00a0\u00a0}\n}\n<\/code><\/pre>\n
      \u4f7f\u7528\u9519\u8bef\u5904\u7406\u5668:<\/p>\n
      @Transactional<\/span>(rollbackFor\u00a0=\u00a0Exception.class<\/span>)\n@GetMapping<\/span>(\"\/hello2<\/span>\")\npublic<\/span>\u00a0String<\/span>\u00a0hello2<\/span>()\u00a0<\/span>{\n\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u53d1\u9001\u6d88\u606f<\/span>\n\u00a0\u00a0\u00a0\u00a0kafkaTemplate.send(\"topic4\"<\/span>,\u00a0\"4\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0return<\/span>\u00a0\"hello2\"<\/span>;\n}\n\n@KafkaListener<\/span>(id\u00a0=\u00a0\"topic4Group\"<\/span>,\u00a0topics\u00a0=\u00a0\"topic4\"<\/span>,\u00a0errorHandler\u00a0=\u00a0\"myErrorHandler\"<\/span>)\npublic<\/span>\u00a0void<\/span>\u00a0processMessage1<\/span>(String\u00a0message)<\/span><\/span>{\n\u00a0\u00a0\u00a0\u00a0\/\/\u00a0\u5904\u7406\u6d88\u606f<\/span>\n\u00a0\u00a0\u00a0\u00a0if<\/span>(true<\/span>)\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0throw<\/span>\u00a0new<\/span>\u00a0RuntimeException(\"\u6d88\u606f\u5904\u7406\u5f02\u5e38\"<\/span>);\n\u00a0\u00a0\u00a0\u00a0log.info(\"hello2\u00a0>>>>>\u00a0{}\"<\/span>,\u00a0message);\n}\n\n<\/code><\/pre>\n
      \u89c2\u5bdf\u63a7\u5236\u53f0\uff1a<\/p>\n
      Error\u00a0occurred\u00a0while<\/span>\u00a0processing\u00a0message:\u00a0\u00a0Listener\u00a0method\u00a0'public\u00a0void\u00a0com.kafka.study.controller.ListenerController.processMessage1(java.lang.String)'<\/span>\u00a0threw\u00a0exception;\u00a0nested\u00a0exception\u00a0is\u00a0java.lang.RuntimeException:\u00a0\u6d88\u606f\u5904\u7406\u5f02\u5e38
      <\/code><\/pre>\n
      \u00a0<\/div>\n<\/section>\n
      \u00a0<\/div>\n
      \n
      \u4e8b\u4f8b\u4ee3\u7801<\/strong><\/div>\n<\/blockquote>\n
      \u00a0<\/div>\n
      \n
      config<\/strong><\/div>\n<\/blockquote>\n
      \n

      @Configuration<\/code>
      @Component<\/code>
      @EnableKafka<\/code>
      public class KafkaConfig {<\/code>

      @Value(\"${spring.kafka.bootstrap-servers}\")<\/code>
      private String bootstrapServers;<\/code>

      \/\/ @Value(\"${spring.kafka.client-id}\")<\/code>
      \/\/ private String clientId;<\/code>
      \/\/<\/code>
      \/\/ @Value(\"${spring.kafka.consumer.enable-auto-commit}\")<\/code>
      \/\/ private Boolean autoCommit;<\/code>
      \/\/<\/code>
      \/\/ @Value(\"${spring.kafka.consumer.auto-commit-interval}\")<\/code>
      \/\/ private Integer autoCommitInterval;<\/code>

      @Value(\"${spring.kafka.consumer.group-id}\")<\/code>
      private String groupId;<\/code>

      @Value(\"${spring.kafka.consumer.max-poll-records}\")<\/code>
      private Integer maxPollRecords;<\/code>

      @Value(\"${spring.kafka.consumer.auto-offset-reset}\")<\/code>
      private String autoOffsetReset;<\/code>

      @Value(\"${spring.kafka.consumer.key-deserializer}\")<\/code>
      private String keyDeserializer;<\/code>

      @Value(\"${spring.kafka.consumer.value-deserializer}\")<\/code>
      private String valueDeserializer;<\/code>

      @Value(\"${spring.kafka.producer.retries}\")<\/code>
      private Integer retries;<\/code>

      @Value(\"${spring.kafka.producer.key-serializer}\")<\/code>
      private String keySerializer;<\/code>

      @Value(\"${spring.kafka.producer.value-serializer}\")<\/code>
      private String valueSerializer;<\/code>

      @Value(\"${spring.kafka.producer.batch-size}\")<\/code>
      private Integer batchSize;<\/code>

      @Value(\"${spring.kafka.producer.buffer-memory}\")<\/code>
      private Integer bufferMemory;<\/code>

      \/\/ @Value(\"${spring.kafka.producer.compression-type}\")<\/code>
      \/\/ private String compressionType;<\/code>

      \/**<\/code>
      * \u751f\u4ea7\u914d\u7f6e\u4fe1\u606f<\/code>
      *<\/code>
      * @return<\/code>
      *\/<\/code>
      @Bean<\/code>
      public Map<String, Object> producerConfig() {<\/code>
      Map<String, Object> props = new HashMap<String, Object>();<\/code>
      props.put(ProducerConfig.ACKS_CONFIG, \"1\");<\/code>
      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);<\/code>
      props.put(ProducerConfig.RETRIES_CONFIG, retries);<\/code>
      props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);<\/code>
      props.put(ProducerConfig.LINGER_MS_CONFIG, 1);<\/code>
      props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);<\/code>
      props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);<\/code>
      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);<\/code>
      return props;<\/code>
      }<\/code>

      \/**<\/code>
      * \u751f\u4ea7\u8005\u5de5\u5382<\/code>
      *<\/code>
      * @return<\/code>
      *\/<\/code>
      @Bean<\/code>
      public ProducerFactory<String, String> producerFactory() {<\/code>
      return new DefaultKafkaProducerFactory<>(producerConfig());<\/code>
      }<\/code>

      \/**<\/code>
      * \u751f\u4ea7\u8005\u6a21\u677f<\/code>
      *<\/code>
      * @return<\/code>
      *\/<\/code>
      @Bean<\/code>
      public KafkaTemplate<String, String> kafkaTemplate() {<\/code>
      return new KafkaTemplate<>(producerFactory());<\/code>
      }<\/code>

      \/**<\/code>
      * \u6d88\u8d39\u8005\u914d\u7f6e\u4fe1\u606f<\/code>
      *\/<\/code>
      @Bean<\/code>
      public Map<String, Object> consumerConfig() {<\/code>
      Map<String, Object> props = new HashMap<String, Object>();<\/code>
      props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);<\/code>
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);<\/code>
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);<\/code>
      props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);<\/code>
      props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);<\/code>
      props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000);<\/code>
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);<\/code>
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);<\/code>
      return props;<\/code>
      }<\/code>

      \/**<\/code>
      * \u6d88\u8d39\u8005\u6279\u91cf\u5de5\u7a0b<\/code>
      *\/<\/code>
      @Bean<\/code>
      public KafkaListenerContainerFactory<?> batchFactor() {<\/code>
      ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<Integer, String>();<\/code>
      factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfig()));<\/code>
      factory.setBatchListener(true);<\/code>
      return factory;<\/code>
      }<\/code>

      }<\/code><\/div>\n<\/div>\n
      \u00a0<\/div>\n
      \u6d88\u8d39\u8005\u793a\u4f8b\u4ee3\u7801<\/strong><\/blockquote>\n
      \u00a0<\/div>\n
      \n
      public class FlowConsumer {<\/code>

      private static Logger log = LoggerFactory.getLogger(FlowConsumer.class);<\/code>

      private static final String topicName = AtlasMQConstants.SendDifferentTopic.TOPIC;<\/code>

      @Autowired<\/code>
      private ShopInspectionGradeAbarbeitungService shopInspectionGradeAbarbeitungService;<\/code>

      @Autowired<\/code>
      private BaseWeixinService baseWeixinService;<\/code>

      @KafkaListener(topics = {topicName})<\/code>
      public void receiveMessageFormKafka(List<ConsumerRecord<?, ?>> recordArr, Acknowledgment ack) throws Exception {<\/code>
      try {<\/code>
      log.info(\"\u6d88\u8d39\u8005\u7b49\u5f852S\u949f\u6d88\u8d39start\");<\/code>
      Thread.sleep(2000);<\/code>
      log.info(\"\u6d88\u8d39\u8005\u7b49\u5f852S\u949f\u6d88\u8d39end\");<\/code>
      } catch (InterruptedException e) {<\/code>
      log.error(\"\u6d88\u8d39\u8005\u7b49\u5f85\u6d88\u8d39\u5f02\u5e38:[{}]\", e.getMessage());<\/code>
      }<\/code>
      Map<String, BaseWeixin> weiXinApi = baseWeixinService.queryWeiXinApi();<\/code>
      for (ConsumerRecord<?, ?> record : recordArr) {<\/code>
      log.info(\"\u76d1\u542c\u6d88\u606f:topic={},partition={},offset={},value={}\", record.topic(), record.partition(), record.offset(), record.value());<\/code>
      Optional<?> optional = Optional.ofNullable(record.value());<\/code>
      if (optional.isPresent()) {<\/code>
      FlowDifferentMessageVi messageVi = JsonUtils.jsonTo(record.value().toString(), FlowDifferentMessageVi.class);<\/code>
      log.info(\"\u552f\u4e00\u7f16\u7801\uff1aUnique coding={},message={}\", messageVi.getMessageId(), messageVi.toString());<\/code>
      shopInspectionGradeAbarbeitungService.sendDifferentMessageByCondition(messageVi, weiXinApi);<\/code>
      }<\/code>
      ack.acknowledge();<\/code>
      }<\/code>
      }<\/code>

      }<\/code><\/div>\n<\/section>\n
      \u00a0<\/div>\n
      \n
      \u751f\u4ea7\u8005\u793a\u4f8b\u4ee3\u7801<\/strong><\/div>\n<\/blockquote>\n
      \u00a0<\/div>\n
      \n
      public class FlowProducerServer {<\/code>

      private static Logger log = LoggerFactory.getLogger(FlowProducerServer.class);<\/code>

      private final KafkaConfig kafkaConfig;<\/code>

      private final UniqueIDMethodUtils uniqueIDMethodUtils;<\/code>

      private final AuthService authService;<\/code>

      private final String topicName = AtlasMQConstants.SendDifferentTopic.TOPIC;<\/code>

      private final Producer<String, Object> producer;<\/code>

      @Autowired<\/code>
      public FlowProducerServer(KafkaConfig kafkaConfig, UniqueIDMethodUtils uniqueIDMethodUtils, AuthService authService) {<\/code>
      this.kafkaConfig = kafkaConfig;<\/code>
      this.uniqueIDMethodUtils = uniqueIDMethodUtils;<\/code>
      this.authService = authService;<\/code>
      this.producer = new KafkaProducer<>(kafkaConfig.producerConfig());<\/code>
      }<\/code>

      public Result sendDifferentMessageByCondition(FlowSendDifferentMessVi vi) {<\/code>
      Auth auth;<\/code>
      try {<\/code>
      auth = authService.getAuthCodeVerifyCode(vi.getEncryptAuth());<\/code>
      } catch (Exception e) {<\/code>
      throw new BizException(\"\u83b7\u53d6\u6388\u6743\u7801\u9a8c\u8bc1\u5f02\u5e38\uff1a\" + e.getMessage());<\/code>
      }<\/code>
      if (StringUtils.checkNull(auth)) {<\/code>
      throw new BizException(ResultCode.DECRYPT_VERIFY_CODE_ERROR.getMsg());<\/code>
      }<\/code>
      if (StringUtils.checkNull(vi.getFlowName()) && StringUtils.checkNull(vi.getFlowId())) {<\/code>
      throw new BizException(ResultCode.FLOW_ID_AND_FLOW_NAME_IS_NULL_ERROR);<\/code>
      }<\/code>
      FlowDifferentMessageVi messageVi = new FlowDifferentMessageVi();<\/code>
      messageVi.setMessageId(uniqueIDMethodUtils.generateUniqueID());<\/code>
      messageVi.setSendNum(0);<\/code>
      messageVi.setServerFlag(topicName);<\/code>
      messageVi.setFlowSendDifferentMessVi(vi);<\/code>
      String toJson = JsonUtils.toJson(messageVi);<\/code>
      producer.send(new ProducerRecord<String, Object>(topicName, toJson), (recordMetadata, e) -> {<\/code>
      if (!StringUtils.checkNull(recordMetadata)) {<\/code>
      log.info(\"\u751f\u4ea7\u6d88\u606f\u552f\u4e00\u6807\u8bc6[{}],\u53d1\u9001\u6b21\u6570[{}],checksum\uff1a[{}],offset:[{}],partition:[{}],topic:[{}]\",<\/code>
      messageVi.getMessageId(), messageVi.getSendNum(), recordMetadata.checksum(), recordMetadata.offset(), recordMetadata.partition(), recordMetadata.topic());<\/code>
      }<\/code>
      if (!StringUtils.checkNull(e)) {<\/code>
      log.error(\"\u751f\u4ea7\u6d88\u606f\u53d1\u9001\u5931\u8d25\uff0c\u5f02\u5e38\u4fe1\u606f:[{}]\", e.getMessage());<\/code>
      throw new BizException(\"\u6d88\u606f\u5f02\u5e38\u4fe1\u606f:\" + e.getMessage());<\/code>
      }<\/code>
      });<\/code>
      return Result.ok();<\/code>
      }<\/code>

      @PreDestroy<\/code>
      public void closeProducer() {<\/code>
      producer.close();<\/code>
      }<\/code>

      }<\/code><\/div>\n<\/div>\n
      \n
      \u7ed3\u675f\u8bed<\/span><\/h2>\n
      \u672c\u8282\u8fd8\u9057\u7559\u4e00\u4e2apartitionOffsets<\/code>\u8fd9\u4e2a\u6211\u4eec\u653e\u5230\u4e0b\u8282\u7ed9\u5927\u5bb6\u8bb2\uff0c\u6d89\u53ca\u5230\u5206\u533a partition<\/code>\u548c\u504f\u79fb\u91cf offset<\/code>\u7684\u6982\u5ff5\uff0c\u4e0d\u662f\u5f88\u597d\u7406\u89e3\uff0c\u4e0b\u8282\u7ed9\u5927\u5bb6\u597d\u597d\u7406\u4e00\u4e0b\u8fd9\u4e2a\u6982\u5ff5~<\/p>\n<\/section>\n
      \u00a0<\/p>\n<\/div><\/div>\n","raw":"","protected":false},"excerpt":{"rendered":"
      \u4e00\u8d77\u6765\u5b66kafka\u4e4b\u6574\u5408SpringBoot\u6df1\u5165\u4f7f\u7528(\u4e00) \u524d\u8a00 \u76ee\u524d\u6b63\u5728\u51fa\u4e00\u4e2aKafka\u4e13\u9898\u7cfb\u5217\u6559\u7a0b, \u7bc7\u5e45\u4f1a\u8f83\u591a, \u559c\u6b22\u7684\u8bdd\uff0c …<\/p>\n","protected":false},"author":1,"featured_media":218,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"emotion":"","emotion_color":"","title_style":"","license":"","footnotes":""},"categories":[16,7],"tags":[],"class_list":["post-217","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-kafka","category-spring-boot"],"post_thumbnail_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230525135201806-150x150.jpg","content_first_image":null,"post_medium_image_300":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230525135201806-300x128.jpg","post_thumbnail_image_624":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230525135201806-1024x437.jpg","post_frist_image":null,"post_medium_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230525135201806-300x128.jpg","post_large_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230525135201806-1024x437.jpg","post_full_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230525135201806.jpg","post_all_images":[],"videoAdId":"","listAd":"0","listAdId":"","listAdEvery":6,"total_comments":0,"category_name":"Kafka","post_date":"2023-05-24","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":272,"next_post_id":221,"next_post_title":"\u4e00\u8d77\u6765\u5b66kafka\u4e4b\u6574\u5408SpringBoot\u57fa\u672c\u4f7f\u7528","previous_post_id":373,"previous_post_title":"@Transactional\u4e8b\u52a1\u6ce8\u89e3","_links":{"self":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/posts\/217","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=217"}],"version-history":[{"count":0,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/posts\/217\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/media\/218"}],"wp:attachment":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/media?parent=217"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/categories?post=217"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/tags?post=217"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}