gitweixin
  • 首页
  • 小程序代码
    • 资讯读书
    • 工具类
    • O2O
    • 地图定位
    • 社交
    • 行业软件
    • 电商类
    • 互联网类
    • 企业类
    • UI控件
  • 大数据开发
    • Hadoop
    • Spark
    • Hbase
    • Elasticsearch
    • Kafka
    • Flink
    • 数据仓库
    • 数据挖掘
    • flume
    • Kafka
    • Hive
    • shardingsphere
    • solr
  • 开发博客
    • Android
    • php
    • python
    • 运维
    • 技术架构
    • 数据库
  • 程序员网赚
  • bug清单
  • 量化投资
  • 在线查询工具
    • 去行号
    • 在线时间戳转换工具
    • 免费图片批量修改尺寸在线工具
    • SVG转JPG在线工具

月度归档7月 2025

精品微信小程序开发门户,代码全部亲测可用

  • 首页   /  2025   /  
  • 7月
前端 7月 27,2025

如何在Chrome中设置启动时自动打开多个默认网页

我喜欢在同时用多个国内的大模型网站,每次启动chrome浏览器,一个个启动觉得麻烦。今天发现其实可以设置chrome启动时默认自动启动。

在Google Chrome中设置启动时自动打开多个默认网页,可通过以下步骤实现:


​​一、进入启动时设置​​

  1. ​​打开Chrome浏览器​​,点击右上角的 ​​三个垂直点​​(菜单按钮),选择 ​​“设置”​​ 。
  2. 在左侧菜单栏中找到 ​​“起始页面”​​ 选项并点击进入。

