K getKeyFromTuple(Tuple/TridentTuple tuple);
V getMessageFromTuple(Tuple/TridentTuple tuple);
顾名思义，调用这些方法可将元组映射到Kafka密钥和Kafka消息。如果只希望将一个字段作为键并将一个字段作为值，则可以使用提供的FieldNameBasedTupleToKafkaMapper.java实现。在KafkaBolt中，如果您出于向后兼容的原因使用默认构造函数构造FieldNameBasedTupleToKafkaMapper，则实现将始终查找具有字段名称“ key”和“ message”的字段。另外，您也可以使用非默认构造函数来指定其他键和消息字段。在TridentKafkaState中，您必须指定密钥和消息的字段名称是什么，因为没有默认的构造函数。在构造FieldNameBasedTupleToKafkaMapper的实例时应指定这些参数。
KafkaTopicSelector和三叉戟KafkaTopicSelector
该接口只有一种方法
public interface KafkaTopicSelector {
    String getTopics(Tuple/TridentTuple tuple);
该接口的实现应返回需要将元组的键/消息映射发布到的主题。您可以返回null，并且消息将被忽略。如果您有一个静态主题名称，则可以使用DefaultTopicSelector.java并在构造函数中设置主题的名称。
FieldNameTopicSelector和FieldIndexTopicSelector可以用来选择应该向其发布元组的主题。用户只需要在元组本身中为主题名称指定字段名称或字段索引。当找不到主题名称时， Field*TopicSelector将消息写入默认主题。请确保已创建默认主题。
指定Kafka生产者属性
您可以通过调用提供Storm拓扑中的所有生产者属性KafkaBolt.withProducerProperties()和TridentKafkaStateFactory.withProducerProperties() 。有关更多详细信息，请参见http://kafka.apache.org/documentation.html#newproducerconfigs部分“生产者的重要配置属性”。这些也定义在org.apache.kafka.clients.producer.ProducerConfig
使用通配符kafka主题匹配
您可以通过添加以下配置来进行通配符主题匹配
Config config = new Config();
config.put("kafka.topic.wildcard.match",true);
之后，您可以指定一个通配符主题以进行匹配，例如clickstream。* .log。这将匹配所有与clickstream.my.log，clickstream.cart.log等匹配的流
对于螺栓：
TopologyBuilder builder = new TopologyBuilder();
Fields fields = new Fields("key", "message");
FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
            new Values("storm", "1"),
            new Values("trident", "1"),
            new Values("needs", "1"),
            new Values("javadoc", "1")
spout.setCycle(true);
builder.setSpout("spout", spout, 5);
//set producer properties.
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "1");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaBolt bolt = new KafkaBolt()
        .withProducerProperties(props)
        .withTopicSelector(new DefaultTopicSelector("test"))
        .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper());
builder.setBolt("forwardToKafka", bolt, 8).shuffleGrouping("spout");
Config conf = new Config();
StormSubmitter.submitTopology("kafkaboltTest", conf, builder.createTopology());
对于三叉戟：
Fields fields = new Fields("word", "count");
FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
        new Values("storm", "1"),
        new Values("trident", "1"),
        new Values("needs", "1"),
        new Values("javadoc", "1")
spout.setCycle(true);
TridentTopology topology = new TridentTopology();
Stream stream = topology.newStream("spout1", spout);
//set producer properties.
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "1");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory()
        .withProducerProperties(props)
        .withKafkaTopicSelector(new DefaultTopicSelector("test"))
        .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("word", "count"));
