SpringBoot aop + kafka + ELK 将用户访问接口通过spring aop将数据传输到kafka，通过logstash将数据存入es。 本次讲解 storm消费kafka数据

pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <groupId>org.springframework.boot</groupId>
  7.         <artifactId>spring-boot-starter-parent</artifactId>
  8.         <version>2.2.1.RELEASE</version>
  9.         <relativePath/> <!-- lookup parent from repository -->
  10.     </parent>
  11.     <groupId>com.ywf</groupId>
  12.     <artifactId>ywf-storm</artifactId>
  13.     <version>0.0.1-SNAPSHOT</version>
  14.     <name>ywf-storm</name>
  15.     <description>Demo project for Spring Boot</description>
  16.     <properties>
  17.         <java.version>1.8</java.version>
  18.         <storm.version>1.2.2</storm.version>
  19.         <kafka.version>1.1.0</kafka.version>
  20.     </properties>
  21.     <dependencies>
  22.         <dependency>
  23.             <groupId>org.springframework.boot</groupId>
  24.             <artifactId>spring-boot-starter</artifactId>
  25.         </dependency>
  26.         <dependency>
  27.             <groupId>org.springframework.boot</groupId>
  28.             <artifactId>spring-boot-starter</artifactId>
  29.             <exclusions>
  30.                 <exclusion>
  31.                     <groupId>org.apache.logging.log4j</groupId>
  32.                     <artifactId>log4j-to-slf4j</artifactId>
  33.                 </exclusion>
  34.                 <exclusion>
  35.                     <groupId>ch.qos.logback</groupId>
  36.                     <artifactId>logback-classic</artifactId>
  37.                 </exclusion>
  38.             </exclusions>
  39.         </dependency>
  40.         <!-- test -->
  41.         <dependency>
  42.             <groupId>org.springframework.boot</groupId>
  43.             <artifactId>spring-boot-starter-test</artifactId>
  44.             <scope>test</scope>
  45.         </dependency>
  46.         <!-- storm -->
  47.         <dependency>
  48.             <groupId>org.apache.storm</groupId>
  49.             <artifactId>storm-core</artifactId>
  50.             <version>${storm.version}</version>
  51.             <!--在本地模式运行的时候需要把下面的给注释掉-->
  52.             <!--            <scope>provided</scope>-->
  53.         </dependency>
  54.         <!-- JVM上的实时监控类库 -->
  55.         <dependency>
  56.             <groupId>com.codahale.metrics</groupId>
  57.             <artifactId>metrics-core</artifactId>
  58.             <version>3.0.2</version>
  59.             <!--   <scope>provided</scope>-->
  60.         </dependency>
  61.         <dependency>
  62.             <groupId>org.apache.storm</groupId>
  63.             <artifactId>storm-kafka</artifactId>
  64.             <version>${storm.version}</version>
  65.         </dependency>
  66.         <dependency>
  67.             <groupId>org.apache.kafka</groupId>
  68.             <artifactId>kafka_2.11</artifactId>
  69.             <version>${kafka.version}</version>
  70.             <exclusions>
  71.                 <exclusion>
  72.                     <groupId>org.apache.zookeeper</groupId>
  73.                     <artifactId>zookeeper</artifactId>
  74.                 </exclusion>
  75.                 <exclusion>
  76.                     <groupId>org.slf4j</groupId>
  77.                     <artifactId>slf4j-log4j12</artifactId>
  78.                 </exclusion>
  79.                 <exclusion>
  80.                     <groupId>log4j</groupId>
  81.                     <artifactId>log4j</artifactId>
  82.                 </exclusion>
  83.             </exclusions>
  84.         </dependency>
  85.         <dependency>
  86.             <groupId>org.apache.kafka</groupId>
  87.             <artifactId>kafka-clients</artifactId>
  88.             <version>${kafka.version}</version>
  89.         </dependency>
  90.         <dependency>
  91.             <groupId>com.alibaba</groupId>
  92.             <artifactId>fastjson</artifactId>
  93.             <version>1.1.26</version>
  94.         </dependency>
  95.     </dependencies>
  96.     <build>
  97.         <plugins>
  98.             <plugin>
  99.                 <groupId>org.apache.maven.plugins</groupId>
  100.                 <artifactId>maven-shade-plugin</artifactId>
  101.                 <executions>
  102.                     <execution>
  103.                         <phase>package</phase>
  104.                         <goals>
  105.                             <goal>shade</goal>
  106.                         </goals>
  107.                     </execution>
  108.                 </executions>
  109.             </plugin>
  110.         </plugins>
  111.     </build>
  112. </project>

SystemLogBolt

package com.ywf.storm.bolt;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
 * @Author:ywf
 */
public class SystemLogBolt extends BaseRichBolt {
    private OutputCollector collector;
    /**
     * 记录每个用户接口访问量
     */
    private Map<String, Integer> userMethodMap = new ConcurrentHashMap<>();
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }
    @Override
    public void execute(Tuple tuple) {
        // !!! 一定要用bytes 上游kafkaspout 传递过来的数据
        byte[] bytes = tuple.getBinaryByField("bytes");
        String value = new String(bytes);
        JSONObject jsonObject = JSON.parseObject(value);
        String paramsStr = jsonObject.getString("params");
        JSONObject params = JSONObject.parseObject(paramsStr);
        String username = params.getString("username");
        int count = userMethodMap.getOrDefault(username, 0) + 1;
        userMethodMap.put(username, count);
        System.out.println(userMethodMap);
        // ack
        collector.ack(tuple);
    }
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    }
}

LogKafkaTopology

package com.ywf.storm;
import com.ywf.storm.bolt.SystemLogBolt;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.kafka.BrokerHosts;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.kafka.SpoutConfig;
import org.apache.storm.kafka.ZkHosts;
import org.apache.storm.topology.TopologyBuilder;
import java.util.UUID;
/**
 * kafka 日志kafka拓扑
 * @Author:ywf
 */
public class LogKafkaTopology {
    public static void main(String[] args) {
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        String zkConnStr = "node1:2181,node2:2181,node3:2181";
        BrokerHosts brokerHosts = new ZkHosts(zkConnStr);
        String topicName = "ywf-system-log";
        String zkRoot = "/ywf-system-log";
        String id = UUID.randomUUID().toString();
        /**
         * BrokerHosts hosts zk集群地址
         * String topic, 主题
         * String zkRoot, zk主题地址
         * String id 唯一标识
         */
        SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topicName, zkRoot, id);
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
        topologyBuilder.setSpout("kafkaSpout", kafkaSpout).setNumTasks(2);
        topologyBuilder.setBolt("systemLogBolt", new SystemLogBolt(),2).shuffleGrouping("kafkaSpout")
                .setNumTasks(2);
        Config config = new Config();
        config.setNumWorkers(2);
        config.setDebug(true);
        /**
         * 本地模式storm
         */
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("logKafkaTopology", config, topologyBuilder.createTopology());
    }
}

本地启动应用