​​二、添加多个默认网页​​

  1. ​​选择启动行为​​
    在“启动时”页面中,选择 ​​“打开特定网页或一组网页”​ 。
  2. ​​添加网页​​
    • 点击 ​​“添加新网页”​​ 或 ​​“设置网页”​​ 按钮(不同版本名称可能略有差异)。
    • 在弹出窗口中输入第一个网页的完整网址(如 https://www.google.com),点击 ​​“添加”​​ 或 ​​“确定”​​ 。
    • 重复上述步骤,添加其他需要默认打开的网页 。
  1. ​​使用当前标签页​​(可选)
    如果希望将当前打开的标签页设为默认网页,可点击 ​​“使用当前网页”​​ 按钮 。

​​三、保存并验证​​

  • 完成添加后,关闭设置页面,​​Chrome会自动保存配置​​。
  • 重启浏览器,检查是否成功打开所有设置的网页 5。
作者 east
未分类 7月 7,2025

spark内存溢出怎样区分是软件还是代码原因

在 Spark 内存溢出(OutOfMemoryError) 的排查中,首先要区分是软件层配置不当还是代码逻辑导致内存占用异常。下面提供一个系统性的排查思路,帮助你判断是软件问题还是代码问题:


一、初步排查方向

维度判断问题来源
内存是否普遍不够软件配置问题
某任务、某节点频繁 OOM代码逻辑问题
所有任务都报相似 OOM软件问题可能更大
Task 在运行中内存缓慢增长代码中数据缓存或累积导致
提升 executor 内存后仍然溢出代码逻辑问题可能更大

二、具体排查步骤

查看 OOM 日志信息

  • 报错关键词:java.lang.OutOfMemoryError
  • 看看是:
    • GC overhead limit exceeded(垃圾回收过于频繁,内存紧张)
    • Java heap space(堆内存不足)
    • Direct buffer memory(off-heap内存问题)

这些信息能帮助我们快速定位是哪一块内存出了问题。


判断是否是 软件资源配置不足

表现:

  • executor 内存(如 --executor-memory 2g)太小
  • executor 数量太多,总内存超机器物理内存
  • shuffle spill 频繁但磁盘也不够用
  • 没有配置 spark.memory.fraction 或设置不合理(默认 0.6)

排查方式:

  • 适当提高 executor memory 和 driver memory 看是否解决问题;
  • 检查 spark.sql.shuffle.partitions 设置是否过大;
  • 观察是否所有节点都在 OOM;
  • 查看 GC 日志:是否频繁 full GC,停顿时间过长。

如果内存调高之后就不报错,说明是配置问题。


判断是否是 代码导致内存异常

典型表现:

  • 某些 task 内存快速膨胀,例如:
    • groupByKey() 或 reduceByKey() 输入太大;
    • 使用 collect() 导致 driver 内存不足;
    • 使用 rdd.cache() 缓存了大量数据;
    • Dataset 操作中存在 mapPartitions 内部收集过多数据;
    • 算子嵌套太深,堆栈过长导致栈溢出;
    • 广播变量(broadcast)太大;
    • join 前未分区或数据倾斜严重,单分区数据过大。

排查方式:

  • 查看具体是哪些 stage/task 报错(Spark UI -> Executors 或 Stages);
  • 查看是否只有少数 executor OOM;
  • 结合源码或业务逻辑,排查数据倾斜、缓存、collect 等问题;
  • 是否使用了 checkpoint?是否放在了合适位置?

三、建议工具与方法

Spark UI 重点页面

  • Stages 页面:查看哪个 stage 出现问题,是否数据倾斜;
  • Executors 页面:某个 executor 使用内存明显偏高;
  • SQL 页面:看哪个 SQL 任务导致内存暴涨;
  • Storage 页面:查看缓存的数据占用情况。

GC 日志分析

  • 如果是 JVM 内存泄漏,建议打开 GC 日志分析:--conf "spark.executor.extraJavaOptions=-XX:+PrintGCDetails -XX:+PrintGCTimeStamps -Xloggc:gc.log"

总结:如何判断

问题特征更可能是…
所有 executor 都 OOM,提升内存后恢复软件配置不合理
单个 executor OOM,内存增长异常快代码问题
使用 collect/cache/join/groupByKey 导致爆炸代码问题
executor memory 明显低于数据体量软件问题
分区不合理导致单个 task 内存爆炸代码问题
作者 east
Spark 7月 7,2025

Spark Shuffle的优化

Spark Shuffle 是连接不同 Stage 的关键环节,也是 Spark 作业中最容易产生性能瓶颈的地方之一。它涉及大量磁盘 I/O、网络传输和内存使用。优化 Shuffle 对提升作业性能和稳定性至关重要。以下是一些关键的 Spark Shuffle 优化策略:

核心目标: 减少 Shuffle 数据量、降低 I/O 开销、提升网络传输效率、优化内存使用、处理数据倾斜。

主要优化策略:

  1. 减少 Shuffle 数据量 (根本之道):
    • Map-Side 预聚合/Combining: 在 Shuffle 写之前,尽可能在 Mapper 端对数据进行聚合(combineByKey, reduceByKey, aggregateByKey)。这能显著减少需要传输的键值对数量。优先使用 reduceByKey/aggregateByKey 而不是 groupByKey。
    • 选择更高效的算子: reduceByKey 优于 groupByKey + reduce;treeReduce/treeAggregate 在聚合深度大时优于 reduce/aggregate(减少网络轮次)。
    • 过滤数据: 尽早使用 filter 过滤掉不需要参与 Shuffle 的数据。
    • 列裁剪: 只选择 Shuffle 后真正需要的列(尤其是在使用 DataFrame/Dataset API 时)。避免传输整个对象。
    • 避免不必要的 distinct: distinct 会触发 Shuffle。考虑是否可以用其他方式(如 Map 端去重)或在更小的数据集上使用。
    • 广播小表: 如果 Join 操作中一个表很小,使用 broadcast 将其分发到所有 Executor,避免大表 Shuffle。spark.sql.autoBroadcastJoinThreshold 控制自动广播的阈值。
  2. 优化 Shuffle 写:
    • 调整 spark.shuffle.file.buffer: 增加 Shuffle 写过程中每个分区文件的内存缓冲区大小(默认 32K)。增大此值(如 64K, 128K)可以减少磁盘 I/O 次数,但会增加内存压力。需权衡。
    • 调整 spark.shuffle.spill.diskWriteBufferSize: 增大溢出到磁盘时使用的缓冲区大小(默认 1024K),同样减少写磁盘次数。
    • 启用 spark.shuffle.unsafe.file.output.buffer: 对于使用 Tungsten 的 Shuffle,设置这个直接内存缓冲区大小(默认 32K),作用类似 spark.shuffle.file.buffer。
    • 优化 spark.shuffle.spill: 确保 spark.shuffle.memoryFraction 或 spark.memory.fraction 设置合理,为 Shuffle 分配足够内存,减少溢出次数。监控 GC 和溢出情况。
    • 选择高效的 Shuffle 实现:
      • Sort Shuffle (sort): Spark 1.2+ 的默认方式。对每个分区排序并合并小文件。通常最稳定高效。
      • Tungsten-Sort (tungsten-sort): 基于 Project Tungsten,使用堆外内存和更高效的编码。在 Spark 1.4+ 可用,有时性能更好(尤其处理原始类型时)。通常当 spark.shuffle.manager=sort 且满足条件(序列化器支持重定位、非聚合 Shuffle 等)时会自动使用。
    • 文件合并 (spark.shuffle.consolidateFiles): (在较新 Spark 版本中已被优化或默认行为替代) 在老版本中启用此选项可以让多个 Reduce Task 共享同一个 Mapper 输出的合并文件,减少小文件数量。新版本 Sort Shuffle 本身已优化文件数量。
  3. 优化 Shuffle 读:
    • 调整 spark.reducer.maxSizeInFlight: 控制每次 Reduce Task 从远程 Executor 拉取数据的最大大小(默认 48M)。增大此值(如 96M)可以提高吞吐量,但会增加内存使用。需监控网络和内存。
    • 调整 spark.shuffle.io.maxRetries 和 spark.shuffle.io.retryWait: 网络不稳定时,增加重试次数和等待时间以避免 Fetch Failed 错误。但过度重试会拖慢作业。
    • 调整 spark.shuffle.io.numConnectionsPerPeer: 如果集群节点很多且网络是瓶颈,适当增加此值(默认 1)可以提升并发连接数。
    • 启用 spark.shuffle.compress: (默认开启) 压缩 Shuffle 数据(写和读)。使用高效的压缩算法:
      • spark.io.compression.codec: 推荐 lz4(速度快)或 zstd(压缩率高,速度也不错)。snappy 是默认值,也是不错的选择。避免使用低效的 lzf。
    • 调整 spark.shuffle.service.enabled: 启用 External Shuffle Service。这允许 Executor 在退出后(如动态资源分配下)Shuffle 文件仍能被访问,提高稳定性。通常在生产环境推荐启用。
  4. 调整分区数量:
    • 关键参数 spark.sql.shuffle.partitions (SQL/DataFrame/Dataset): 控制 Shuffle 后(如 Join, Aggregation)的分区数(默认 200)。这是最重要的优化点之一。
      • 分区过少: 每个分区数据量过大 -> 可能导致 OOM、GC 时间长、Task 执行慢、无法充分利用集群资源。
      • 分区过多: 每个分区数据量过小 -> Task 调度开销增大、产生大量小文件、网络请求次数增多(影响 Shuffle 读)。
      • 如何调整: 根据集群总核心数和数据量估算。经验值通常是集群总核心数的 2-3 倍。例如,集群有 100 个 Executor,每个 4 核,总核心数 400,可设置为 800 – 1200。需要根据实际作业数据量和执行情况(查看 Spark UI 中的 Shuffle Read Size/Records)反复调整测试。 数据量极大时可设置更高。
    • RDD API: repartition/coalesce: 在 RDD 操作中显式控制分区数。
  5. 处理数据倾斜:
    • 识别倾斜: 通过 Spark UI 查看各 Stage 中 Task 的执行时间分布。执行时间显著长于其他 Task 的通常处理了倾斜的分区。查看 Shuffle Read Size 差异。
    • 缓解策略:
      • 过滤倾斜 Key: 如果极少数倾斜 Key 可以单独处理或过滤掉。
      • 加盐打散: 给倾斜的 Key 添加随机前缀(扩容),在局部聚合后去掉前缀再全局聚合。
      • 提高并行度: 增加 spark.sql.shuffle.partitions,让倾斜 Key 分散到更多分区(对于单个 Key 数据量特别大的情况效果有限)。
      • 使用 skew join 优化 (Spark 3.0+ AQE): 自适应查询执行 (AQE) 能自动检测倾斜 Join 并将倾斜的分区分裂成更小的子分区进行处理。强烈推荐启用 AQE (spark.sql.adaptive.enabled=true)。
      • 特定算子: reduceByKey 比 groupByKey 更能容忍一定程度的倾斜(因为 Map 端合并了)。对于 Join 倾斜,考虑广播小表或使用 SortMergeJoin/ShuffleHashJoin 的替代方案。
  6. 利用自适应查询执行:
    • 启用 AQE (spark.sql.adaptive.enabled=true): Spark 3.0+ 的核心优化特性。
      • 动态合并 Shuffle 分区: 根据 Shuffle 后实际数据大小,自动将过小的分区合并,避免大量小 Task 的开销。
      • 动态调整 Join 策略: 在运行时根据统计信息将 SortMergeJoin 切换为 BroadcastJoin(如果发现小表符合广播条件)。
      • 动态优化倾斜 Join: 自动检测并处理 Shuffle Join 中的数据倾斜问题。
    • 相关参数: spark.sql.adaptive.coalescePartitions.enabled, spark.sql.adaptive.coalescePartitions.minPartitionNum, spark.sql.adaptive.advisoryPartitionSizeInBytes, spark.sql.adaptive.skewJoin.enabled, spark.sql.adaptive.skewJoin.skewedPartitionFactor, spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes 等。强烈建议在生产环境中启用并适当配置 AQE。
  7. 硬件与集群配置:
    • 使用 SSD: 将 Shuffle 溢出目录 (spark.local.dir) 配置到 SSD 磁盘上能极大提升 Shuffle 写/读的 I/O 性能。这是非常有效的优化。
    • 充足内存: 确保 Executor 有足够内存(spark.executor.memory),特别是当 spark.memory.fraction 分配给 Storage 和 Execution(包含 Shuffle)时。减少溢出到磁盘的次数。
    • 高速网络: 万兆甚至更高带宽、低延迟的网络能显著加速 Shuffle 数据传输。
    • 合理 CPU 核数: 避免单个 Executor 分配过多 CPU 核(如 > 5),因为多个 Task 竞争磁盘/网络 I/O 可能成为瓶颈。通常每个 Executor 配置 3-5 核是一个较好的平衡点。

优化流程建议:

  1. 监控与诊断: 使用 Spark Web UI 仔细分析作业运行情况。重点关注:
    • Shuffle Read/Write 的总数据量和在各 Stage/Task 上的分布。
    • Task 的执行时间分布(识别倾斜)。
    • GC 时间。
    • 是否有溢出到磁盘 (Spill (Memory), Spill (Disk))。
    • Shuffle Write Time / Shuffle Read Time。
    • 日志中的 WARN/ERROR 信息(如 FetchFailed, OOM)。
  2. 定位瓶颈: 根据监控信息判断是数据量太大、I/O 慢、网络慢、内存不足还是数据倾斜。
  3. 应用策略: 针对性地选择上述优化策略进行调整。优先考虑减少数据量和调整分区数。
  4. 迭代测试: 修改配置或代码后,在小规模数据或测试集群上运行测试,观察效果。每次最好只修改一个主要配置,以便定位效果。
  5. 利用 AQE: 确保在 Spark 3.x 环境中启用并配置好 AQE,它能自动处理很多棘手的优化问题(小分区合并、倾斜 Join)。

Spark Shuffle 优化原则

类别优化建议
算子选择使用 reduceByKey 代替 groupByKey,选择合适的 Join
分区策略控制合理的并发度、分区数,避免极端数据倾斜
参数调优内存、缓冲区、网络传输参数细致设置
数据倾斜通过打散、随机 key、局部聚合等方式规避热点 key
AQE开启 Spark SQL 的自适应执行,自动处理 Join/倾斜问题
文件合并启用 consolidateFiles 降低磁盘负担
作者 east
嵌入式 7月 3,2025

MQTT完全解析和实践

第1章:MQTT的江湖地位——轻量级通讯的王者

你可能听过HTTP、WebSocket、CoAP……但如果你做的是物联网(IoT),MQTT 这个缩写,几乎避无可避。为啥?因为它轻、它快、它省电、它设计初衷就是为带宽差、电量少、网络不稳的设备服务的。

它的全称是 Message Queuing Telemetry Transport。虽然名字里带个“消息队列”,但和 Kafka 这类“正儿八经”的消息队列还真不是一路人。MQTT 更像是推送消息的邮差,而不是存储消息的邮局。

MQTT的核心机制:Pub/Sub模型

MQTT的核心机制是 发布/订阅(Publish/Subscribe),不是客户端直接连接到另一个客户端,而是由一个“中间人”——Broker(消息中介服务器)进行转发。

打个比方:

  • 发布者 = 在微信群里发言的人。
  • 订阅者 = 设置了关键词提醒的群友。
  • Broker = 微信服务器。

你只管发,Broker会帮你广播。订阅者只关心自己订阅的话题(Topic),不会管群里还有多少人在水群。

为什么物联网设备爱它?

MQTT从设计之初就有几个“讨喜”的特点:

  • 超轻量协议头:最小只有2个字节,远比HTTP动不动就几十上百字节要苗条。
  • 支持断线重连:掉线不怕,Client ID在,Broker会记住你。
  • QoS保证:支持三种消息服务质量(0、1、2),你要多可靠,它就有多可靠。
  • 持久会话:不在线也能留消息,回来还能收。
  • 订阅灵活:支持通配符(+、#),批量订阅根本不是问题。

MQTT vs HTTP:别再让设备“重型通信”了!

特性MQTTHTTP
连接方式长连接(保持)短连接(每次请求都重建)
消息头大小极小(2字节起)较大(几十到上百字节)
传输模式推送(服务器主动发)拉取(客户端主动要)
实时性高中低
能耗低高

用HTTP做物联网通讯,就像拿压路机送外卖——能用,但太浪费。


第2章:Topic不是“话题”,是你通讯的地盘

别被中文“话题”这个词骗了,MQTT里的Topic,是设备之间交流的“频道”。每个客户端只要订阅了相同的Topic,就能收到发布在这个Topic上的消息。

MQTT的Topic是层级结构,用斜杠(/)分隔,例如:

smartfarm/sensor/temperature
home/livingroom/light
factory/line1/robotarm/position

你可以把它想成文件路径,每一级都有它的意义。

通配符的魔力

MQTT支持两种通配符:

  • +:单级通配符,匹配某一级的任意名称。
  • #:多级通配符,匹配后面所有层级。

举个例子:

  • home/+/light 可以匹配 home/livingroom/light 和 home/kitchen/light
  • home/# 可以匹配 home/bedroom/fan/status、home/garden/temperature 等等

但注意哦,Topic是区分大小写的,还不能用空格,也别搞中文进去(虽然UTF-8可以,但兼容性差)。

Topic不是你想用就能用的

有些Broker会做Topic的访问控制。比如:

  • /system/# 是系统专用的Topic,不让你乱动
  • 公司级平台会配置每个设备只能订阅/发布某些Topic

所以设计MQTT Topic结构时,一定要清晰、分层、可扩展。

一个实用建议:用产品线/设备类型/功能做分类,例如:

factory/productA/temp_sensor01/data
factory/productA/temp_sensor01/command

第3章:QoS,决定你消息的“命运等级”

QoS,全称 Quality of Service(服务质量),是MQTT里一个超级关键的参数。决定了消息传递的“可靠度”。你可以根据场景选不同的级别:

QoS 0:至多一次(At most once)

  • 消息最多发送一次,不确认,不重发。
  • 最快,也最省资源,但可能丢消息。
  • 适用场景:温湿度、心跳包、实时状态上报。

QoS 1:至少一次(At least once)

  • 保证消息至少送达一次,可能重复送达。
  • 接收端需做幂等处理(不要重复处理同一条消息)。
  • 适用场景:开关指令、报警信息。

QoS 2:只有一次(Exactly once)

  • 复杂握手流程,确保消息只到达一次。
  • 网络差也能保证不重复、不丢失。
  • 适用场景:金融交易、关键配置变更。

但请记住:QoS不是越高越好。越高开销越大,设备负载也上升。合理权衡最重要。

在某些低功耗设备上,QoS 0 + 断电重发机制可能比QoS 1/2更划算。


第4章:Broker的心脏——选型与部署全指南

MQTT的灵魂是 Broker,选不好Broker,你的系统就像搭在沙子上的房子。主流Broker有几个大头:

Broker语言特点
MosquittoC轻量、开源、易部署,适合嵌入式、单机测试
EMQXErlang高并发、分布式、企业级能力强,支持集群与规则引擎
VerneMQErlang支持百万连接,企业级,社区活跃
HiveMQJava商用为主,配套工具完善,Web可视化强

你可以根据以下维度进行选择:

  • 性能要求:连接数多少、消息频率多高?
  • 部署环境:云部署还是边缘设备?
  • 安全需求:是否需要鉴权、TLS?
  • 扩展性:未来是否要接Kafka、MySQL、InfluxDB等?

下面这张表,直接拍板也没问题:

场景推荐Broker
小型智能家居项目Mosquitto
工厂边缘网关EMQX(部署在边缘计算设备)
城市级监控平台HiveMQ或EMQX集群
开发调试用Mosquitto + MQTT.fx/MQTT Explorer

第5章:连接的那一刻——MQTT客户端的秘密仪式

在物联网世界里,设备和Broker之间的第一次“见面礼”尤为重要。连接流程不仅要迅速稳定,还得包含身份验证、遗言声明和KeepAlive协议。

Connect报文:你的身份证明

每个MQTT客户端连接时都必须发送一个 CONNECT报文。这个报文里藏着很多关键信息:

  • Client ID:你的身份证,Broker用它来识别你。
  • Username & Password(可选):用于鉴权认证。
  • Clean Session:告诉Broker是否保存你的“上次对话记录”。
  • Will Message(遗嘱消息):掉线时Broker要不要帮你发一条“我挂了”的通告?
  • Keep Alive:告诉Broker多久没动静就要来ping我一下。

Client ID务必唯一,不然你和别的设备撞了名字,Broker会直接踢掉其中一个(或者两个全踢)。

遗嘱消息:设备断线的“最后遗言”

Will Message 是MQTT最“人性化”的设计之一。

假设你的设备突然断电或炸机,还没来得及优雅地发送离线通知——Broker就会代为转达一条“遗言”,提前设置好就行。

{
  "topic": "device/monitor01/status",
  "payload": "offline",
  "qos": 1,
  "retain": true
}

用它可以实现故障自动上报、设备状态感知、掉线告警等等,非常实用。

KeepAlive机制:别让你“假死”

Keep Alive 是一个时间间隔(单位秒),告诉Broker:

“如果我这段时间内没说话,你主动 ping 我一下。”

如果客户端没回应,Broker就会认为你掉线,然后把你踢出聊天室。这个机制能有效防止网络“假死”——其实你早没连上了,但系统以为你还活着。

建议设置在60秒左右,过短会增加网络开销,过长可能延迟感知掉线。

第6章:用代码点亮通讯——MQTT客户端实战开发

纸上得来终觉浅,得把代码跑一遍才算数。本章我们不谈理论,直接上手,从最常见的三类客户端出发:Java(Paho)、嵌入式C(ESP8266)、Node.js(JavaScript),逐一踩坑逐一解决。


6.1 Java + Eclipse Paho:稳定老将出马

Paho 是 Eclipse 基金会推出的 MQTT 客户端库,支持 Java、Python、JavaScript 等多语言实现。

✅ 添加依赖(以 Maven 为例):

<dependency>
  <groupId>org.eclipse.paho</groupId>
  <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  <version>1.2.5</version>
</dependency>

✅ 编写客户端代码:

MqttClient client = new MqttClient("tcp://broker.emqx.io:1883", "demoClientId");
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(true);
options.setUserName("user");
options.setPassword("pass".toCharArray());

client.setCallback(new MqttCallback() {
    public void connectionLost(Throwable cause) {
        System.out.println("连接断了:" + cause.getMessage());
    }
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println("收到消息:" + message.toString());
    }
    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("消息投递成功");
    }
});