stream.partitionPersist(stateFactory, fields, new TridentKafkaStateUpdater(), new Fields());
Config conf = new Config();
StormSubmitter.submitTopology("kafkaTridentTest", conf, topology.build());
从卡夫卡读书（壶嘴）
通过使用以下命令配置喷口实现KafkaSpoutConfig类。此类使用Builder模式，可以通过调用Builders构造函数之一或通过在KafkaSpoutConfig类中调用静态方法builder来启动。
用于创建构建器的构造器或静态方法需要一些键值（以后可以更改），但这是启动喷口所需的最低配置。
bootstrapServers与Kafka消费者属性“ bootstrap.servers”相同。
topics壶嘴将消耗的主题可以是Collection特定主题名称（1个或多个）或正则表达式Pattern ，它指定将使用与该正则表达式匹配的所有主题。
如果您使用的是Builder构造函数，而不是其中一种builder方法，您还需要指定键解串器和值解串器。这是通过使用Java泛型来帮助确保类型安全。可以通过使用以下属性设置的使用者属性来指定反序列化器setProp 。有关详细信息，请参见KafkaConsumer配置文档。
有一些关键配置要注意。
setFirstPollOffsetStrategy允许您设置从何处开始使用数据。在故障恢复和第一次启动喷嘴时都使用此功能。允许的值在FirstPollOffsetStrategy javadocs中列出。
setProcessingGuarantee使您可以配置喷嘴可以提供的处理能力。这会影响消耗的偏移量可以提交的时间以及提交的频率。有关详细信息，请参见ProcessingGuarantee javadoc 。
setRecordTranslator允许您修改喷口将Kafka消费者记录转换为元组的方式，以及该元组将发布到的流中。默认情况下，“主题”，“分区”，“偏移”，“键”和“值”将被发送到“默认”流。如果要根据主题将条目输出到不同的流，storm提供ByTopicRecordTranslator 。有关如何使用它们的更多示例，请参见下文。
setProp和setProps可用于设置KafkaConsumer属性。这些属性的列表可以在Kafka网站上的KafkaConsumer配置文档中找到 。请注意，不支持KafkaConsumer自动提交。如果设置了“ enable.auto.commit”属性，则KafkaSpoutConfig构造函数将引发异常，并且壶嘴使用的使用者将始终将该属性设置为false。您可以配置类似的行为，以通过setProcessingGuarantee KafkaSpoutConfig构建器上的方法。
创建一个简单的不安全喷嘴
下面将使用发布到“主题”的所有事件，并将它们与“主题”，“分区”，“偏移”，“键”，“值”的字段一起发送到MyBolt。
final TopologyBuilder tp = new TopologyBuilder();
tp.setSpout("kafka_spout", new KafkaSpout<>(KafkaSpoutConfig.builder("127.0.0.1:" + port, "topic").build()), 1);
tp.setBolt("bolt", new myBolt()).shuffleGrouping("kafka_spout");
通配符主题
通配符主题将从指定代理列表中存在的所有主题中使用，并与模式匹配。因此，在以下示例中，“ topic”，“ topic_foo”和“ topic_bar”将全部与模式“ topic。*”匹配，但“ not_my_topic”将不匹配。
final TopologyBuilder tp = new TopologyBuilder();
tp.setSpout("kafka_spout", new KafkaSpout<>(KafkaSpoutConfig.builder("127.0.0.1:" + port, Pattern.compile("topic.*")).build()), 1);
tp.setBolt("bolt", new myBolt()).shuffleGrouping("kafka_spout");
final TopologyBuilder tp = new TopologyBuilder();
//By default all topics not covered by another rule, but consumed by the spout will be emitted to "STREAM_1" as "topic", "key", and "value"
ByTopicRecordTranslator<String, String> byTopic = new ByTopicRecordTranslator<>(
    (r) -> new Values(r.topic(), r.key(), r.value()),
    new Fields("topic", "key", "value"), "STREAM_1");
//For topic_2 all events will be emitted to "STREAM_2" as just "key" and "value"
byTopic.forTopic("topic_2", (r) -> new Values(r.key(), r.value()), new Fields("key", "value"), "STREAM_2");
tp.setSpout("kafka_spout", new KafkaSpout<>(KafkaSpoutConfig.builder("127.0.0.1:" + port, "topic_1", "topic_2", "topic_3").build()), 1);
tp.setBolt("bolt", new myBolt()).shuffleGrouping("kafka_spout", "STREAM_1");
tp.setBolt("another", new myOtherBolt()).shuffleGrouping("kafka_spout", "STREAM_2");
final TridentTopology tridentTopology = new TridentTopology();
final Stream spoutStream = tridentTopology.newStream("kafkaSpout",
    new KafkaTridentSpoutOpaque<>(KafkaSpoutConfig.builder("127.0.0.1:" + port, Pattern.compile("topic.*")).build()))
      .parallelismHint(1)
Trident不支持多个流，并且将忽略为输出设置的任何流。但是，如果每个输出主题的字段都不相同，它将抛出异常，并且不会继续。
使用storm-kafka-client的示例拓扑可以在Storm源或二进制发行版中包含的examples / storm-kafka-client-examples目录中找到。
自定义RecordTranslators（高级）
在大多数情况下，内置的SimpleRecordTranslator和ByTopicRecordTranslator应该可以解决您的用例。如果确实遇到需要定制的情况，那么本文档将描述如何正确执行此操作，以及涉及的一些不太明显的类。
要点apply是采取一个ConsumerRecord并将其变成一个List