本次实验在Windows上实现，使用了IDEA和Docker Desktop。

特别提醒，做实验一定要关注自己的文件位置和名称，否则会出现各种各样的报错，切记切记！！！

一、启动集群（包括zookeeper，kafka，storm）

1.启动zookeeper集群

创建一个文件夹zookeeper，在其中创建三个文件夹：zoo1,zoo2,zoo3，分别在创建的三个文件夹中创建两个文件夹：data，data_log

在zookeeper文件夹中创建一个yml文件，名为"zookeeper-compose.yml"

注意：networks需要自己定义，绿色箭头为本次实验用到的网络，只在此处说明一次，后续不重复

docker network create 网络名字

version: '3.8'

services:
  zookeeper-1:
    image: zookeeper:3.8
    container_name: zookeeper-1
    hostname: zookeeper-1
    ports:
      - "2181:2181"
    environment:
      ZOO_JUTE_MAXBUFFER: 4194304
      ZOO_MY_ID: 1
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=zookeeper-2:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
      ZOO_4LW_COMMANDS_WHITELIST: "*"
      ALLOW_ANONYMOUS_LOGIN: "yes"
    networks:
      - zk-net
    volumes:
      - zoo1_data:/data
      - zoo1_datalog:/datalog
    restart: unless-stopped

  zookeeper-2:
    image: zookeeper:3.8
    container_name: zookeeper-2
    hostname: zookeeper-2
    ports:
      - "2182:2181"
    environment:
      ZOO_JUTE_MAXBUFFER: 4194304
      ZOO_MY_ID: 2
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=zookeeper-2:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
      ZOO_4LW_COMMANDS_WHITELIST: "*"
      ALLOW_ANONYMOUS_LOGIN: "yes"
    networks:
      - zk-net
    volumes:
      - zoo2_data:/data
      - zoo2_datalog:/datalog
    restart: unless-stopped
    depends_on:
      - zookeeper-1

  zookeeper-3:
    image: zookeeper:3.8
    container_name: zookeeper-3
    hostname: zookeeper-3
    ports:
      - "2183:2181"
    environment:
      ZOO_JUTE_MAXBUFFER: 4194304
      ZOO_MY_ID: 3
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=zookeeper-2:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
      ZOO_4LW_COMMANDS_WHITELIST: "*"
      ALLOW_ANONYMOUS_LOGIN: "yes"
    networks:
      - zk-net
    volumes:
      - zoo3_data:/data
      - zoo3_datalog:/datalog
    restart: unless-stopped
    depends_on:
      - zookeeper-1
      - zookeeper-2

volumes:
  zoo1_data:
  zoo1_datalog:
  zoo2_data:
  zoo2_datalog:
  zoo3_data:
  zoo3_datalog:

networks:
  zk-net:
    external: true

打开docker desktop

打开终端进入zookeeper-compose.yml所在的文件夹，输入

docker-compose -f zookeeper-compose.yml up -d

进入docker desktop也可以查看集群信息

2.启动kafka集群

创建一个yml文件，名为"kafka.yml"

需要特别注意的是：配置文件中有多处需要调整，相关位置已在代码中明确标注

version: "3.6"
services:
  kafka1:
    container_name: kafka1
    image: 'bitnami/kafka:3.6.1'
    user: root
    ports:
      - '19092:9092'
      - '19093:9093'
    environment:
      - KAFKA_ENABLE_KRAFT=yes
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 配置为你的IP（将192.168.215.138改为你的ip）
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.215.138:19092
      - KAFKA_CFG_NODE_ID=1
      - KAFKA_KRAFT_CLUSTER_ID=iZWRiSqjZAlYwlKEqHFQWI
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@172.23.0.11:9093,2@172.23.0.12:9093,3@172.23.0.13:9093
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
    volumes:
      # 配置为你的路径（可以不用先创建，运行了这个配置文件会自动在指定位置创建）
      - D:\public\kafka\broker01:/bitnami/kafka:rw
    networks:
      netkafka:
        ipv4_address: 172.23.0.11

  kafka2:
    container_name: kafka2
    image: 'bitnami/kafka:3.6.1'
    user: root
    ports:
      - '29092:9092'
      - '29093:9093'
    environment:
      - KAFKA_ENABLE_KRAFT=yes
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 配置为你的IP: 192.168.32.53
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.215.138:29092
      - KAFKA_CFG_NODE_ID=2
      - KAFKA_KRAFT_CLUSTER_ID=iZWRiSqjZAlYwlKEqHFQWI
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@172.23.0.11:9093,2@172.23.0.12:9093,3@172.23.0.13:9093
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
    volumes:
      # 配置为你的路径
      - D:\public\kafka\broker02:/bitnami/kafka:rw
    networks:
      netkafka:
        ipv4_address: 172.23.0.12

  kafka3:
    container_name: kafka3
    image: 'bitnami/kafka:3.6.1'
    user: root
    ports:
      - '39092:9092'
      - '39093:9093'
    environment:
      - KAFKA_ENABLE_KRAFT=yes
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      # 已配置为你的IP
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.215.138:39092
      - KAFKA_CFG_NODE_ID=3
      - KAFKA_KRAFT_CLUSTER_ID=iZWRiSqjZAlYwlKEqHFQWI
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@172.23.0.11:9093,2@172.23.0.12:9093,3@172.23.0.13:9093
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_HEAP_OPTS=-Xmx512M -Xms256M
    volumes:
      # 已配置为你的路径
      - D:\public\kafka\broker03:/bitnami/kafka:rw
    networks:
      netkafka:
        ipv4_address: 172.23.0.13