client.connect(options);
client.subscribe("demo/topic", 1);
client.publish("demo/topic", new MqttMessage("hello mqtt".getBytes()));

⚠️ 小提示:

  • Client ID 请保持唯一,不然连接会被踢掉。
  • 不建议每次发消息都断连重连,连接是宝贵资源!
  • 异步连接版本是 MqttAsyncClient,适合UI线程中避免卡顿。

6.2 ESP8266 + PubSubClient:嵌入式的王道

Arduino 和 ESP8266 圈子里,用得最多的就是 PubSubClient。轻巧、稳定、开源。

✅ 安装库:

通过 Arduino IDE 管理库,搜索 PubSubClient 并安装即可。

✅ 示例代码:

#include <ESP8266WiFi.h>
#include <PubSubClient.h>

const char* ssid = "yourSSID";
const char* password = "yourPASS";
const char* mqtt_server = "broker.emqx.io";

WiFiClient espClient;
PubSubClient client(espClient);

void setup_wifi() {
  delay(10);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500); Serial.print(".");
  }
  Serial.println("WiFi connected");
}

void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("Message arrived ["); Serial.print(topic); Serial.print("] ");
  for (int i = 0; i < length; i++) Serial.print((char)payload[i]);
  Serial.println();
}

void reconnect() {
  while (!client.connected()) {
    if (client.connect("ESP8266Client")) {
      client.subscribe("esp/test");
    } else {
      delay(5000);
    }
  }
}

