AI导航网
微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦！

为什么一个星期工作量的工作，我做了一个多月，还没结束 (基于socket的分布式数据处理程序Java版)

分类：C#作者：秋荷雨翔原文地址

    为什么一个星期工作量的工作，我做了一个多月，还没结束

    为什么一个简单的小任务，我遇到这么多难题

    这是一个HIK平台WIFI数据接入的工作，先看下我的代码提交记录：

     首先有这两方面的原因：1、初学Java时间不长，不够熟练，这个原因浪费的时间并不多。2、与数据提供方沟通浪费的时间，因数据有问题，即deviceid和数据库中的deviceid对不上，导致程序筛选不到数据，需要对方修改相关配置，由于我感觉事情并不急，所以是隔几天催一下，浪费了不少天的时间，最后deviceid依然对不上，我是怎么解决的呢？由于基本上是一个设备布在一个地点，所以我是根据数据中的经纬度来筛选数据的，这种办法其实还是有一点点问题的。

    再说写代码花费的时间，服务部署到现网后，就开始发现BUG了，程序处理数据的速度不够，导致FTP上的ZIP包处理不完大量积压，因为我用的是单线程处理数据，遂改成多线程处理数据。

    经过优化和测试，发现数据处理速度还是不够，FTP上数据产生的速度大约是380条每秒，可能会更多，而单机每秒最多只能处理200多条数据，继续尝试优化，但还是不行。所以我面临第一个重要问题：把程序部署到多台机器，提高数据处理速度，程序该怎么写？

    对于这个问题，经过探索，我最终的解决办法是：把程序部署在5台机器上，程序跑起来后，能拿到机器名称，FTP上的ZIP文件名带有时间戳，根据机器名和时间戳的最后一位，把ZIP文件分配到5台机器处理，这样就解决了数据处理速度不够的问题。

    我想解决的第二个重要问题是：由于我目前做的是大数据维护，学了点Spark和Flink知识，所以我想用Spark改写，已经写好的程序是SpringBoot的，所以我要做的是SpringBoot整合Spark。周末我在家搭了一个Hadoop+Spark的分布式集群环境，写了个纯Spark的Demo跑，正常。然后写了个SpringBoot整合Spark的Demo，本地模式跑，即.setMaster("local[3]")，正常。但是提交到集群跑，web api和swagger在线文档正常，任务跑起来后在Spark Web页面可以看到，但日志报错。我试了3种方式，一种是打war包部署在tomcat里，一种是用命令java -jar运行jar包，一种是用spark-submit命令运行jar包，任务跑起来后都报错了，三种方式错误也不相同，搞到凌晨4点多，没有解决，放弃。不过，就算我成功了，接下来的问题，可能依然无法解决，就是数据处理完后，要推送到Kafka，网上的教学都是教Spark怎么处理Kafka流数据，Kafka数据是作为数据源的，不是作为目的地的，所以我的想法可能本身就是个问题。

    解释一下为什么要SpringBoot整合Spark，只用Spark不用SpringBoot不行吗？实际上之前同事就是这么干的，要么用SpringBoot，要么用Spark或Flink，没有把SpringBoot和二者在一起用过，为什么我要这么做？因为我要读MysqL数据库，我要用JDBC或者Mybatis等，如果不用SpringBoot，有些东西可能要自己搞，不太方便。

    既然SpringBoot整合Spark没有成功，那数据分配不均匀的问题怎么解决(FTP上ZIP文件5分种产生5个文件，2个数据量大，3个不是我需要的数据可以直接删了，但不管是按时间戳这个特征分配，还是按序号这个特征分配，都无法均匀分配，可能会导致一个节点数据积压，另外3个节点没有要处理的数据，即一个节点有难，3个节点围观，虽然时间尺度拉大后，数据分配是均匀的，但是数据处理延迟大了，5分钟产生一个ZIP文件，意味着，有的数据已经延迟处理了5分钟，而我这边还要延迟几分钟到几十分钟不等)？这是我准备解决的第三个重要问题。

    我用Socket解决了这个问题，我把程序部署在5个节点上，一个作为master节点，4个作为worker节点，程序启动后根据机器名判断，确定master节点，master节点从FTP上下载ZIP文件，FTP上ZIP文件可能很多，先只下载一个文件进行处理，一个文件中可能有多达10万条数据，也可能就几万条数据，还可能不是我需要的数据，直接删掉该ZIP文件即可，然后解析数据，再然后把数据集合平均分成4份，通过Socket发送给4个worker节点，worker节点收到数据集合，进行筛选和处理，然后推送到Kafka，数据处理完后，给master节点发送一条消息，可以重复发送几次，再加上是局域网，以确保master节点能收到消息(这个很重要，但这里也有问题)，master节点收到4个节点数据处理完成的消息后，删除FTP上的该ZIP文件。然后进行下一次处理，直到FTP上的ZIP文件全部处理完成并删除。我今天写完后，把写ES日志和发Kafka以及删除ZIP文件的代码注释掉，耗时的地方用Thread.sleep代替，部署到真实环境，看能不能稳定跑上一天。我感觉用Socket实现分布式数据处理，虽然能解决数据分配不均匀的问题，但是程序稳定性变差了，如果master节点收不到worker节点发来的数据处理完成的消息怎么办？假设其中一个worker节点的程序挂掉了呢？

    虽然只是一个简单的小任务，我真的是非常努力，如果我最初的设想成功，以后的类似服务都可以这么写，意义重大，可惜没搞成功，退而求其次，用Socket写分布式处理程序，我好像迷迷糊糊明白为什么Spark要依赖Hadoop了，我自己用Socket写漏洞很多啊，没有把数据持久化，万一漰了，数据就丢了。

    有没有大佬给指点一下，我努力的思路是不是有问题，有没有代码又容易写，又不容易写错，程序又稳定可靠的方案？

 

