strom-kafka簡單整合

官方文檔：http://storm.apache.org/releases/1.2.2/storm-kafka-client.html

原理：

KafkaSpout封裝了一個kafka的消費者，通過簡單配置servers以及topic後即可自動從kafka消費資料，并通過自定義的輸出字段發送資料到bolt進行處理

pom.xml

kafka和storm都使用了較新的版本

<dependencies>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>1.1.3</version>
		<!--送出到叢集運作時，需要取消注釋，否則jar包沖突-->
        <!--<scope>provided</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-kafka-client</artifactId>
        <version>1.2.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.12</artifactId>
        <version>2.1.0</version>
        <exclusions>
           <!-- <exclusion>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
            </exclusion>-->
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
            </exclusion>
            <exclusion>
                <groupId>log4j</groupId>
                <artifactId>log4j</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.1.0</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.14</version>
    </dependency>
</dependencies>
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
                <appendAssemblyId>false</appendAssemblyId>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
                <archive>
                    <manifest>
                        <!-- 此處指定main方法入口的class -->
                        <mainClass>com.kafkatest.TopologyTest</mainClass>
                    </manifest>
                </archive>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>assembly</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>

    </plugins>
</build>
           

Toplogy:

/**
 * Created by wxg on 2018/11/14 14:37
 */
public class TopologyTest {
    public static void main(String[] args) {
        TopologyTest topologyTest = new TopologyTest();
        if (!ArrayUtils.isEmpty(args)) {
            try {
                StormSubmitter.submitTopology("kafka-test", topologyTest.stormConfig(), topologyTest.builder().createTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("kafka-test", topologyTest.stormConfig(), topologyTest.builder().createTopology());
        }
    }

    private Config stormConfig() {
        Config config = new Config();
        config.setDebug(true);
        config.setNumAckers(1);
        return config;
    }

    private TopologyBuilder builder() {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("spout", kafkaSpout(), 1);
        builder.setBolt("bolt1", new Bolt1Test(), 1).shuffleGrouping("spout");
        return builder;
    }

    private KafkaSpout<String, String> kafkaSpout() {
        final Fields outputFields = new Fields("topic", "partition", "offset", "timestamp", "key", "msg_from_kafka");
        final String host = "ip_address:9092";
        final String topic = "topic_test";
        KafkaSpoutConfig<String, String> kafkaSpoutConfig;
        kafkaSpoutConfig = KafkaSpoutConfig
                .builder(host, topic)
                .setRecordTranslator((r) -> new Values(r.topic(), r.partition(), r.offset(), r.timestamp(), r.key(), r.value()), outputFields)
                .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST)
                .build();
        return new KafkaSpout<>(kafkaSpoutConfig);
    }
}
           

Bolt:

/**
 * Created by wxg on 2018/11/14 14:46
 */
public class Bolt1Test extends BaseBasicBolt {
    private Logger logger = LoggerFactory.getLogger(Bolt1Test.class);

    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
        String msg = tuple.getStringByField("msg_from_kafka");
        logger.info("bolt1處理資料: {}", msg);
        basicOutputCollector.emit(new Values(msg));
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("msg"));
    }
}
           

輸出

可以看到每發送一個tuple，就會跟一個ack，用來保證資料的可靠性，防止丢失。