void setup() {
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, 1883);
  client.setCallback(callback);
}

void loop() {
  if (!client.connected()) reconnect();
  client.loop();
}

⚠️ 常见问题:

  • 有些免费的公共 Broker 不支持遗嘱机制或 QoS 2,要验证前先查清楚。
  • ESP8266 内存有限,不要滥用 String 对象。

6.3 Node.js + MQTT.js:快速原型的神器

JavaScript 就是快,上手容易,调试方便。MQTT.js 是目前最主流的 Node MQTT 客户端。

✅ 安装依赖:

npm install mqtt

✅ 示例代码:

const mqtt = require('mqtt')
const client  = mqtt.connect('mqtt://broker.emqx.io')

client.on('connect', () => {
  console.log('连接成功')
  client.subscribe('node/topic', (err) => {
    if (!err) {
      client.publish('node/topic', 'Hello MQTT from Node.js')
    }
  })
})

client.on('message', (topic, message) => {
  console.log(`收到消息:${topic} -> ${message.toString()}`)
})

⚠️ 小贴士:

  • 如果用在 Electron 中开发可视化工具,MQTT.js 与 WebSocket Broker 配合非常完美。
  • 遇到 ECONNREFUSED 记得检查防火墙、端口是否开放。

第7章:把门锁好——MQTT安全机制的全副武装