====================== 分隔线 ==============================================================================================

上面的使用Socket的方案存在的问题：1.master节点把大量数据发送到worker节点处理，占用大量带宽，并且耗时。2.master节点需要等待所有worker节点全部处理完成后，才能进行下一次处理，因网络传输耗时和各机器性能差异，导致某些worker节点空闲时间长，浪费大量时间

下面是新的使用Socket的方案，有下列优点：1.master节点和worker节点通信，只传输命令，不传输数据内容，节省带宽，省时可靠。2.worker节点处理完数据，立即请求下一次处理，每一个worker节点得到充分利用

 

最终实现方案(基于socket的分布式数据处理程序Java版)：

通过分布式锁，把FTP上的zip文件，均匀的分配到7个worker主机节点进行处理，master节点不处理数据，只负责处理7个woker节点的锁请求与数据处理请求，7个worker节点只要有3个正常工作，就能跟上FTP上数据产生的速度，不会造成zip文件积压，该方式数据处理延迟很小，能够及时处理FTP上产生的数据


大致流程：

1.任务启动，判断当前worker节点在FTP上是否存在私有文件，如果存在，则进入直接处理流程
2.如果私有文件不存在，则worker节点向master节点请求锁
3.master节点收到请求，发放锁
4.worker节点收到锁，重命名FTP上的一个文件为私有文件名，该文件名对其它节点不可见(各worker节点通过正则表达式判断处理)，然后释放锁
5.master节点收到释放锁的请求，释放信号量，使其能够继续处理下一个节点的锁请求
6.worker节点释放锁后，下载文件并处理数据，数据处理完，删除私有文件，并再次请求锁，以进行下一个文件的处理

主要实现代码：

1.ReadFtpFileService类：任务启动入口，判断FTP上私有文件是否存在，向master节点请求锁或者请求直接处理

package com.suncreate.wifi.service;

import com.suncreate.wifi.hikmodel.socketData;
 com.suncreate.wifi.tool.ftputils;
 com.suncreate.wifi.tool.socketUtil;
 org.slf4j.Logger;
 org.slf4j.LoggerFactory;
 org.springframework.beans.factory.annotation.Value;
 org.springframework.stereotype.Service;

 java.util.ArrayList;
 java.util.List;

@Service
public class ReadFtpFileService {
    private static final Logger log = LoggerFactory.getLogger(ReadFtpFileService.);

    @Value("${ftp.host}")
    private String ftpHost;

