import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
/**
 * reads the config info from yml file
 * 
 * @return
 */
@Bean
public Map<String, Object> consumerConfigs() {
	Map<String, Object> propsMap = new HashMap<>();
	propsMap.put("bootstrap.servers",
			env.getProperty("kafka.broker"));
	propsMap.put("enable.auto.commit",
			env.getProperty("enable.auto.commit"));
	propsMap.put("auto.commit.interval.ms",
			env.getProperty("auto.commit.interval.ms"));
	propsMap.put("key.deserializer",
			StringDeserializer.class);
	propsMap.put("value.deserializer",
			JsonDeserializer.class);
	propsMap.put("group.id",
			env.getProperty("group.id"));
	propsMap.put("auto.offset.reset",
			env.getProperty("kafka.auto.offset.reset"));
	return propsMap;

}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    // allows a pool of processes to divide the work of consuming and processing records
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "bot");

    return props;
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
/**
 * Create the consumers by loading the config
 * 
 * @return
 */
@Bean
public ConsumerFactory<String, Student> consumerFactory() {

	return new DefaultKafkaConsumerFactory<>(consumerConfigs(),
			new StringDeserializer(), new JsonDeserializer<>(
					Student.class));
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
@Bean
public ConsumerFactory<String, Foo> consumerFactory() {
	Map<String, Object> props = new HashMap<>();
	props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configProperties.getBrokerAddress());
	props.put(ConsumerConfig.GROUP_ID_CONFIG, "s1pGroup");
	props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
	props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
	props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000);
	JsonDeserializer<Foo> jsonDeserializer = new JsonDeserializer<>(Foo.class);
	return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), jsonDeserializer);
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
@Bean
public ConsumerFactory<String, DefaultAvMessage> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(
            consumerConfigs(),
            new StringDeserializer(),
            new JsonDeserializer<>(DefaultAvMessage.class)
    );
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
private Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "client");

    return props;
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
@Bean
public ConsumerFactory<String, DefaultAvMessage> fileServerConsumerFactory() {
    return new DefaultKafkaConsumerFactory<>(
            consumerConfigs(),
            new StringDeserializer(),
            new JsonDeserializer<>(DefaultAvMessage.class)
    );
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
private Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "server");

    return props;
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
@Bean
public Map<String, Object> consumerConfigs() {
  Map<String, Object> props = new HashMap<>();
  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
  props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
  props.put(ConsumerConfig.GROUP_ID_CONFIG, "json");

  return props;
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
@Bean
public ConsumerFactory<String, LifecycleEvent> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), new JsonDeserializer<>(LifecycleEvent.class));
}
 

import org.springframework.kafka.support.serializer.JsonDeserializer; //导入依赖的package包/类
@Bean
public ConsumerFactory<String, Car> consumerFactory() {
  return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
      new JsonDeserializer<>(Car.class));
}