物联网是个“草台班子”?可别这样说,安全性才是成败的关键。只要设备连上了公网,就必须考虑安全问题,不然哪天就被“黑客甲”把你家灯全开了。

这一章我们来认真讨论:MQTT如何防火防盗防偷窥——三层防护:身份认证、传输加密、权限控制。


7.1 用户名 + 密码:基础中的基础

大多数MQTT Broker都支持基于用户名和密码的简单鉴权。

Broker 端配置方式:

  • Mosquitto 使用 mosquitto_passwd 命令生成密码文件:
mosquitto_passwd -c /etc/mosquitto/passwd user1

然后在配置文件里加入:

password_file /etc/mosquitto/passwd
auth_plugin allow_anonymous false
  • EMQX 可以通过 Dashboard 或配置文件添加认证用户。

客户端连接示例(Java):

options.setUserName("user1");
options.setPassword("secret".toCharArray());

✅ 优点:实现简单,快速上手。 ❌ 缺点:密码泄露风险高,尤其是没有加密的传输通道。

所以,用户名密码只是“地基”,加密通道才是地锁。


7.2 TLS加密传输:不能让别人偷听

要让黑客无从下嘴,传输过程必须加密。MQTT天然支持TLS/SSL。

最常见两种方式:

  • 使用自签名证书(便宜,但不被浏览器信任)
  • 使用Let’s Encrypt等CA机构签发的证书(更靠谱)