networks:
  netkafka:
    external: true
    driver: bridge
    ipam:
      config:
        - subnet: 172.23.0.0/25
          gateway: 172.23.0.1

打开终端，cd到kafka.yml所在的文件夹。再输入命令启动kafka集群

docker-compose -f kafka.yml up -d

在docker desktop查看结果

由于本次实验需要使用主题，我们在这里先把主题创建好，进入kafka容器后创建storm-topic主题

docker exec -it kafka1 bash

/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server 172.23.0.11:9092,172.23.0.12:9092,172.23.0.13:9092  --create --topic storm-topic --partitions 3 --replication-factor 1  

3.启动storm集群

创建yml文件，名为"storm-compose.yml"

注意：zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181要改成自己的zookeeper名称，如果前面和我的配置文件一样就不需要改

version: '3'
services:
  nimbus1:
    image: storm
    restart: always
    command: storm nimbus
    container_name: nimbus1
    environment:
      STORM_ZOOKEEPER_SERVERS: "zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181"
    ports:
      - "6627:6627"
    networks:
      - storm-network
      - zk-net

  supervisor1:
    image: storm
    restart: always
    command: storm supervisor
    container_name: supervisor1
    depends_on:
      - nimbus1
    environment:
      STORM_ZOOKEEPER_SERVERS: "zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181"
      STORM_NIMBUS_HOST: "nimbus1"
    links:
      - nimbus1:nimbus1
    ports:
      - "6700:6700"
      - "6701:6701"
      - "6702:6702"
      - "6703:6703"
      - "8000:8000"
    networks:
      - storm-network
      - zk-net

  ui1:
    image: storm
    restart: always
    command: storm ui
    container_name: ui1
    depends_on:
      - nimbus1
    environment:
      STORM_ZOOKEEPER_SERVERS: "zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181"
      STORM_NIMBUS_HOST: "nimbus1"
    links:
      - nimbus1:nimbus1
    ports:
      - "8080:8080"
    networks:
      - storm-network
      - zk-net

  nimbus2:
    image: storm
    restart: always
    command: storm nimbus
    container_name: nimbus2
    environment:
      STORM_ZOOKEEPER_SERVERS: "zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181"
      STORM_NIMBUS_HOST: "nimbus1"
    ports:
      - "6628:6627"
    networks:
      - storm-network
      - zk-net

  supervisor2:
    image: storm
    restart: always
    command: storm supervisor
    container_name: supervisor2
    depends_on:
      - nimbus2
    environment:
      STORM_ZOOKEEPER_SERVERS: "zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181"
      STORM_NIMBUS_HOST: "nimbus1"
    links:
      - nimbus2:nimbus2
    ports:
      - "6704:6700"
      - "6705:6701"
      - "6706:6702"
      - "6707:6703"
      - "8001:8000"
    networks:
      - storm-network
      - zk-net

  nimbus3:
    image: storm
    restart: always
    command: storm nimbus
    container_name: nimbus3
    environment:
      STORM_ZOOKEEPER_SERVERS: "zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181"
      STORM_NIMBUS_HOST: "nimbus1"
    ports:
      - "6629:6627"
    networks:
      - storm-network
      - zk-net

  supervisor3:
    image: storm
    restart: always
    command: storm supervisor
    container_name: supervisor3
    depends_on:
      - nimbus3
    environment:
      STORM_ZOOKEEPER_SERVERS: "zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181"
      STORM_NIMBUS_HOST: "nimbus1"
    links:
      - nimbus3:nimbus3
    ports:
      - "6708:6700"
      - "6709:6701"
      - "6710:6702"
      - "6711:6703"
      - "8002:8000"
    networks:
      - storm_storm-network
      - zk-net

