{"id":135542,"date":"2020-07-16T01:00:30","date_gmt":"2020-07-15T17:00:30","guid":{"rendered":"http:\/\/4563.org\/?p=135542"},"modified":"2020-07-16T01:00:30","modified_gmt":"2020-07-15T17:00:30","slug":"kafka-%e5%a4%9a%e5%ae%b9%e5%99%a8%e5%b7%a5%e5%8e%82%e5%8f%8d%e5%ba%8f%e5%88%97%e5%8c%96-kafkalistenercontainerfactory","status":"publish","type":"post","link":"http:\/\/4563.org\/?p=135542","title":{"rendered":"kafka \u591a\u5bb9\u5668\u5de5\u5382\u53cd\u5e8f\u5217\u5316 kafkaListenerContainerFactory"},"content":{"rendered":"
\n
\n
\n

kafka \u591a\u5bb9\u5668\u5de5\u5382\u53cd\u5e8f\u5217\u5316 kafkaListenerContainerFactory <\/h1>\n

<\/p>\n

\n
\u8cc7\u6df1\u5927\u4f6c : qw564518158 <\/span> <\/i> 13<\/span> <\/div>\n
<\/div>\n<\/p><\/div>\n<\/p><\/div>\n<\/p><\/div>\n
<\/p>\n

\u539f\u521b<\/p>\n

\u4e1a\u52a1\u9700\u8981\uff0c\u6279\u91cf\u6d88\u8d39\uff0c\u4f46\u662f\u53c8\u60f3\u6279\u91cf\u76f4\u63a5\u6309 List<ModelDTO> \u6a21\u5f0f\u76f4\u63a5\u62c9\u53d6\u6570\u636e\uff0c\u6279\u91cf\u5904\u7406\u3002<\/p>\n

private final String topic = “queue_notify”;<\/p>\n

@KafkaListener(topics = topic ,containerFactory = “kafkaLiveListenerContainerFactory”) public void listen(List pushLiveDTOS) { Long startTime = System.currentTimeMillis(); \/\/\u6279\u91cf\u4e2a\u63a8 pushService.notifyLiveGetui (pushLiveDTOS); Long endTime = System.currentTimeMillis(); }<\/pushlivedto><\/p>\n

private final String topic = “queue_push”; \/\/containerFactory \u5bb9\u5668\u5de5\u5382\u65b9\u6cd5 @KafkaListener(topics = topic,containerFactory = “kafkaListenerContainerFactory”) public void listen(List mallPushMongoDBS) { pushService.saveNoPush(mallPushMongoDBS); } \u5982\u4e0a\u4ee3\u7801<\/pushtestdto><\/p>\n

\u6279\u91cf\u62c9\u53d6\u4e0d\u540c topic \u83b7\u53d6\u5230 list \u7684 DTO\uff0c\u8fdb\u884c\u5904\u7406\u3002<\/p>\n

package com.dg.mall.push.config;<\/p>\n

import com.dg.mall.push.kafka.PushJsonDeserializer; import com.dg.mall.push.kafka.LiveJsonDeserializer; import com.dg.mall.push.kafka.listen.LiveNotifyListener; import com.dg.mall.push.kafka.listen.PushListener; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.EnableKafka; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;<\/p>\n

import java.util.HashMap; import java.util.Map;<\/p>\n

@Configuration @EnableKafka public class KafkaConsumerConfig {<\/p>\n

@Value(\"${spring.kafka.consumer.bootstrap-servers}\") private String servers; @Value(\"${spring.kafka.consumer.enable-auto-commit}\") private boolean enableAutoCommit; @Value(\"${spring.kafka.consumer.auto-commit-interval}\") private String autoCommitInterval; @Value(\"${spring.kafka.consumer.group-id}\") private String groupId; @Value(\"${spring.kafka.consumer.auto-offset-reset}\") private String autoOffsetReset; @Value(\"${spring.kafka.consumer.concurrency}\") private int concurrency; @Value(\"${spring.kafka.consumer.max-consumer-number}\") private Integer maxConsumerNumber; @Bean public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, byte[]>> kafkaListenerContainerFactory() {     ConcurrentKafkaListenerContainerFactory<String, byte[]> factory = new ConcurrentKafkaListenerContainerFactory<>();     factory.setConsumerFactory(consumerFactory());     \/\/\u5e76\u53d1\u6570\u91cf     factory.setConcurrency(concurrency);     \/\/\u6279\u91cf\u83b7\u53d6     factory.setBatchListener(true);     factory.getContainerProperties().setPollTimeout(1500);     return factory; }  public ConsumerFactory<String, byte[]> consumerFactory() {     return new DefaultKafkaConsumerFactory(consumerConfigs(),new StringDeserializer(),new pushJsonDeserializer()); }   public Map<String, Object> consumerConfigs() {     Map<String, Object> propsMap = new HashMap<>();     propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);     propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);     propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);     propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);     \/\/   \/\/\u8fd9\u91cc\u662f\u53cd\u5e8f\u5217\u5316\u7684 pushJsonDeserializer      propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, pushJsonDeserializer.class);     propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);     propsMap.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,40000);     propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);     propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);     \/\/\u6700\u591a\u6279\u91cf\u83b7\u53d6 100 \u4e2a     propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,maxConsumerNumber);     return propsMap; }  @Bean public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, byte[]>> kafkaLiveListenerContainerFactory() {     ConcurrentKafkaListenerContainerFactory<String, byte[]> factory = new ConcurrentKafkaListenerContainerFactory<>();     factory.setConsumerFactory(liveConsumerFactory());     \/\/\u5e76\u53d1\u6570\u91cf     factory.setConcurrency(concurrency);     \/\/\u6279\u91cf\u83b7\u53d6     factory.setBatchListener(true);     factory.getContainerProperties().setPollTimeout(1500);     return factory; }  public ConsumerFactory<String, byte[]> liveConsumerFactory() {     return new DefaultKafkaConsumerFactory(liveConsumerConfigs(),new StringDeserializer(),new PushLiveJsonDeserializer()); }   public Map<String, Object> liveConsumerConfigs() {     Map<String, Object> propsMap = new HashMap<>();     propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);     propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);     propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);     propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);     \/\/\u8fd9\u91cc\u662f\u53cd\u5e8f\u5217\u5316\u7684 liveJsonDeserializer     propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, liveJsonDeserializer.class);     propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);     propsMap.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,40000);     propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);     propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);     \/\/\u6700\u591a\u6279\u91cf\u83b7\u53d6 100 \u4e2a     propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,maxConsumerNumber);     return propsMap; }  @Bean public PushListener listener() {     return new PushListener(); }   @Bean public NotifyListener livelistener() {     return new NotifyListener(); } <\/code><\/pre>\n
} \/\/\u53cd\u5e8f\u5217\u5316 package com.dg.mall.push.kafka;<\/p>\n
import com.dg.mall.push.model.message.PushLiveDTO; import com.gexin.fastjson.JSON; import org.apache.kafka.common.serialization.Deserializer;<\/p>\n
import java.util.Map;<\/p>\n
public class LiveJsonDeserializer implements Deserializer<PushLiveDTO> {<\/p>\n
@Override public void configure(Map<String, ?> configs, boolean isKey) {  }  @Override public PushLiveDTO deserialize(String topic, byte[] data) {     return JSON.parseObject(data, PushLiveDTO.class); }  @Override public void close() {  } <\/code><\/pre>\n
} \/\/\u53cd\u5e8f\u5217\u5316 package com.dg.mall.push.kafka;<\/p>\n
import com.dg.mall.push.model.message.MallPushMongoDB; import com.gexin.fastjson.JSON; import org.apache.kafka.common.serialization.Serializer;<\/p>\n
import java.util.Map;<\/p>\n
public class MallPushJsonSerializer implements Serializer<MallPushMongoDB> {<\/p>\n
@Override public void configure(Map<String, ?> configs, boolean isKey) {  }  @Override public byte[] serialize(String topic, MallPushMongoDB data) {     return JSON.toJSONBytes(data); }  @Override public void close() {  } <\/code><\/pre>\n
}<\/p>\n
\u603b\u7ed3 \u6d88\u8d39<\/p>\n
@KafkaListener(topics = topic ,containerFactory = “kafkaLiveListenerContainerFactory”) \u76d1\u542c\u6d88\u8d39\u4e00\u5b9a\u8981\u52a0 containerFactory \u5bf9\u5e94 \u5bb9\u5668\u5de5\u5382\u7c7b \uff0c\u5bb9\u5668\u5de5\u5382\u7c7b\u91cc\u9762\u6709\u4e2a\u53cd\u5e8f\u5217\u5316\uff0c\u9700\u8981\u66ff\u6362\uff0c\u4e00\u822c\u90fd\u662f String \u53cd\u5e8f\u5217\u5316\uff0c\u8fd9\u91cc\u6211\u4eec\u66ff\u6362\u6210\u6211\u4eec\u81ea\u5df1\u521b\u5efa\u7684 DTO\uff0c\u5728\u6765\u8fdb\u884c\u53cd\u5e8f\u5217\u5316 LiveJsonDeserializer \u3002<\/p>\n
\u6700\u540e\u7684\u6700\u540e\u76d1\u542c\u6d88\u8d39\u6279\u91cf\u83b7\u53d6\u7684\u65f6\u5019<\/p>\n
\u6570\u636e\u5c31\u8fd9\u6837\u5168\u90e8\u83b7\u53d6\u5230\u4e86\u3002<\/p>\n
end!<\/p>\n<\/p><\/div>\n
 \u5927\u4f6c\u6709\u8a71\u8aaa<\/b> (0<\/span>)        <\/div>\n
 <\/div>\n<\/p><\/div>\n<\/p><\/div>\n
    \n
  • \n","protected":false},"excerpt":{"rendered":"
    kafka \u591a\u5bb9\u5668\u5de5\u5382\u53cd\u5e8f\u5217\u5316 k…<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[],"tags":[],"_links":{"self":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts\/135542"}],"collection":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=135542"}],"version-history":[{"count":0,"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts\/135542\/revisions"}],"wp:attachment":[{"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=135542"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=135542"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=135542"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}