Mosquitto 示例配置:

listener 8883
cafile /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key
require_certificate false
tls_version tlsv1.2

客户端连接(Node.js):

const fs = require('fs')
const mqtt = require('mqtt')

const options = {
  host: 'broker.example.com',
  port: 8883,
  protocol: 'mqtts',
  ca: fs.readFileSync('./ca.crt'),
  cert: fs.readFileSync('./client.crt'),
  key: fs.readFileSync('./client.key')
};

const client = mqtt.connect(options);

⚠️ 注意:

  • TLS连接比明文连接更慢一点,资源占用也高。
  • 有些老旧设备(尤其是STM32类)TLS支持并不友好,可能需要硬件加速或专门适配。

7.3 ACL权限控制:我能发什么?我能订啥?

光有认证还不够,还得有“授权”。这就是 ACL(Access Control List)的角色。

举个例子:你不希望普通用户能订阅别人的摄像头画面吧?

ACL 可以精细到每个客户端的 订阅与发布权限:

Mosquitto 示例(aclfile):

user user1
publish topic home/user1/#
subscribe topic home/user1/#

user admin
publish topic #
subscribe topic #

EMQX 的 ACL 可以接 MySQL、Redis、JWT 等多种后端,非常灵活。

⚠️ 提醒:权限规则建议配合 Client ID 或用户名,否则容易被伪造身份钻空子。