networks:
  storm-network:
  zk-net:
    external: true

打开终端，cd到storm-compose.yml所在的文件夹。再输入命令启动storm集群

 docker-compose -f storm-compose.yml up -d

再docker desktop查看

接下来的操作非常重要！！！

点击nimbus容器，点击File，找到storm.yaml，如图：

右键点击该文件，选择edit file，后续操作如图所示（第一步的需要改为自己的zookeeper集群名称）。storm集群的所有文件都需要修改这个地方，修改好后记得保存（第三步），最后记得重启（第四步）。

最后访问：localhost:8080看到如图：

至此，集群搭建全部完毕

 

二、Storm 案例实现

1.java代码编写

注意：代码中的ip地址和kafka主题，mysql相关消息需要修改为自己的内容

首先编写Kafka中写入数据WriteTopology类：

package org.example.kafka_storm2;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.bolt.KafkaBolt;
import org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
import org.apache.storm.kafka.bolt.selector.DefaultTopicSelector;
import org.apache.storm.topology.TopologyBuilder;

import java.util.Properties;

/**
 * 将读取到的数据分发到Kafka中
 */
public class WriteTopology {
    private static final String BOOTSTRAP_SERVERS = "192.168.215.138:19092,192.168.215.138:29092,192.168.215.138:39092"; //kafka 地址
    //private static final String BOOTSTRAP_SERVERS = "localhost:9093"; //kafka 地址
    private static final String TOPIC_NAME = "storm-topic";

    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();

        // 定义Kafka生产者属性
        Properties props = new Properties();
        /*
         * 指定broker的地址清单，清单里不需要包含所有的broker地址，生产者会从
         * 给定的broker里查找其他broker的信息。
         * 不过建议至少要提供两个broker的信息作为容错。
         */
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        /*
         * acks参数指定了必须要有多少个分区副本收到消息，生产者才会认为消息写
         * 入是成功的。
         * acks=0 : 生产者在成功写入消息之前不会等待任何来自服务器的响应。
         * acks=1 : 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成
         * 功响应。
         * acks=all : 只有当所有参与复制的节点全部收到消息时，生产者才会收到一个
         * 来自服务器的成功响应。
         */
        props.put("acks", "all");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("client.encoding.default", "UTF-8");
        props.put("serializer.encoding", "UTF-8");
        props.put("deserializer.encoding", "UTF-8");

        Config config = new Config();
        config.setNumWorkers(2);
        config.setDebug(true);
        config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1000); // 允许Spout在处理数据时保持更多的未处理数据，确保持续地从数据源中读取数据

        KafkaBolt<String, String> bolt = new KafkaBolt<String, String>()
                .withProducerProperties(props)
                .withTopicSelector(new DefaultTopicSelector(TOPIC_NAME))
                .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper<>());

        // 设置Spout
        builder.setSpout("sourceSpout", new DataSourceSpout2(), 2).setNumTasks(3);

        // 设置Bolt - 修正了这行语法错误
        builder.setBolt("kafkaBolt", bolt, 2)
                .shuffleGrouping("sourceSpout")
                .setNumTasks(3);

        if (args.length > 0 && args[0].equals("cluster")) {
            try {
                StormSubmitter.submitTopology("StormClusterWritingToKafkaClusterApp",
                        config,
                        builder.createTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("LocalWritingToKafkaApp",
                    config, builder.createTopology());

            // 本地模式运行一段时间后关闭
            Thread.sleep(60000); // 运行60秒
            cluster.shutdown();
        }
    }
}

编写DataSourceSpout2类

package org.example.kafka_storm2;

import org.apache.storm.shade.org.apache.commons.lang.StringUtils;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * 从产生的股票文件中读取数据
 */
