要创建一个KafkaSpout对象，必须要传入一个SpoutConfig对象，KafkaSpout的构造函数定义如下：

public KafkaSpout(SpoutConfig spoutConf) {_spoutConfig = spoutConf;
}

SpoutConfig继承KafkaConfig，并实现Serializable，由于在KafkaConfig中所有的属性字段都是public的因此在SpoutConfig中可以直接引用，SpoutConfig类的定义如下其中核心字段添加了注释。

public class SpoutConfig extends KafkaConfig implements Serializable {//记录zookeeper的地址列表public List<String> zkServers = null;//zookeeper端口号public Integer zkPort = null;//该参数是Consumer消费的meta信息，保存在zk的路径，自己指定public String zkRoot = null;//唯一idpublic String id = null;//向zookeeper记录offset的间隔时间public long stateUpdateIntervalMs = 2000;public SpoutConfig(BrokerHosts hosts, String topic, String zkRoot, String id) {super(hosts, topic);this.zkRoot = zkRoot;this.id = id;}
}

KafkaConfig的定义如下：

public class KafkaConfig implements Serializable {//用以获取Kafka broker和partition的信息public final BrokerHosts hosts;//读消息的topicpublic final String topic;//消息者所用的client idpublic final String clientId;//每次从kafka读取的byte数public int fetchSizeBytes = 1024 * 1024;//Consumer连接kafka server超时时间public int socketTimeoutMs = 10000;//当服务器没有新消息时，消费者会等待这些时间public int fetchMaxWait = 10000;//consumer段的缓冲区大小public int bufferSizeBytes = 1024 * 1024;//数据发送的序列化和反序列化定义的Schemepublic MultiScheme scheme = new RawMultiScheme();//是否强制从kafka中offset最小开始读数据，和startOffsetTime，一起用，默认情况下，为false，一旦startOffsetTime被设置，就要置为true  public boolean forceFromStart = false;//从何offset时间开始读，默认为最旧的offsetpublic long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();//每次kafka会读取一批offset存放在list中，当zk offset比当前本地保存的commitOffse相减大于这个值时，重新设置commitOffset为当前zk offsetpublic long maxOffsetBehind = Long.MAX_VALUE;//如果所请求的offset对应的消息在Kafka中不存在，是否使用startOffsetTimepublic boolean useStartOffsetTimeIfOffsetOutOfRange = true;//多长时间统计一次metricspublic int metricsTimeBucketSizeInSecs = 60;public KafkaConfig(BrokerHosts hosts, String topic) {this(hosts, topic, kafka.api.OffsetRequest.DefaultClientId());}public KafkaConfig(BrokerHosts hosts, String topic, String clientId) {this.hosts = hosts;this.topic = topic;this.clientId = clientId;}
}

影响初始读取进度的配置

在一个topology上线后，KafkaSpout从何处开始读消息呢，有几个配置影响读消息的位置，先罗列如下：

• SpoutConfig中的id字段:如果想让一个topology从另一个topology之前的处理进度处读取数据，他们需要有相同的id
• KafkaConfig的forceFromStart字段：如果该字段为true，在topology上线后会忽略之前id相同的topology的进度，重新从最早的消息处读取
• KafkaConfig的startOffsetTime字段：默认为kafka.api.OffsetRequest.EarliestTime()开始读，也就是从Kafka中最早的消息开始处理。也可以设成kafka.api.OffsetRequest.LatestOffset,也就是最晚的消息开始读。也可以自己指定具体的值
• KafkaConfig的maxOffsetBehind字段：这个字段对于KafkaSpout的多个处理流程都有影响。当提交一个新topology时，如果没有forceFromStart, 当KafkaSpout对某个partition的处理进度落后startOffsetTime对应的offset多于此值时，KafkaSpout会丢弃中间的消息，从而强制赶上目标进度.比如，如果startOffsetTime设成了lastestTime，那么如果进度落后超过maxOffsetBehind，KafkaSpout会直接从latestTime对应的offset开始处理。如果设成了froceFromStart，则在提交新任务时，始终会从EarliestTime开始读。
• KafkaSpout的userStartOffsetTimeIfOffsetOutOfRange字段：如果设成true，那么当fetch消息时出错，且FetchResponse显示的出错原因是OFFSET_OUT_OF_RANGE，那么就会尝试从KafkaSpout指定的startOffsetTime对应的消息开始读。例如，如果有一批消息因为超过了保存期限被Kafka删除，并且zk里记录的消息在这批被删除的消息里。如果KafkaSpout试图从zk的记录继续读，那么就会出现OFFSET_OUT_OF_RANGE的错误，从而触发这个配置