    @Value("${ftp.port}"int ftpPort;

    @Value("${ftp.username}" String ftpUserName;

    @Value("${ftp.password}" String ftpPassword;

    @Value("${node.name}" String nodeName;

    @Value("${master.ip}" String masterIp;

    @Value("${master.name}" String masterName;

    /**
     * 处理FTP上的zip数据
     */
    void ProcessZip() {
        try {
            if (nodeName.equals(masterName)) return; //master节点不处理数据，跳过

            log.info("ProcessZip() 开始");

            List<String> ftpFileList = getFtpFileList();
            if (ftpFileList != null && ftpFileList.size() > 0) {
                boolean bl = false;
                String fileName = null;
                for (String ftpFileName : ftpFileList) {
                    if (ftpFileName.endsWith(nodeName + ".zip")) {
                        bl = true;
                        fileName = ftpFileName;
                    }
                }

                if (bl) {
                    SocketUtil.Send(masterIp,new SocketData(nodeName,3,fileName)); 3:直接处理请求
                    ZipProcesstime.updateTime();
                } else {
                    SocketUtil.Send(masterIp,0)); 0:请求锁
                    ZipProcesstime.updateTime();
                }
            }  {
                ZipProcesstime.updateTime();

                RunProcessZipThread thread = new RunProcessZipThread(this);
                thread.start();
            }

            log.info("ProcessZip() 结束");
        } catch (Exception e) {
            log.error("ProcessZip 出错",e);
        }
    }

    private List<String> getFtpFileList() {
        List<String> result = new ArrayList<>();
        ftputils ftputils = new ftputils();

         {
            ftputils.login(ftpHost,ftpPort,ftpUserName,ftpPassword);
            ftputils.listFileName("/");
            result = ftputils.getFileNameList();
        }  (Exception e) {
            log.error("getFtpFileList() Failed"finally {
            ftputils.disConnection();
        }

        return result;
    }

}
View Code

2.socketServer类：用于启动Socket服务端线程

 com.suncreate.wifi.model.HotspotInfoCollected;
 com.suncreate.wifi.model.TermInfocharacteristics;
 com.suncreate.wifi.task.CheckScheduleConfig;
 com.suncreate.wifi.tool.SerializeUtil;
 org.springframework.beans.factory.annotation.Autowired;
import java.io.*;
 java.net.ServerSocket;
 java.net.socket;
 java.util.List;
 java.util.concurrent.CountDownLatch;
 java.util.concurrent.Executors;
 java.util.concurrent.ThreadPoolExecutor;

@Service
 SocketServer {
    final Logger log = LoggerFactory.getLogger(SocketServer.);

    @Autowired
     KafkaSendService kafkaSendService;

    @Autowired
     ReadFtpFileService readFtpFileService;

    @Value("${ftp.host}" String ftpPassword;

    @Value("${master.ip}" String masterIp;

    @Value("${node.name}" String nodeName;

    @Value("${master.name}"void start() throws IOException {
        ServerSocket serverSocket = new ServerSocket(18060);
        log.info("当前节点主机名称：" + nodeName);

        if (nodeName.equals(masterName)) { master节点多起几个线程与worker节点通信
            for (int i = 0; i < 7; i++) {
                SocketThread socketThread =  SocketThread(serverSocket,kafkaSendService,readFtpFileService,ftpHost,ftpPassword,masterIp,nodeName);
                socketThread.start();
            }
        } else { worker节点起一个线程与master通信即可
            SocketThread socketThread = View Code

3.socketThread类：Socket服务端接收命令，处理命令

 com.suncreate.wifi.hikmodel.ZipData;
 org.slf4j.LoggerFactory;

import java.util.concurrent.* java.util.regex.Pattern;

class SocketThread extends Thread {
    final Logger log = LoggerFactory.getLogger(SocketThread.);

    static ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(256static Semaphore semaphore = new Semaphore(1 ServerSocket serverSocket;

     KafkaSendService kafkaSendService;

     ReadFtpFileService readFtpFileService;

     String ftpHost;

     ftpPort;

     String ftpUserName;

     String ftpPassword;

    private String masterIp; master节点IP

     String nodeName;

    public SocketThread(ServerSocket serverSocket,KafkaSendService kafkaSendService,ReadFtpFileService readFtpFileService,String ftpHost, ftpPort,String ftpUserName,String ftpPassword,String masterIp,String nodeName) {
        this.serverSocket = serverSocket;
        this.kafkaSendService = kafkaSendService;
        this.readFtpFileService = readFtpFileService;
        this.ftpHost = ftpHost;
        this.ftpPort = ftpPort;
        this.ftpUserName = ftpUserName;
        this.ftpPassword = ftpPassword;
        this.masterIp = masterIp;
        this.nodeName = nodeName;
    }

     run() {
        while ( {
                    Socket socket = serverSocket.accept();
                    SocketData socketData = ;

                     {
                        InputStream inputStream = socket.getInputStream();
                        DataInputStream dataInputStream =  DataInputStream(inputStream);
                        BufferedInputStream bufferedInputStream =  BufferedInputStream(dataInputStream);
                        ByteArrayOutputStream byteArrayOutputStream =  ByteArrayOutputStream();

                        byte[] bytes = new byte[10240];
                         len;
                        while ((len = bufferedInputStream.read(bytes)) > 0) {
                            byteArrayOutputStream.write(bytes,0 (SocketData) SerializeUtil.Deserialize(byteArrayOutputStream.toByteArray());
                    }  (Exception e2) {
                        log.error("SocketServer socket 出错" {
                        socket.close();
                    }

                    if (socketData == ) {
                        continue;
                    }

                    0:请求锁
                    if (socketData.getCommand().equals(0)) {
                        log.info("收到节点 " + socketData.getNodeName() + " " + socket.getInetAddress().getHostAddress() + " 的命令 0:请求锁");
                        semaphore.tryAcquire(301:发放锁
                        log.info("向节点 " + socket.getInetAddress().getHostAddress() + " 发放锁");
                    }

                    1:发放锁
                    if (socketData.getCommand().equals(1)) {
                        log.info("收到节点 " + socketData.getNodeName() + " " + socket.getInetAddress().getHostAddress() + " 的命令 1:发放锁");
                        List<String> ftpFileList = getFtpFileList();
                        筛选
                        String pattern = "^[\\S\\s]*-[0-9]*-[0-9]*-[0-9]*.zip$";
                        int i = ftpFileList.size() - 1; i >= 0; i--) {
                            String ftpFileName = ftpFileList.get(i);
                            boolean bl = Pattern.matches(pattern,ftpFileName);
                            if (!bl) {
                                ftpFileList.remove(i);
                            }
                        }
                        ) {
                            String newFtpFileName = ftpFileList.get(0).replace(".zip","") + "-" + nodeName + ".zip";
                            renameFtpFile(ftpFileList.get(0),newFtpFileName);
                            SocketUtil.Send(socket.getInetAddress().getHostAddress(),2)); 2:释放锁

                            byte[] file = downloadFtpFile(newFtpFileName);
                            if (file != ) {
                                ZipData zipData =  ZipData(file);

                                if (zipData.getHotsoptList().size() > 0 || zipData.getTerminfoList().size() > 0) {
                                    processZipData(zipData);

                                    removeFtpFile(ftpHost,newFtpFileName); 删除FTP上的zip文件

                                    SocketUtil.Send(masterIp,1)">0:再次请求锁
                                    ZipProcesstime.updateTime();
                                }  {
                                    removeFtpFile(ftpHost,1)">
                                    RunProcessZipThread thread =  RunProcessZipThread(readFtpFileService);
                                    thread.start();
                                }
                            }  {
                                RunProcessZipThread thread =  RunProcessZipThread(readFtpFileService);
                                thread.start();
                            }
                        }  {
                            SocketUtil.Send(socket.getInetAddress().getHostAddress(),1)">2:释放锁

                            RunProcessZipThread thread =  RunProcessZipThread(readFtpFileService);
                            thread.start();
                        }
                    }

                    2:释放锁
                    if (socketData.getCommand().equals(2)) {
                        log.info("收到节点 " + socketData.getNodeName() + " " + socket.getInetAddress().getHostAddress() + " 的命令 2:释放锁");
                        semaphore.release();
                    }

                    3:直接处理请求
                    if (socketData.getCommand().equals(3)) {
                        log.info("收到节点 " + socketData.getNodeName() + " " + socket.getInetAddress().getHostAddress() + " 的命令 3:直接处理请求");
                        SocketUtil.Send(socket.getInetAddress().getHostAddress(),4,socketData.getFtpFileName())); 4:直接处理命令
                    }

                    4:直接处理命令
                    if (socketData.getCommand().equals(4)) {
                        log.info("收到节点 " + socketData.getNodeName() + " " + socket.getInetAddress().getHostAddress() + " 的命令 4:直接处理命令");
                        String ftpFileName = socketData.getFtpFileName();
                         downloadFtpFile(ftpFileName);
                        ) {
                            ZipData zipData =  ZipData(file);

                            ) {
                                processZipData(zipData);

                                removeFtpFile(ftpHost,ftpFileName); 
                                SocketUtil.Send(masterIp,1)">                                ZipProcesstime.updateTime();
                            }  {
                                removeFtpFile(ftpHost,1)">
                                RunProcessZipThread thread =  {
                            RunProcessZipThread thread =  RunProcessZipThread(readFtpFileService);
                            thread.start();
                        }
                    }
                }  (Exception e) {
                    log.error("SocketServer while 出错" (Exception e) {
            log.error("SocketServer run 出错" processZipData(ZipData zipData) {
        CountDownLatch countDownLatch = new CountDownLatch(zipData.getHotsoptList().size() + zipData.getTerminfoList().size());

         (HotspotInfoCollected hotspot : zipData.getHotsoptList()) {
            RunnableSendHotspot runnableSendHotspot =  RunnableSendHotspot(kafkaSendService,hotspot,countDownLatch);
            threadPool.submit(runnableSendHotspot);
        }

         (TermInfocharacteristics terminfo : zipData.getTerminfoList()) {
            RunnableSendTerm runnableSendTerm =  RunnableSendTerm(kafkaSendService,terminfo,countDownLatch);
            threadPool.submit(runnableSendTerm);
        }

         {
            countDownLatch.await();
        }  (InterruptedException e) {
            log.error("countDownLatch.await() Failed" renameFtpFile(String oldName,String newName) {
        ftputils ftputils = ftputils.renameFtpFile(oldName,newName);
        }  (Exception e) {
            log.error("RenameFtpFile() Failed" {
            ftputils.disConnection();
        }
    }

     result;
    }

    byte[] downloadFtpFile(String filePath) {
        byte[] result = ;
        ftputils ftputils =  ftputils.downloadFtpFile(filePath);
            int tryCount = 1;
            while (result == null && tryCount < 5) {
                log.info("下载FTP文件 " + filePath + " 失败，尝试再次下载" + tryCount);
                result = ftputils.downloadFtpFile(filePath);
                Thread.sleep(3000);
                tryCount++;
            }
        }  (Exception e) {
            log.error("downloadFtpFile(filePath) Failed"boolean removeFtpFile(String ftpHost,String filePath) {
        boolean result =  ftputils.removeFtpFile(filePath);
        }  (Exception e) {
            log.error("removeFtpFile() Failed" result;
    }

}
View Code

4.RunProcessZipThread类：负责任务的重启


 * 用于重启数据处理任务
 */
class RunProcessZipThread  Thread {

    public RunProcessZipThread(ReadFtpFileService readFtpFileService) {
         readFtpFileService;
    }

     {
            Thread.sleep(20000 (InterruptedException e) {
            e.printstacktrace();
        }

        readFtpFileService.ProcessZip();
    }
}
View Code

实际处理数据的代码：

1.RunnableSendHotspot类：

 com.google.gson.Gson;
 com.suncreate.logback.elasticsearch.metric.DataType;
 com.suncreate.logback.elasticsearch.metric.ProcPhase;
 com.suncreate.logback.elasticsearch.metric.ProcStatus;
 com.suncreate.logback.elasticsearch.metric.SinkType;
 com.suncreate.logback.elasticsearch.util.MetricUtil;
 java.util.HashMap;
 java.util.concurrent.CountDownLatch;

class RunnableSendHotspot implements Runnable {
    final Logger log = LoggerFactory.getLogger(RunnableSendHotspot. HotspotInfoCollected hotspot;

     CountDownLatch countDownLatch;

     RunnableSendHotspot(KafkaSendService kafkaSendService,HotspotInfoCollected hotspot,CountDownLatch countDownLatch) {
        this.hotspot = hotspot;
        this.countDownLatch = countDownLatch;
    }

    @Override
     run() {
        log2ES(ProcPhase.collect.toString(),ProcStatus.suc.toString(),1);
        kafkaSendService.sendHotspotInfoCollected(hotspot);
        countDownLatch.countDown();
    }

     log2ES(String procPhase,String procStatus,Integer count) {
        HashMap<String,Object> logMap;
        logMap = (HashMap<String,Object>) MetricUtil.getMap("wifi_probe","hik"new Gson().toJson(logMap) + "    Count：" + countDownLatch.getCount());
    }
}
View Code

2.RunnableSendTerm类：

class RunnableSendTerm final Logger log = LoggerFactory.getLogger(RunnableSendTerm. TermInfocharacteristics termInfo;

     RunnableSendTerm(KafkaSendService kafkaSendService,TermInfocharacteristics termInfo,1)">this.termInfo = termInfo;
        );
        kafkaSendService.sendTermInfocharacteristics(termInfo);
        countDownLatch.countDown();
    }

     countDownLatch.getCount());
    }
}
View Code

辅助代码：

1.ZipProcesstime类：记录数据处理任务最后活动的时间

 java.util.Date;


 * 记录数据处理任务最后活动的时间
  ZipProcesstime {
    final Logger log = LoggerFactory.getLogger(ZipProcesstime.long time = System.currentTimeMillis();

     updateTime() {
        time = System.currentTimeMillis();
        log.info("ZipProcesstime 已更新，时间戳=" + time);
    }

    long getTime() {
         time;
    }

}
View Code

2.CheckScheduleConfig类：配置一个定时任务用于监控数据处理任务是否存活，如果长时间不存活，则重启任务(实际运行过程中，这种情况未出现过)

 com.suncreate.wifi.task;

 com.suncreate.wifi.service.ZipProcesstime;
 com.suncreate.wifi.service.ReadFtpFileService;
 org.springframework.context.annotation.Configuration;
 org.springframework.scheduling.annotation.EnableScheduling;
 org.springframework.scheduling.annotation.SchedulingConfigurer;
 org.springframework.scheduling.config.ScheduledTaskRegistrar;
 org.springframework.scheduling.support.crontrigger;


 * master节点与worker节点之间的socket通信有可能失败，所以需要一个监控，以重启数据处理任务
 */
@Configuration
@EnableScheduling
class CheckScheduleConfig  SchedulingConfigurer {
    final Logger log = LoggerFactory.getLogger(CheckScheduleConfig.);

    @Value("${checkQuarter}" String checkQuarter;

    @Autowired
     ReadFtpFileService readFtpFileService;

    @Value("${node.name}" String masterName;

    @Override
     configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        taskRegistrar.addTriggerTask(() ->if (!nodeName.equals(masterName)) { master节点不处理数据，跳过
                double sec = (System.currentTimeMillis() - ZipProcesstime.getTime()) / 1000.0;

                log.info("ZipProcesstime 已经 " + sec + " 秒没有更新");
                if (sec > 1800) {
                    log.info("ZipProcesstime 已经长时间没有更新，重启 ProcessZip");
                    readFtpFileService.ProcessZip();
                }
            }

        },triggerContext ->  crontrigger(checkQuarter).nextExecutionTime(triggerContext));
    }
}
View Code

 

====================== 分隔线 ==============================================================================================

目前程序已在现网连续稳定运行一个月

 

====================== 分隔线 ==============================================================================================

2021年5月27日，突然想到一个更加稳定并且容易实现的方法：每台机器上跑的服务，各处理各的，不再相互通信，通过重命名FTP文件的方式实现并发处理，以增强服务稳定性

主要代码如下：

ReadFtpFileService类代码：

 java.util.regex.Pattern;

@Service
 String ftpPassword;

    @Value("${hostname}" String hostname;

     {
            log.info("ProcessZip() 开始" (String ftpFileName : ftpFileList) {
                String newFtpFileName = ;

                if (ftpFileName.endsWith(hostname + ".zip")) {
                    newFtpFileName = ftpFileName;
                }  {
                     (isOriginalFileName(ftpFileName)) {
                        newFtpFileName = ftpFileName.replace(".zip","") + "-" + hostname + ".zip" renameFtpFile(ftpFileName,newFtpFileName);
                        bl) {
                            newFtpFileName = ;
                        }
                    }
                }

                if (newFtpFileName != ) {
                     downloadFtpFile(newFtpFileName);
                    ) {
                        ZipData zipData =  ZipData(file);

                        CountDownLatch countDownLatch =  zipData.getTerminfoList().size());

                         (HotspotInfoCollected hotspot : zipData.getHotsoptList()) {
                            RunnableSendHotspot runnableSendHotspot =  (TermInfocharacteristics terminfo : zipData.getTerminfoList()) {
                            RunnableSendTerm runnableSendTerm =  {
                            countDownLatch.await();
                        }  (InterruptedException e) {
                            log.error("countDownLatch.await() Failed"删除ftp中的数据
                        removeFtpFile(ftpHost,newFtpFileName);
                    }
                }
            }

            log.info("ProcessZip() 结束" (Exception e) {
            log.error("ProcessZip() Failed"
     * 判断FTP文件是否是未改过名的原始文件
     boolean isOriginalFileName(String ftpFileName) {
        String pattern = "^[\\S\\s]*-[0-9]*-[0-9]*-[0-9]*.zip$";
         ftputils.renameFtpFile(oldName,1)"> bl;
    }

     result;
    }

}
View Code

 

====================== 分隔线 ==============================================================================================

2021年6月16日，距第一次提交代码已经有半年了，怎么又要修改了呢？同事说你部署的节点太多，容易出问题。“因为部署的节点多，所以容易出问题”这句话逻辑上对吗？

所以我又有想法了，最好能优化一下，只部署在一台机器上，一个进程搞定。其实几个月前我就怀疑代码中kafka生产者性能不行，但是修改测试其实是麻烦的，以前的同事写的好几个服务都是这样写的，按说不会有什么问题，所以始终没有花时间去验证。

原kafka生产者代码：

KafkaConfig.java：

 com.suncreate.wifi.config;

 org.apache.kafka.clients.producer.ProducerConfig;
 org.apache.kafka.common.serialization.StringSerializer;
 org.springframework.context.annotation.Bean;
 org.springframework.kafka.annotation.EnableKafka;
 org.springframework.kafka.core.DefaultKafkaProducerFactory;
 org.springframework.kafka.core.KafkaTemplate;
 org.springframework.kafka.core.ProducerFactory;

 java.util.Map;

@Configuration
@EnableKafka
 KafkaConfig {
    final Logger log = LoggerFactory.getLogger(KafkaConfig.);

    @Value("${kafka.bootstrap-servers}" String bootstrapServers;

    @Bean
    public ProducerFactory<String,String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String,1)"> producerConfigs() {
        Map<String,Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONfig,bootstrapServers);
        props.put(ProducerConfig.RETRIES_CONfig,"3");
        props.put(ProducerConfig.LINGER_MS_CONfig,"1");
        props.put(ProducerConfig.BATCH_SIZE_CONfig,"65536");
        props.put(ProducerConfig.BUFFER_MEMORY_CONfig,"524288");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONfig,StringSerializer.);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONfig,1)">);
         props;
    }

    @Bean
    public KafkaTemplate<String,1)"> kafkaTemplate() {
        new KafkaTemplate<>(producerFactory());
    }

}
View Code

KafkaSendService.java：

 java.sql.Date;
 java.util.HashMap;

 com.alibaba.fastjson.JSONObject;
 com.google.gson.Gson;

 com.suncreate.wifi.model.DeviceInfo;
 com.suncreate.wifi.model.WifiData;

 org.springframework.kafka.core.ProducerFactory;
 org.springframework.kafka.support.SendResult;
 org.springframework.stereotype.Service;
 org.springframework.util.concurrent.ListenableFuture;
 org.springframework.util.concurrent.ListenableFutureCallback;

@Service
 KafkaSendService {
    final Logger log = LoggerFactory.getLogger(KafkaSendService.);

    @Autowired
    KafkaTemplate<String,1)"> kafkaTemplate;

    @Autowired
     BuildBaseinfo buildBaseinfo;

    @Value("${kafka.topic}" String topic;

     sendHotspotInfoCollected(HotspotInfoCollected data) {
         {
            DeviceInfo di = buildBaseinfo.GetDeviceInfo(data.getCollectionEquipmentLon(),data.getCollectionEquipmentLat());
            di.Verify()) {
                log2ES(ProcPhase.clear.toString(),ProcStatus.fail.toString(),1)">);
                System.out.println("Equipment id=" + data.getCollectionEquipmentid() + " can't get baseinfo";
            }
            WifiData wd =  WifiData();
            wd.setdeviceid(di.getCameraNo());
            wd.setdiscoverMacAddr(data.getApMac().replace('-',':'));
            wd.setStartTime();
            wd.setEndTime();
            wd.setPower(Integer.parseInt(data.getApfieldStrength()));
            wd.setMacAppearTimes(1);
            wd.setIpAddr(di.getIp());
            wd.setLatitude(di.getLatitude());
            wd.setLongitude(di.getLongitude());
            wd.setDataSource("4");  数据来源，1-宇视，2-科达，3-大华，4-海康
            wd.setCaptureTime(new Date(Long.parseLong(data.getCaptureTime()) * 1000));
            wd.setInsertTime( Date(System.currentTimeMillis()));
            wd.setBrand("");
            wd.setXCoordinate(0d);
            wd.setYCoordinate(0d);
            wd.setCaptureType(1);
            wd.setApSSID(data.getApSsid());
            wd.setApEncryptType(data.getEncryptAlgorithmType());
            wd.setApMac("");
            log2ES(ProcPhase.clear.toString(),wd.getIpAddr());
            sendMessage(JSONObject.toJSONString(wd));
        }  (Exception e) {
            log2ES(ProcPhase.clear.toString(),1)">);
            log.error("Failed to clear HotspotInfo" sendTermInfocharacteristics(TermInfocharacteristics data) {
         WifiData();
            wd.setdeviceid(di.getCameraNo());
            wd.setdiscoverMacAddr(data.getMac().replace('-',1)">);
            wd.setPower(Integer.parseInt(data.getTerminalFieldStrength()));
            wd.setMacAppearTimes(1 Date(System.currentTimeMillis()));
            wd.setBrand(data.getBrand());
            wd.setXCoordinate(0d);
            wd.setYCoordinate(0d);
            wd.setCaptureType(0);
            wd.setApSSID(data.getSsidPisition());
            wd.setApEncryptType("");
            wd.setApMac(data.getAccessApMac() == null ? "" : data.getAccessApMac().replace('-',1)">));
            log2ES(ProcPhase.clear.toString(),1)">);
            log.error("Failed to clear TermInfo" sendMessage(String message) {
        ListenableFuture<SendResult<String,String>> future = kafkaTemplate.send(topic,message);
        kafkaTemplate.flush();
        future.addCallback(new ListenableFutureCallback<SendResult<String,String>>() {
            @Override
            void onSuccess(SendResult<String,1)"> result) {
                log2ES("transport",1);
            }

            @Override
             onFailure(Throwable ex) {
                log2ES("transport",1)">);
                log.error("Failed to send to kafka" Gson().toJson(logMap));
    }

     Gson().toJson(logMap));
    }
}
View Code

优化后的kafka生产者代码：

KafkaProducer.java(注意和KafkaConfig.java对比)：

 javax.annotation.postconstruct;
 java.util.Map;
 java.util.Random;

@Service
 KafkaProducer {
    final Logger log = LoggerFactory.getLogger(KafkaProducer. String bootstrapServers;

    private Map<Integer,KafkaTemplate<String,1)"> map;

    private Random random =  Random();

    int producerCount = 10;

    @postconstruct
     init() {
        map = ();
        int i = 0; i < producerCount; i++) {
            map.put(i,1)">new KafkaTemplate<>((getProducerConfigs())));
        }
    }

    private Map<String,1)"> getProducerConfigs() {
        Map<String,1)"> props;
    }

     getKafkaTemplate() {
         map.get(random.nextInt(producerCount));
    }

}
View Code

KafkaSendService.java：

);

    @Autowired
    KafkaProducer kafkaProducer;

    @Autowired
     sendMessage(String message) {
        KafkaTemplate<String,String> kafkaTemplate = kafkaProducer.getKafkaTemplate();
        ListenableFuture<SendResult<String,1)"> Gson().toJson(logMap));
    }
}
View Code

修改后的结果：

单个程序进程生产kafka消息的速度从300多每秒提高到了3000多每秒，性能提升了一个数量级，因为我new了10个KafkaTemplate。但是我想不明白的是，按说DefaultKafkaProducerFactory应该是有连接池的啊，为什么只new一个DefaultKafkaProducerFactory性能这么差呢？

贴一下git记录(master分支，socket分支的记录就不贴了)，这么单纯的一个小需求，前前后后花费了多少心思，标题我就不改了，不然过于夸张(新标题：本来一个星期工作量的工作，我干了半年)：

 

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。

相关推荐

Blazor 调用 Clipboard API 读写剪贴板数据
目录简介使用JS互操作使用ClipLazor库创建项目使用方法简单测试参考链接 简介 Clipboard API 是一种允许网页读取剪贴板数据或向其中写入数据的API，主要有两个方法： navigator.clipboard.writeText() ：用于将文本写...
C# 轻量级 ORM 框架 NPoco 的简单应用
目录简介快速入门安装 NuGet 包实体类User数据库类DbFactory增删改查InsertSelectUpdateDelete总结 简介 NPoco 是 PetaPoco 的一个分支，具有一些额外的功能，截至现在 github 星数 839。NPoco 中文资料没多少，我是...
C#简单配置类及数据绑定
本文实现一个简单的配置类，原理比较简单，适用于一些小型项目。主要实现以下功能：保存配置到json文件、从文件或实例加载配置类的属性值、数据绑定到界面控件。&#xA;一般情况下，项目都会提供配置的设置界面，很...
C#获取Description特性的扩展类
C#中Description特性主要用于枚举和属性，方法比较简单，记录一下以便后期使用。 扩展类DescriptionExtension代码如下： using System; using System.ComponentModel; using System.Reflection; /// &lt;
C#实现软件开机自启动（不需要管理员权限）
[TOC] # 原理简介 本文参考[C#/WPF/WinForm/程序实现软件开机自动启动的两种常用方法](https://blog.csdn.net/weixin_42288432/article/details/120059296)，将里面中的第一种方法做了封装成**AutoStart**类，使
C#爬取动态网页上的信息：B站主页
目录简介获取 HTML 文档解析 HTML 文档测试补充：使用 CSS 选择器参考文章 简介 动态内容网站使用 JavaScript 脚本动态检索和渲染数据，爬取信息时需要模拟浏览器行为，否则获取到的源码基本是空的。 本文使用的爬取...
C#项目管理引用的dll文件
这篇文章主要简单记录一下C#项目的dll文件管理方法，以便后期使用。 设置dll路径 参考C#开发奇技淫巧三：把dll放在不同的目录让你的程序更整洁中间的 方法一：配置App.config文件的privatePath ： &lt;runtime&a...
C# 常用特性(Attribute)
使用特性，可以有效地将元数据或声明性信息与代码（程序集、类型、方法、属性等）相关联。 将特性与程序实体相关联后，可以在运行时使用反射这项技术查询特性。&#xA;在 C# 中，通过用方括号 ([]) 将特性名称括起...
C#事件总线
事件总线是对发布-订阅模式的一种实现，是一种集中式事件处理机制，允许不同的组件之间进行彼此通信而又不需要相互依赖，达到一种解耦的目的。&#xA;EventBus维护一个事件的字典，发布者、订阅者在事件总线中获取...
C#中使用jieba.NET、WordCloudSharp制作词云图
词云”由美国西北大学新闻学副教授、新媒体专业主任里奇·戈登（Rich Gordon）于2006年最先使用，是通过形成“关键词云层”或“关键词渲染”，对文本中出现频率较高的“关键词”的视觉上的突出。词云图过滤掉大量的文...
    pythonJavaScriptjavaHTMLreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-bootreact-native