public class DataSourceSpout2 extends BaseRichSpout {
    private SpoutOutputCollector spoutOutputCollector;
    private Set<String> processedData; // 用于存储已处理的数据
    private BufferedReader reader; // 文件读取器
    private List<String> fileNames;
    private String directoryPath;

    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.spoutOutputCollector = spoutOutputCollector;
        this.processedData = new HashSet<>();
        this.fileNames = Arrays.asList("src/main/resources/data/stock-part1.csv", "src/main/resources/data/stock-part2.csv");
        this.directoryPath = "src/main/resources/data";
    }

    @Override
    public void nextTuple() {
        File directory = new File(directoryPath);
        File[] files = directory.listFiles((dir, name) -> name.toLowerCase().endsWith(".csv"));

        if (files != null) {
            for (File file : files) {
                try {
                    FileInputStream fis = new FileInputStream(file);
                    InputStreamReader isr = new InputStreamReader(fis, StandardCharsets.UTF_8);
                    BufferedReader reader = new BufferedReader(isr);

                    // 跳过第一行
                    reader.readLine();

                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (!processedData.contains(file.getName() + ":" + line)) {
                            byte[] lineBytes = line.getBytes("UTF-8");
                            spoutOutputCollector.emit(new Values(file.getName(), line));
                            processedData.add(file.getName() + ":" + line);
                        }
                    }
                    reader.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }

        // 添加适当的延迟，以避免循环过快
        Utils.sleep(1000);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("key", "message"));
    }
}

之后使用KafkaStormTopology进行读取kafka中的数据，并将数据传入bolt中进行处理 并写入数据。

package org.example.kafka_storm2;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff;
import org.apache.storm.kafka.spout.KafkaSpoutRetryService;
import org.apache.storm.topology.TopologyBuilder;

import java.util.Collections;
import java.util.logging.Logger;

/**
 * 从Kafka中读取数据的Storm拓扑
 */
public class KafkaStormTopology {

    private static final String BOOTSTRAP_SERVERS = "192.168.215.138:19092,192.168.215.138:29092,192.168.215.138:39092";
    private static final String TOPIC_NAME = "storm-topic";

    private static final Logger logger = Logger.getLogger(KafkaStormTopology.class.getName());

    static {
        // 关闭父级日志记录器
        Logger parentLogger = logger.getParent();
        parentLogger.setLevel(java.util.logging.Level.OFF);
    }

    public static void main(String[] args) throws Exception {
        new KafkaStormTopology().execute(args);
    }

    private void execute(String[] args) throws Exception {
        if (args.length < 1) {
            System.err.println("Usage: java -jar your-app.jar <topology-name>");
            System.exit(1);
        }

        String topologyName = args[0];

        final TopologyBuilder builder = new TopologyBuilder();

        // 构建拓扑结构
        builder.setSpout("kafka_spout",
                new KafkaSpout<>(getKafkaSpoutConfig(BOOTSTRAP_SERVERS, TOPIC_NAME)), 1);

        builder.setBolt("split_bolt", new SplitBolt(), 2)
                .shuffleGrouping("kafka_spout")
                .setNumTasks(2);

        builder.setBolt("stat_store_bolt", new StatAndStoreBolt(), 1)
                .shuffleGrouping("split_bolt")
                .setNumTasks(1);

        // 提交到远程集群
        submitToRemoteByCode(builder, topologyName);
    }