7.4 防止DoS攻击:限流和最大连接数

物联网平台很容易被低成本刷连接刷消息搞挂。所以生产环境务必做下面几件事:

  • 设置客户端连接最大数、最大消息速率:
max_connections 10000
max_inflight_messages 20
message_size_limit 1MB
  • 对异常客户端(频繁掉线重连)封禁IP或限制连接频率。
  • Broker 支持接入防火墙策略或旁路 WAF (例如部署在Nginx后)。
作者 east
Hive 7月 1,2025

hive sql行转列,列转行sql的实例

在 Hive SQL 中,行转列和列转行是常见的数据转换操作。以下是详细说明和示例:

一、行转列(多行转多列)

场景:将同一分组下的多行数据合并为单行,不同值转为新列(如学生各科成绩横向展示)。

数据示例 (student_scores):

student_idsubjectscore
1Math90
1English85
2Math92

目标(每个学生一行,科目成绩转为列):

student_idMathEnglish
19085
292NULL

SQL 实现:

SELECT
  student_id,
  MAX(CASE WHEN subject = 'Math' THEN score END) AS Math,
  MAX(CASE WHEN subject = 'English' THEN score END) AS English
FROM student_scores
GROUP BY student_id;

关键点:

  1. 使用 CASE WHEN 按条件生成新列
  2. 用聚合函数(如 MAX/MIN/SUM)确保单行输出
  3. 必须配合 GROUP BY 分组字段

二、列转行(多列转多行)

场景:将单行的多列数据拆分为多行(如将横向科目成绩转为纵向记录)。

数据示例 (student_scores_wide):

student_idMathEnglish
19085
292NULL

目标(恢复为行式存储):

student_idsubjectscore
1Math90
1English85
2Math92
2EnglishNULL

SQL 实现:

SELECT student_id, 'Math' AS subject, Math AS score 
FROM student_scores_wide
UNION ALL
SELECT student_id, 'English' AS subject, English AS score 
FROM student_scores_wide;

优化方案(使用 LATERAL VIEW explode(),Hive 特有):

SELECT student_id, subject, score
FROM student_scores_wide
LATERAL VIEW explode(
  map(   -- 构建科目-分数的键值对
    'Math', Math,
    'English', English
  )
) subjects AS subject, score;

关键点:

  1. UNION ALL 适合列数量固定的场景
  2. LATERAL VIEW + explode(map) 动态处理多列
  3. 注意处理 NULL 值(Hive 中 NULL 会正常保留)

三、高级用法:动态行列转换

当列名不固定时(如新增科目),需结合 Hive 动态 SQL 或预处理:

  1. 行转列动态化:
    使用 collect_list + 拼接字符串,再通过 UDF 解析(较复杂,通常需借助外部脚本生成 SQL)。
  2. 列转行动态化:
    通过 Hive 元数据获取列名,动态生成 UNION ALL 语句(示例略,需依赖 Hive 变量或编程接口)。

总结对比

操作核心方法适用场景
行转列CASE WHEN + 聚合函数 + GROUP BY固定列名,聚合统计
列转行UNION ALL 或 LATERAL VIEW explode列数量少或需动态处理

提示:Hive 3.0+ 支持 LATERAL VIEW inline(array_of_structs),但需先构造结构体数组。实际使用时请根据数据规模和复杂度选择合适方案。

作者 east

关注公众号“大模型全栈程序员”回复“小程序”获取1000个小程序打包源码。回复”chatgpt”获取免注册可用chatgpt。回复“大数据”获取多本大数据电子书

标签

AIGC AI创作 bert chatgpt github GPT-3 gpt3 GTP-3 hive mysql O2O tensorflow UI控件 不含后台 交流 共享经济 出行 图像 地图定位 外卖 多媒体 娱乐 小程序 布局 带后台完整项目 开源项目 搜索 支付 效率 教育 日历 机器学习 深度学习 物流 用户系统 电商 画图 画布(canvas) 社交 签到 联网 读书 资讯 阅读 预订