    private void submitToRemoteByCode(TopologyBuilder builder, String topologyName) throws Exception {
        // 配置
        Config config = new Config();
        config.put(Config.NIMBUS_SEEDS, Collections.singletonList("192.168.215.138"));
        config.put(Config.NIMBUS_THRIFT_PORT, 6627);
        config.put(Config.STORM_ZOOKEEPER_SERVERS, Collections.singletonList("192.168.215.138"));
        config.put(Config.STORM_ZOOKEEPER_PORT, 2181);
        config.put(Config.TASK_HEARTBEAT_FREQUENCY_SECS, 10000);
        config.setDebug(false);
        config.setNumAckers(3);
        config.setMaxTaskParallelism(20);

        // assembly模式打包的本机jar包路径
//        String jarLocalPath = "D:\\tools\\IDEA\\idea_projects\\Flink\\target\\flink-stock-analysis-1.0-SNAPSHOT-jar-with-dependencies.jar";
//        System.setProperty("storm.jar", jarLocalPath);

        try {
            StormSubmitter.submitTopologyAs(topologyName, config, builder.createTopology(), null, null, "root");
        } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
            logger.severe("Failed to submit topology to cluster: " + e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * 创建Kafka Spout配置
     */
    private static KafkaSpoutConfig<String, String> getKafkaSpoutConfig(String bootstrapServers, String topic) {
        return KafkaSpoutConfig.builder(bootstrapServers, topic)
                .setProp(ConsumerConfig.GROUP_ID_CONFIG, "testGroup")
                .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
                .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
                .setProp("key.deserializer.encoding", "UTF-8")
                .setProp("value.deserializer.encoding", "UTF-8")
                .setRetry(getRetryService())
                .setOffsetCommitPeriodMs(10_000)
                .build();
    }

    /**
     * 定义重试策略
     */
    private static KafkaSpoutRetryService getRetryService() {
        return new KafkaSpoutRetryExponentialBackoff(
                KafkaSpoutRetryExponentialBackoff.TimeInterval.microSeconds(500),
                KafkaSpoutRetryExponentialBackoff.TimeInterval.milliSeconds(2),
                Integer.MAX_VALUE,
                KafkaSpoutRetryExponentialBackoff.TimeInterval.seconds(10));
    }
}

bolt定义编写，这里以分词任务为例， splitBolt的代码如下：

package org.example.kafka_storm2;

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 org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 切分并统计从 Kafka 获取的数据
 */
public class SplitBolt extends BaseRichBolt {
    private OutputCollector collector;
    private int count;
    private List<Tuple> tuples;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.count = 0;
        this.tuples = new ArrayList<>();
    }

    @Override
    public void execute(Tuple tuple) {
        try {
            String[] words = processAndEmitData(tuple);

            // 使用逗号作为分隔符
            int volume = Integer.parseInt(words[4]);
            double amount = Double.parseDouble(words[3]);
            String time = words[0];
            String tradeType = words[5];
            String stockCode = words[1];
            String stockName = words[2];
            String tradePlace = words[6];
            String tradePlatform = words[7];
            String industryType = words[8];

            collector.emit(new Values(volume, amount, time, tradeType, stockCode, stockName, tradePlace, tradePlatform, industryType));
            // 必须ack,否则会重复消费kafka中的消息
            collector.ack(tuple);
        } catch (Exception e) {
            e.printStackTrace();
            collector.fail(tuple);
        }
    }

    public String[] processAndEmitData(Tuple tuple) {
        String[] words;
        String line = tuple.getStringByField("value");
        System.out.println("received from kafka : " + line);
        words = line.split(",");
        return words;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new org.apache.storm.tuple.Fields("volume", "amount", "time", "tradeType", "stockCode", "stockName", "tradePlace", "tradePlatform", "industryType"));
    }
}

 

编写统计数据类

package org.example.kafka_storm2;

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.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.HashMap;
import java.util.Map;

/**
 * 统计和存储数据到数据库
 */
public class StatAndStoreBolt extends BaseRichBolt {
    private OutputCollector collector;
    private Connection connection;
    private Map<String, Integer> stockVolumeStats;
    private Map<String, Double> stockAmountStats;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.stockVolumeStats = new HashMap<>();
        this.stockAmountStats = new HashMap<>();

        // 初始化数据库连接
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection(
                    "jdbc:mysql://192.168.215.138:3306/storm_analysis?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
                    "sss",
                    "123"
            );
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void execute(Tuple tuple) {
        try {
            int volume = tuple.getIntegerByField("volume");
            double amount = tuple.getDoubleByField("amount");
            String time = tuple.getStringByField("time");
            String tradeType = tuple.getStringByField("tradeType");
            String stockCode = tuple.getStringByField("stockCode");
            String stockName = tuple.getStringByField("stockName");
            String tradePlace = tuple.getStringByField("tradePlace");
            String tradePlatform = tuple.getStringByField("tradePlatform");
            String industryType = tuple.getStringByField("industryType");

            // 统计逻辑
            updateStatistics(stockCode, volume, amount);

            // 存储到数据库
            storeToDatabase(volume, amount, time, tradeType, stockCode, stockName,
                    tradePlace, tradePlatform, industryType);

            System.out.println("Processed: " + stockCode + " - Volume: " + volume + " - Amount: " + amount);

            collector.ack(tuple);
        } catch (Exception e) {
            e.printStackTrace();
            collector.fail(tuple);
        }
    }

    private void updateStatistics(String stockCode, int volume, double amount) {
        // 更新成交量统计
        stockVolumeStats.put(stockCode, stockVolumeStats.getOrDefault(stockCode, 0) + volume);
        // 更新成交额统计
        stockAmountStats.put(stockCode, stockAmountStats.getOrDefault(stockCode, 0.0) + amount);

        // 定期输出统计信息
        if (stockVolumeStats.size() % 10 == 0) {
            System.out.println("=== Stock Statistics ===");
            for (Map.Entry<String, Integer> entry : stockVolumeStats.entrySet()) {
                System.out.println(entry.getKey() + ": Volume=" + entry.getValue() +
                        ", Amount=" + stockAmountStats.get(entry.getKey()));
            }
            System.out.println("========================");
        }
    }

    private void storeToDatabase(int volume, double amount, String time, String tradeType,
                                 String stockCode, String stockName, String tradePlace,
                                 String tradePlatform, String industryType) {
        String sql = "INSERT INTO stock_trades (volume, amount, time, trade_type, stock_code, " +
                "stock_name, trade_place, trade_platform, industry_type) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";

        try (PreparedStatement pstmt = connection.prepareStatement(sql)) {
            pstmt.setInt(1, volume);
            pstmt.setDouble(2, amount);
            pstmt.setString(3, time);
            pstmt.setString(4, tradeType);
            pstmt.setString(5, stockCode);
            pstmt.setString(6, stockName);
            pstmt.setString(7, tradePlace);
            pstmt.setString(8, tradePlatform);
            pstmt.setString(9, industryType);

            pstmt.executeUpdate();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 这个 Bolt 是拓扑的终点，不需要输出字段
    }

    @Override
    public void cleanup() {
        // 关闭数据库连接
        try {
            if (connection != null && !connection.isClosed()) {
                connection.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

pom.xml配置文件（其中有一些是flink的配置文件，我懒得再搞一个文件索性写一起了）

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.yourcompany</groupId>
    <artifactId>flink-stock-analysis</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.18.0</flink.version>
        <!-- 添加 Storm 版本属性 -->
        <storm.version>2.4.0</storm.version>
    </properties>

    <dependencies>
        <!-- Flink 核心库 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- Flink Kafka Connector - 使用兼容版本 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>3.0.2-1.18</version>
        </dependency>

        <!-- 添加Kafka连接器基础依赖 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- Flink JDBC Connector -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc</artifactId>
            <version>3.1.2-1.18</version>
        </dependency>

        <!-- MySQL驱动 -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.33</version>
        </dependency>

        <!-- ========== 添加 Storm 相关依赖 ========== -->
        <!-- Apache Storm Core -->
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>${storm.version}</version>
            <!-- 本地测试时注释掉 provided，集群部署时取消注释 -->
             <scope>provided</scope>
        </dependency>

        <!-- Storm Kafka Integration -->
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-kafka-client</artifactId>
            <version>${storm.version}</version>
        </dependency>

        <!-- Kafka Clients (确保版本兼容) -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.4.0</version>
        </dependency>

        <!-- 添加SLF4J日志实现，解决SLF4J警告 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.36</version>
        </dependency>

        <!-- 添加Apache Commons依赖 -->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.12.0</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- 编译器插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>

            <!-- 打包插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.6.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                            <!-- 根据您的需要设置主类 -->
                            <mainClass>org.example.kafka_storm2.KafkaStormTopology</mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

打包，在IDEA右边找到maven（及第一步的m样式的选项），再按步骤进行打包。

2.使用storm集群实现综合数据流处理

先执行WriteTopology，可以创建kafka生产者，向kafka中传输数据

ps：我的终端是开了一个消费者查看数据是不是传进去了，正常都会传进去的

接下来我们打开终端，进入将刚刚打包的jar文件复制到storm集群中，再进入storm集群实现流处理

其中D:\tools\IDEA\idea_projects\Flink\target\flink-stock-analysis-1.0-SNAPSHOT-jar-with-dependencies.jar 改成自己的jar文件（那个更大的）

docker cp D:\tools\IDEA\idea_projects\Flink\target\flink-stock-analysis-1.0-SNAPSHOT-jar-with-dependencies.jar nimbus1:/tmp/

docker exec -it nimbus1 bash

storm jar /tmp/flink-stock-analysis-1.0-SNAPSHOT-jar-with-dependencies.jar org.example.kafka_storm2.KafkaStormTopology my-storm-topology

再去浏览器输入localhost:8080查看ui界面

可以发现Topology Summary中已经清晰地展示了当前网络拓扑结构的各项关键指标。

后续再续写性能调优

至此实验成功，再会^-^

ps：实验参考了Java整合Storm上传到远程服务器执行_storm 发布到线上运行-CSDN博客