官方QQ群

小程序开发群:74052405

大数据开发群: 952493060

近期文章

  • 如何在Chrome中设置启动时自动打开多个默认网页
  • spark内存溢出怎样区分是软件还是代码原因
  • MQTT完全解析和实践
  • 解决运行Selenium报错:self.driver = webdriver.Chrome(service=service) TypeError: __init__() got an unexpected keyword argument ‘service’
  • python 3.6使用mysql-connector-python报错:SyntaxError: future feature annotations is not defined
  • 详解Python当中的pip常用命令
  • AUTOSAR如何在多个供应商交付的配置中避免ARXML不兼容?
  • C++thread pool(线程池)设计应关注哪些扩展性问题?
  • 各类MCAL(Microcontroller Abstraction Layer)如何与AUTOSAR工具链解耦?
  • 如何设计AUTOSAR中的“域控制器”以支持未来扩展?

文章归档

  • 2025年8月
  • 2025年7月
  • 2025年6月
  • 2025年5月
  • 2025年4月
  • 2025年3月
  • 2025年2月
  • 2025年1月
  • 2024年12月
  • 2024年11月
  • 2024年10月
  • 2024年9月
  • 2024年8月
  • 2024年7月
  • 2024年6月
  • 2024年5月
  • 2024年4月
  • 2024年3月
  • 2023年11月
  • 2023年10月
  • 2023年9月
  • 2023年8月
  • 2023年7月
  • 2023年6月
  • 2023年5月
  • 2023年4月
  • 2023年3月
  • 2023年1月
  • 2022年11月
  • 2022年10月
  • 2022年9月
  • 2022年8月
  • 2022年7月
  • 2022年6月
  • 2022年5月
  • 2022年4月
  • 2022年3月
  • 2022年2月
  • 2022年1月
  • 2021年12月
  • 2021年11月
  • 2021年9月
  • 2021年8月
  • 2021年7月
  • 2021年6月
  • 2021年5月
  • 2021年4月
  • 2021年3月
  • 2021年2月
  • 2021年1月
  • 2020年12月
  • 2020年11月
  • 2020年10月
  • 2020年9月
  • 2020年8月
  • 2020年7月
  • 2020年6月
  • 2020年5月
  • 2020年4月
  • 2020年3月
  • 2020年2月
  • 2020年1月
  • 2019年7月
  • 2019年6月
  • 2019年5月
  • 2019年4月
  • 2019年3月
  • 2019年2月
  • 2019年1月
  • 2018年12月
  • 2018年7月
  • 2018年6月

分类目录

  • Android (73)
  • bug清单 (79)
  • C++ (34)
  • Fuchsia (15)
  • php (4)
  • python (45)
  • sklearn (1)
  • 云计算 (20)
  • 人工智能 (61)
    • chatgpt (21)
      • 提示词 (6)
    • Keras (1)
    • Tensorflow (3)
    • 大模型 (1)
    • 智能体 (4)
    • 深度学习 (14)
  • 储能 (44)
  • 前端 (5)
  • 大数据开发 (492)
    • CDH (6)
    • datax (4)
    • doris (31)
    • Elasticsearch (15)
    • Flink (79)
    • flume (7)
    • Hadoop (19)
    • Hbase (23)
    • Hive (41)
    • Impala (2)
    • Java (71)
    • Kafka (10)
    • neo4j (5)
    • shardingsphere (6)
    • solr (5)
    • Spark (100)
    • spring (11)
    • 数据仓库 (9)
    • 数据挖掘 (7)
    • 海豚调度器 (10)
    • 运维 (34)
      • Docker (3)
  • 小游戏代码 (1)
  • 小程序代码 (139)
    • O2O (16)
    • UI控件 (5)
    • 互联网类 (23)
    • 企业类 (6)
    • 地图定位 (9)
    • 多媒体 (6)
    • 工具类 (25)
    • 电商类 (22)
    • 社交 (7)
    • 行业软件 (7)
    • 资讯读书 (11)
  • 嵌入式 (71)
    • autosar (63)
    • RTOS (1)
    • 总线 (1)
  • 开发博客 (16)
    • Harmony (9)
  • 技术架构 (6)
  • 数据库 (32)
    • mongodb (1)
    • mysql (13)
    • pgsql (2)
    • redis (1)
    • tdengine (4)
  • 未分类 (7)
  • 程序员网赚 (20)
    • 广告联盟 (3)
    • 私域流量 (5)
    • 自媒体 (5)
  • 量化投资 (4)
  • 面试 (14)

功能

  • 登录
  • 文章RSS
  • 评论RSS
  • WordPress.org

All Rights Reserved by Gitweixin.本站收集网友上传代码, 如有侵犯版权,请发邮件联系yiyuyos@gmail.com删除.