您现在的位置是：网站首页> 开发积累
日常开发技术收集积累
- 开发积累
- 2026-07-02
- 1863人已阅读
日常开发技术收集积累
基于Windows的Kafka有关环境搭建、以及使用.NET环境开发的案例代码与演示
物联网大数据组合-》EMQX+Kafka-》后端消费或生产(有Go解决方案的尽量用Go代替脱离其他语言)
微信还能接入HTTP协议？定制微信机器人，做消息推送，AI聊天
正片叠底（Multiply）和滤色（Screen）是两种基本的混合模式
游戏开发
扫描连接WIFI
生成播放视频的URL的二维码，用户扫描开始播放视频，播放完后展示连接WIFI二维码，用户手动扫码连接
小程序实现设置WIFI
/**
* 页面的初始数据
*/
data: {
wifiname:'wifi商家',
ssid: '@Ruijie-sDE3D',
bssid: '0',
password: '12345678',
},
/**
* 生命周期函数--监听页面加载
*/
onLoad: function (options) {
console.log("111",options)
var wifiname = options.wifiname
var qssid = options.alias;
// var qbssid = options.bssid;
var qpassword = options.wifipass;
this.setData({
wifiname:wifiname,
ssid: qssid,
// bssid: qbssid,
password: qpassword
})
var that = this
},
/**
* 点击连接按钮触发
*/
connectWifi: function() {
const that = this;
wx.showToast({
title: '请稍等',
icon: 'loading',
duration: 500//持续的时间
})
that.startWiFi();
},
/**
* 加载WiFi模块
*/
startWiFi: function() {
const that = this;
wx.startWifi({
complete: (res) => {
that.connected();
},
})
},
/**
* 连接WiFi
*/
connected: function() {
const that = this;
wx.connectWifi({
SSID: that.data.ssid,
// BSSID: that.data.bssid,
password: that.data.password,
success: () => {
wx.showToast({
title: 'WiFi连接成功',
})
// 跳转至成功页面
wx.redirectTo({
url: '/pages/success/success',
})
},
fail: (res) => {
that.errorDialog(res);
}
})
},
/**
* 连接失败弹窗
* @param {错误返回} res
*/
errorDialog: function(res) {
const that = this;
wx.showModal({
title: '连接失败',
content: '连接失败请打开wifi或复制连接',
confirmText: '复制密码',
success (res) {
if (res.confirm) {
that.copyPassword();
} else if (res.cancel) {
console.log('cancel')
}
},
fail(res) {
wx.showToast({
title: res.errMsg,
})
}
});
},
/**
* 复制密码到剪贴板
*/
copyPassword: function() {
const that = this;
wx.setClipboardData({
data: that.data.password,
success (res) {
wx.getClipboardData({
success (res) {
console.log(res.data);
}
})
}
})
},
/**
* 生命周期函数--监听页面初次渲染完成
*/
onReady: function () {
},
/**
* 生命周期函数--监听页面显示
*/
onShow: function () {
},
/**
* 生命周期函数--监听页面隐藏
*/
onHide: function () {
},
/**
* 生命周期函数--监听页面卸载
*/
onUnload: function () {
},
/**
* 页面相关事件处理函数--监听用户下拉动作
*/
onPullDownRefresh: function () {
},
/**
* 页面上拉触底事件的处理函数
*/
onReachBottom: function () {
},
/**
* 用户点击右上角分享
*/
onShareAppMessage: function () {
}
高并发场景设计
设计一个高并发秒杀场景需要考虑以下几个方面
1.数据库设计：选择适合高并发场景的数据库，如使用关系型数据库时可以采用分库分表的方式来提高数据库的读写性能。可以使用缓存技术来减轻数据库的压力，如使用Redis缓存热门商品信息。
2.队列系统：引入消息队列系统可以有效地削峰填谷，平衡请求的处理速度和系统的承载能力。秒杀请求可以先进入消息队列，再由后台异步处理，避免直接对数据库进行高并发的读写操作。
3.缓存优化：使用缓存来存储热门商品的库存信息，减少对数据库的频繁访问。可以使用分布式缓存，如Redis，来提高读取速度。
4.分布式系统：采用分布式架构来分担系统的负载，将请求分散到多个服务器上进行处理。可以使用负载均衡技术，如Nginx，进行请求的分发，保证每个服务器的压力均衡。
5.限流与排队：为了保护系统的稳定性，可以对请求进行限流，限制每秒的请求数量。同时，可以为每个用户分配一个唯一的标识符，通过排队系统来控制用户的访问顺序，避免系统被过多的请求拖垮。
6.异步处理：将秒杀请求的处理过程异步化，如使用消息队列、多线程等方式，保证请求的快速响应，避免阻塞其他请求的处理。
7.前端优化：通过前端技术，如CDN加速、页面静态化等，减少对后端的请求，提高用户访问的响应速度。可以采用前端缓存技术，如Local Storage，将商品信息缓存在前端，减少对后端的请求次数。
8.安全措施：为避免恶意请求和刷单行为，可以引入验证码、IP限制、用户身份验证等安全措施，保障系统的公平性和安全性。
以上是设计高并发秒杀场景的一些思路，具体的设计方案还需根据实际需求和系统架构进行详细的规划和实施。
在高并发场景下，常用的可靠任务队列
1.RabbitMQ：RabbitMQ是一个流行的开源消息队列系统，支持多种消息传输协议，如AMQP、MQTT等。它具有高可靠性、可持久化、分布式和高并发处理能力，适用于各种异步任务和消息传递场景。Kafka是一个分布式的发布-订阅消息系统，能够支撑海量数据的数据传递。在离线和实时的消 息处理业务系统中，Kafka都有广泛的应用。Kafka将消息持久化到磁盘中，并对消息创建了备份保证了 数据的安全。Kafka在保证了较高的处理速度的同时，又能保证数据处理的低延迟和数据的零丢失。
2.Apache Kafka：Kafka是一个分布式的流处理平台，也可以用作可靠的消息队列。Kafka以高吞吐量、持久性和可扩展性而闻名，适用于大规模数据处理和实时流处理场景。
3.Redis：Redis是一个内存中的数据结构存储系统，也可以用作消息队列。它支持发布/订阅模式和列表数据结构，可用于处理高并发的任务队列。
4.Apache ActiveMQ：ActiveMQ是一个基于Java的开源消息队列系统，支持多种消息传输协议，如AMQP、STOMP等。它具有高可用性、可靠性和扩展性，适用于异步任务处理和分布式系统集成。
5.Amazon Simple Queue Service (SQS)：SQS是亚马逊提供的一项托管消息队列服务，具有高度可靠性和可伸缩性。它支持标准队列和先入先出队列，适用于分布式系统和云计算环境。
这些可靠任务队列都具有不同的特点和适用场景，选择适合自己项目需求的队列系统需要综合考虑系统架构、可用性、性能要求和开发语言等因素。
总结
1：吞吐量较低：Kafka和RabbitMQ都可以。
2：吞吐量高：Kafka
以下是一个使用C#操作Kafka的可靠任务队列的简单示例：
首先，你需要安装NuGet包来引入Kafka相关库。使用Visual Studio的话，可以通过NuGet包管理器添加以下两个包：
Confluent.Kafka：提供了Kafka的客户端库。
Confluent.Kafka.Serialization：提供了Kafka消息的序列化和反序列化支持。
接下来，可以使用以下代码示例来创建一个生产者（Producer）和一个消费者（Consumer）：
using Confluent.Kafka;
using System;
using System.Threading;
class Program
{
static void Main(string[] args)
{
string bootstrapServers = "localhost:9092"; // Kafka服务器地址和端口
string topic = "my-topic"; // Kafka主题名称
// 生产者
var producerConfig = new ProducerConfig { BootstrapServers = bootstrapServers };
using (var producer = new ProducerBuilder<Null, string>(producerConfig).Build())
{
// 发送消息
var message = new Message<Null, string> { Value = "Hello Kafka" };
var deliveryReport = producer.ProduceAsync(topic, message).Result;
Console.WriteLine($"Message delivered to {deliveryReport.TopicPartitionOffset}");
// 等待消息发送完成
producer.Flush(TimeSpan.FromSeconds(10));
}
// 消费者
var consumerConfig = new ConsumerConfig
{
BootstrapServers = bootstrapServers,
GroupId = "my-group",
AutoOffsetReset = AutoOffsetReset.Earliest
};
using (var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build())
{
// 订阅主题
consumer.Subscribe(topic);
CancellationTokenSource cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
e.Cancel = true; // 阻止进程退出
cts.Cancel();
};
try
{
// 消费消息
while (true)
{
var consumeResult = consumer.Consume(cts.Token);
Console.WriteLine($"Received message: {consumeResult.Message.Value}");
}
}
catch (OperationCanceledException)
{
// 用户取消操作
}
finally
{
consumer.Close();
}
}
}
}
在这个示例中，我们创建了一个生产者来发送消息到指定的Kafka主题，然后创建了一个消费者来订阅该主题并接收消息。你可以根据自己的需求进行扩展和修改。
请确保将bootstrapServers设置为你的Kafka服务器地址和端口，将topic设置为你要使用的主题名称。此外，你还可以根据实际需求配置其他的Kafka相关参数。
这只是一个简单的示例，实际使用中可能需要更多的错误处理和配置。你可以参考Confluent.Kafka库的文档和示例来深入了解C#操作Kafka的更多功能和用法。
原理
人员标签，信息标签，人员和信息标签的匹配
人员和信息的标签定时更改
以人员看信息多的同一标签做人员标签（点赞，评论做加权）
信息标签也是以看的人多的共同标签打标签
定时多长时间对信息和个人打标签，何时信息标签打完毕
实现
信息发布预热期，作者自己打的标签，首推送关注者，再次符合标签的人员
远程桌面解决断开后UI交互问题
query session
tscon rdp-tcp#0 /dest:console
给机器人分配一个远程账户，登录进去后执行连接到本地
日常维护的用另一个远程账户
微信还能接入HTTP协议？定制微信机器人，做消息推送，AI聊天
3.9.2.23版本微信
提取码：1234
如何防止线上活动优惠券被仿
可在二维码加上加密的时间戳和个人信息等，时间验证个人信息验证，同时也可让用户输入密码验证使用优惠券
免费地图解决方案
国家地理信息公共服务平台 天地图
凯立德官网
维智物联全域定位
图炫（aimap.gl）地图可视化
http://location-dev.newayz.com/aimap-gl-v2/docs/
多媒体开发知识积累
正片叠底（Multiply）和滤色（Screen）是两种基本的混合模式
正片叠底（Multiply）和滤色（Screen）是两种基本的混合模式，分别用于使图片变暗和变亮。它们之间的组合还可以形成更复杂的混合模式，如叠加（Overlay）和柔光（Soft Light）。
正片叠底 —— 就是把两层图像的像素相乘，最后会得到一个更暗的图像。这个模式是对称的，也就是说交换基色和混合色得到的结果是一样的。
，其中a是基色，b是混合色。
滤色 —— 首先把两层图像的像素值取互补数，然后将它们相乘，最后再去互补数。这和正片叠底得到的结果是相反的。它会得到一个更亮的图像。
，其中a是基色，b是混合色。
叠加 —— 结合了正片叠底和滤色两种混合模式。基色中亮色的部分会更加亮，而暗色的部分会更暗。
，其中a是基色，b是混合色
Puppeteer相关技术收集
Puppeteer 简介
1、Puppeteer 简介
2、运行环境
- Nodejs 的版本不能低于 v7.6.0, 需要支持 async, await.
- 需要最新的 chrome driver, 这个你在通过 npm 安装 Puppeteer 的时候系统会自动下载的
npm install puppeteer --save
3、基本用法
const puppeteer = require('puppeteer');
(async () => {
const browser = await puppeteer.launch();
const page = await browser.newPage();
await page.goto('https://example.com');
await page.screenshot({path: 'example.png'});
await browser.close();
})();- 先通过 puppeteer.launch() 创建一个浏览器实例 Browser 对象
- 然后通过 Browser 对象创建页面 Page 对象
- 然后 page.goto() 跳转到指定的页面
- 调用 page.screenshot() 对页面进行截图
- 关闭浏览器
3.1 puppeteer.launch(options)
| 参数名称 | 参数类型 | 参数说明 |
| ignoreHTTPSErrors | boolean | 在请求的过程中是否忽略 Https 报错信息，默认为 false |
| headless | boolean | 是否以”无头”的模式运行 chrome, 也就是不显示 UI， 默认为 true |
| executablePath | string | 可执行文件的路劲，Puppeteer 默认是使用它自带的 chrome webdriver, 如果你想指定一个自己的 webdriver 路径，可以通过这个参数设置 |
| slowMo | number | 使 Puppeteer 操作减速，单位是毫秒。如果你想看看 Puppeteer 的整个工作过程，这个参数将非常有用。 |
| args | Array(String) | 传递给 chrome 实例的其他参数，比如你可以使用”–ash-host-window-bounds=1024x768” 来设置浏览器窗口大小。更多参数参数列表可以参考这里 |
| handleSIGINT | boolean | 是否允许通过进程信号控制 chrome 进程，也就是说是否可以使用 CTRL+C 关闭并退出浏览器. |
| timeout | number | 等待 Chrome 实例启动的最长时间。默认为30000（30秒）。如果传入 0 的话则不限制时间 |
| dumpio | boolean | 是否将浏览器进程stdout和stderr导入到process.stdout和process.stderr中。默认为false。 |
| userDataDir | string | 设置用户数据目录，默认linux 是在 ~/.config 目录，window 默认在 C:\Users{USER}\AppData\Local\Google\Chrome\User Data, 其中 {USER} 代表当前登录的用户名 |
| env | Object | 指定对Chromium可见的环境变量。默认为process.env。 |
| devtools | boolean | 是否为每个选项卡自动打开DevTools面板， 这个选项只有当 headless 设置为 false 的时候有效 |
3.2 Browser 对象
const puppeteer = require('puppeteer');
puppeteer.launch().then(async browser => {
// 保存 Endpoint，这样就可以重新连接
Chromiumconst browserWSEndpoint = browser.wsEndpoint();
// 从Chromium 断开连接
browser.disconnect();
// 使用endpoint 重新和 Chromiunm 建立连接
const browser2 = await puppeteer.connect({browserWSEndpoint});
// Close Chromium
await browser2.close();});| 方法名称 | 返回值 | 说明 |
| browser.close() | Promise | 关闭浏览器 |
| browser.disconnect() | void | 断开浏览器连接 |
| browser.newPage() | Promise(Page) | 创建一个 Page 实例 |
| browser.pages() | Promise(Array(Page)) | 获取所有打开的 Page 实例 |
| browser.targets() | Array(Target) | 获取所有活动的 targets |
| browser.version() | Promise(String) | 获取浏览器的版本 |
| browser.wsEndpoint() | String | 返回浏览器实例的 socket 连接 URL, 可以通过这个 URL 重连接 chrome 实例 |
4、Puppeteer 实战
- 打开京东首页
- 将光标 focus 到搜索输入框
- 键盘点击输入文字
- 点击搜索按钮
- 打开京东首页
- 找到输入框的 input 元素
- 设置 input 的值为要搜索文字
- 触发搜索按钮的单机事件
- 打开京东首页
- 输入“手机”关键字并搜索
- 获取前10个商品的 A 标签，并获取 href 属性值，获取商品详情链接
- 分别打开10个商品的详情页，截取网页图片
4.1 获取元素
let inputElement = await page.$("#search", input => input);
//下面写法等价
let inputElement = await page.$('#search');const links = await page.$$("a");
//下面写法等价
const links = await page.$$("a", links => links);4.2 获取元素属性
const value = await page.$eval('input[name=search]', input => input.value);
const href = await page.$eval('#a", ele => ele.href);
const content = await page.$eval('.content', ele => ele.outerHTML);4.3 执行自定义的 JS 脚本
const result = await page.evaluate(() => {
return Promise.resolve(8 * 7);
});
console.log(result); // prints "56"(async () => {
const browser = await puppeteer.launch({headless:true});
const page = await browser.newPage();
await page.goto('https://jr.dayi35.com');
await page.setViewport({width:1920, height:1080});
const documentSize = await page.evaluate(() => {
return {
width: document.documentElement.clientWidth,
height : document.body.clientHeight,
}
})
await page.screenshot({path:"example.png", clip : {x:0, y:0, width:1920, height:documentSize.height}});
await browser.close();
})();const aWindowHandle = await page.evaluateHandle(() => Promise.resolve(window));
//aindowHandle Handle for the window object.
const aHandle = await page.evaluateHandle('document'); // Handle for the 'document'.const aHandle = await page.evaluateHandle(() => document.body); const resultHandle = await page.evaluateHandle(body => body.innerHTML, aHandle); console.log(await resultHandle.jsonValue()); await resultHandle.dispose();
4.4 Page.exposeFunction
const docSize = await page.evaluate(()=> {
function getPageSize() {
return {
width: document.documentElement.clientWidth,
height : document.body.clientHeight,
}
}
return getPageSize();
});const puppeteer = require('puppeteer');
const crypto = require('crypto');
puppeteer.launch().then(async browser => {
const page = await browser.newPage();
page.on('console', msg => console.log(msg.text));
await page.exposeFunction('md5', text =>
crypto.createHash('md5').update(text).digest('hex')
);
await page.evaluate(async () => {
// use window.md5 to compute hashes
const myString = 'PUPPETEER';
const myHash = await window.md5(myString);
console.log(`md5 of ${myString} is ${myHash}`);});
await browser.close();
});const puppeteer = require('puppeteer');
const fs = require('fs');
puppeteer.launch().then(async browser => {
const page = await browser.newPage();
page.on('console', msg => console.log(msg.text));
await page.exposeFunction('readfile', async filePath => {
return new Promise((resolve, reject) => {
fs.readFile(filePath, 'utf8', (err, text) => {
if (err)
reject(err);
else
resolve(text);
});
});
});
await page.evaluate(async () => {
// use window.readfile to read contents of a file
const content = await window.readfile('/etc/hosts');
console.log(content);
});
await browser.close();
});5、Page.emulate 修改模拟器(客户端)运行配置
- Page.setViewport() 修改浏览器视窗大小
- Page.setUserAgent() 设置浏览器的 UserAgent 信息
- Page.emulateMedia() 更改页面的CSS媒体类型，用于进行模拟媒体仿真。 可选值为 “screen”, “print”, “null”, 如果设置为 null 则表示禁用媒体仿真。
- Page.emulate() 模拟设备，参数设备对象，比如 iPhone, Mac, Android 等
page.setViewport({width:1920, height:1080}); //设置视窗大小为 1920x1080
page.setUserAgent('Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.90 Safari/537.36');
page.emulateMedia('print'); //设置打印机媒体样式const puppeteer = require('puppeteer');
const devices = require('puppeteer/DeviceDescriptors');
const iPhone = devices['iPhone 6'];
puppeteer.launch().then(async browser => {
const page = await browser.newPage();
await page.emulate(iPhone);
await page.goto('https://www.google.com');
// other actions...
await browser.close();});6、键盘和鼠标
- keyboard.down(key[, options]) 触发 keydown 事件
- keyboard.press(key[, options]) 按下某个键，key 表示键的名称，比如 ‘ArrowLeft’ 向左键，详细的键名映射请戳这里
- keyboard.sendCharacter(char) 输入一个字符
- keyboard.type(text, options) 输入一个字符串
keyboard.up(key) 触发 keyup 事件
page.keyboard.press("Shift"); //按下 Shift 键
page.keyboard.sendCharacter('嗨');
page.keyboard.type('Hello'); // 一次输入完成
page.keyboard.type('World', {delay: 100}); // 像用户一样慢慢输入
鼠标操作：
- mouse.click(x, y, [options]) 移动鼠标指针到指定的位置，然后按下鼠标，这个其实 mouse.move 和 mouse.down 或 mouse.up 的快捷操作
- mouse.down([options]) 触发 mousedown 事件，options 可配置:
- options.button 按下了哪个键，可选值为[left, right, middle], 默认是 left, 表示鼠标左键
- options.clickCount 按下的次数，单击，双击或者其他次数
- delay 按键延时时间
- mouse.move(x, y, [options]) 移动鼠标到指定位置， options.steps 表示移动的步长
- mouse.up([options]) 触发 mouseup 事件
7、另外几个有用的 API
7.1 Page.waitFor 系列 API
- page.waitFor(selectorOrFunctionOrTimeout[, options[, …args]]) 下面三个的综合 API
- page.waitForFunction(pageFunction[, options[, …args]]) 等待 pageFunction 执行完成之后
- page.waitForNavigation(options) 等待页面基本元素加载完之后，比如同步的 HTML, CSS, JS 等代码
- page.waitForSelector(selector[, options]) 等待某个选择器的元素加载之后，这个元素可以是异步加载的，这个 API 非常有用，你懂的。
await page.waitForSelector('.gl-item'); //等待元素加载之后，否则获取不到异步加载的元素
const links = await page.$$eval('.gl-item > .gl-i-wrap > .p-img > a', links => {
return links.map(a => {
return {
href: a.href.trim(),
name: a.title
}
});
});
7.2 page.getMetrics()
- Timestamp 度量标准采样的时间戳
- Documents 页面文档数
- Frames 页面 frame 数
- JSEventListeners 页面内事件监听器数
- Nodes 页面 DOM 节点数
- LayoutCount 页面布局总数
- RecalcStyleCount 样式重算数
- LayoutDuration 所有页面布局的合并持续时间
- RecalcStyleDuration 所有页面样式重新计算的组合持续时间。
- ScriptDuration 所有脚本执行的持续时间
- TaskDuration 所有浏览器任务时长
- JSHeapUsedSize JavaScript 占用堆大小
- JSHeapTotalSize JavaScript 堆总量
8、总结和源码
//延时函数
function sleep(delay) {
return new Promise((resolve, reject) => {
setTimeout(() => {
try {
resolve(1)
} catch (e) {
reject(0)
}
}, delay)
})
}
const puppeteer = require('puppeteer');
puppeteer.launch({
ignoreHTTPSErrors:true,
headless:false,slowMo:250,
timeout:0}).then(async browser => {
let page = await browser.newPage();
await page.setJavaScriptEnabled(true);
await page.goto("https://www.jd.com/");
const searchInput = await page.$("#key");
await searchInput.focus(); //定位到搜索框
await page.keyboard.type("手机");
const searchBtn = await page.$(".button");
await searchBtn.click();
await page.waitForSelector('.gl-item'); //等待元素加载之后，否则获取不异步加载的元素
const links = await page.$$eval('.gl-item > .gl-i-wrap > .p-img > a', links => {
return links.map(a => {
return {
href: a.href.trim(),
title: a.title
}
});
});
page.close();
const aTags = links.splice(0, 10);
for (var i = 1; i < aTags.length; i++) {
page = await browser.newPage()
page.setJavaScriptEnabled(true);
await page.setViewport({width:1920, height:1080});
var a = aTags[i];
await page.goto(a.href, {timeout:0}); //防止页面太长，加载超时
//注入代码，慢慢把滚动条滑到最底部，保证所有的元素被全部加载
let scrollEnable = true;
let scrollStep = 500; //每次滚动的步长
while (scrollEnable) {
scrollEnable = await page.evaluate((scrollStep) => {
let scrollTop = document.scrollingElement.scrollTop;
document.scrollingElement.scrollTop = scrollTop + scrollStep;
return document.body.clientHeight > scrollTop + 1080 ? true : false
}, scrollStep);
await sleep(100);
}
await page.waitForSelector("#footer-2014", {timeout:0}); //判断是否到达底部了
let filename = "images/items-"+i+".png";
//这里有个Puppeteer的bug一直没有解决，发现截图的高度最大只能是16384px， 超出部分被截掉了。
await page.screenshot({path:filename, fullPage:true});
page.close();
}
browser.close();
});
await page.setUserAgent(fake.user_agent())
可以看到 HTTP 状态码 为 403（403 Forbidden），即使我们设置了 User-Agent，美团依然能够检测到 WebDriver。
Pyppeteer 的 Page 对象有一个 evaluateOnNewDocument 方法，可以在每次加载网页的时候执行某个语句，此处执行将 WebDriver 隐藏的命令
browser = await launch({headless: False,
args: ['--disable-infobars']
})
page = await browser.newPage()
await page.setUserAgent(fake.user_agent())
await page.evaluateOnNewDocument('function(){Object.defineProperty(navigator, "webdriver", {get: () => undefined})}')
await page.goto(URL, options={'timeout': 10000})
await asyncio.sleep(412)
await browser.close()
Puppeteer一些高级玩儿法
用户数据持久化
平时访问网站时关键 Cookies 已经保存到本地浏览器，因此下次登录时可以直接读取并保持登录状态，这些信息保存在用户目录下，其不仅包含浏览器的基本配置信息，还有一些 Cache、Cookies 等信息，若能在浏览器启动时读取这些信息，则可以恢复一些历史记录以及登录状态信息。
Pyppeteer 提供了实现手段，即在启动的时候设置 userDataDir：
import asyncio
from faker import Faker
from pyppeteer import launch
fake = Faker()
URL = 'https://www.zhihu.com/'
async def main():
browser = await launch({'headless': False,
'args': ['--disable-infobars'],
'userDataDir': './userdata'
})
page = await browser.newPage()
await page.setViewport({'width': 1530, 'height': 800})
await page.setUserAgent(fake.user_agent())
await page.evaluateOnNewDocument('function(){Object.defineProperty(navigator, "webdriver", {get: () => undefined})}')
await page.goto(URL, options={'timeout': 10000})
await asyncio.sleep(412)
await browser.close()
if __name__ == '__main__':
asyncio.get_event_loop().run_until_complete(main())
第一次启动时先手动登录：
登录后相关信息会保存在用户目录下，下次登录时即可直接读取：
此后再启动，无需重新登录（除非 Cookies 过期）。
Browser
launch 方法返回的是 Browser 对象（浏览器对象），即 Browser 类的一个实例，其拥有许多用于操作浏览器的方法。
无痕模式
无痕模式的好处就是环境干净，不与其他的浏览器示例共享 Cache、Cookies 等内容，其开启方式可以通过 createIncognitoBrowserContext 方法，其返回一个 context 对象，用其创建新选项卡：
import asyncio
from faker import Faker
from pyppeteer import launch
fake = Faker()
URL = 'https://gz.meituan.com/meishi/'
async def main():
browser = await launch({'headless': False,
'args': ['--disable-infobars'],
'userDataDir': './userdata'
})
context = await browser.createIncognitoBrowserContext()
page = await context.newPage()
await page.setViewport({'width': 1530, 'height': 800})
await page.setUserAgent(fake.user_agent())
await page.evaluateOnNewDocument('function(){Object.defineProperty(navigator, "webdriver", {get: () => undefined})}')
await page.goto(URL, options={'timeout': 10000})
await asyncio.sleep(412)
await browser.close()
if __name__ == '__main__':
asyncio.get_event_loop().run_until_complete(main())
通过Puppeteer实现滚动加载
/* 引入相关 工具 */
const fs = require('fs')
const cheerio = require('cheerio')
const puppeteer = require('puppeteer')
/* 定义函数 */
let getListData = async function(Category) {
/* 初始化 puppeteer*/
const browser = await puppeteer.launch({
executablePath: 'D:\\chrome-win\\chrome.exe',//把项目中的这个chrome-win文件夹放到D盘根目录
headless: false //这个是 是否打开chrome可视化窗口 true是不打开 false是打开
})
//获取page实例
const page = await browser.newPage()
//我这里是通过 入参传过来的 分类来判断抓取相应页面的数据
let url = ''
switch (Category) {
case '0':
url = 'https://www.toutiao.com/ch/news_game/'
break;
case '1':
url = 'https://www.toutiao.com/ch/news_entertainment/'
break;
case '2':
url = 'https://www.toutiao.com/ch/news_history/'
break;
case '3':
url = 'https://www.toutiao.com/ch/news_finance/'
break;
}
//打开页面
await page.goto(url)
//定义页面内容及Jquery
var content , $
/* 页面滚动方法 */
async function scrollPage(i) {
content = await page.content();
$ = cheerio.load(content);
/*执行js代码（滚动页面）*/
await page.evaluate(function () {
/* 这里做的是渐进滚动，如果一次性滚动则不会触发获取新数据的监听 */
for (var y = 0; y <= 1000*i; y += 100) {
window.scrollTo(0,y)
}
})
// 获取数据列表
const li = $($('.feedBox').find('ul')[0]).find('li')
return li
}
let i = 0
let li = await scrollPage(++i)
//如果数据列表 不够30 则一直获取
while (li.length < 30) {
li = await scrollPage(++i)
}
let data = {
list: []
}
/* 封装返回的数据*/
li.map(function (index,item) {
$(item).find('img').attr('src') != undefined ?
data.list.push({
src: $(item).find('img').attr('src'),
title: $($(item).find('.title')).text(),
source:$($(item).find('.source')).text(),
href:$($(item).find('.title')).attr('href')
}):''
})
//顺手存入本地文件
fs.writeFileSync('tt.JSON',JSON.stringify(data))
fs.writeFileSync('tt.html',content)
/* 关闭 puppeteer*/
await browser.close()
return data
}
module.exports = getListData
通过API将图片上传到免费图床-遇见图床
通过API将图片上传到免费图床-遇见图床
C# async 函数
public async static void GetInfoAsync()
{
Task<bool> task = Task.Run<bool>(() =>
{
Thread.Sleep(10000); //模拟耗时
return true;
});
//以下两种方式
bool taskResult1 = await task; //内部自己执行了GetAwaiter()
bool taskResult = task.GetAwaiter().GetResult(); //自己手动执行Awaiter(), 但是阻塞UI
Console.WriteLine(taskResult);
}
一个上传文件的例子
async Task<string> UpFile()
{
System.Net.Http.HttpClient client = new System.Net.Http.HttpClient();
var content = new MultipartFormDataContent();
//添加字符串参数，参数名为qq
//content.Add(new StringContent("123456"), "qq");
string path = "g:\\1.png";// Path.Combine(System.Environment.CurrentDirectory, "1.png");
//添加文件参数，参数名为files，文件名为123.png
var fileContent = new ByteArrayContent(System.IO.File.ReadAllBytes(path));
fileContent.Headers.ContentType = MediaTypeHeaderValue.Parse("multipart/form-data");
content.Add(fileContent, "image", "123.png");
content.Add(new StringContent("chaoneng"), "apiType");
content.Add(new StringContent("14ff9ab178c34097ff171964fdb166f4"), "token");
var requestUri = new Uri("https://www.hualigs.cn/api/upload"); //"https://www.hualigs.cn/api/upload";
ServicePointManager.SecurityProtocol = SecurityProtocolType.Ssl3
| SecurityProtocolType.Tls
| (SecurityProtocolType)0x300 //Tls11
| (SecurityProtocolType)0xC00; //Tls12
var response = await client.PostAsync(requestUri, content);
string url = "";
if (response.IsSuccessStatusCode)
{
Console.WriteLine("上传成功！");
var result = response.Content.ReadAsStringAsync().Result;
//MessageBox.Show(result);
try
{
Hashtable m_HH = JsonHelper.JsonStrToOBJ<Hashtable>(result);
if (m_HH.ContainsKey("code"))
{
string code = m_HH["code"].ToString();
if (code == "200")
{
if (m_HH.ContainsKey("data"))
{
Hashtable m_dataHH = JsonHelper.OBJToType<Hashtable>(m_HH["data"]);
if (m_dataHH.ContainsKey("url"))
{
Hashtable m_urlHH = JsonHelper.OBJToType<Hashtable>(m_dataHH["url"]);
if (m_urlHH.ContainsKey("distribute"))
{
Console.WriteLine(m_urlHH["distribute"].ToString());
//MessageBox.Show(m_urlHH["distribute"].ToString());
url = m_urlHH["distribute"].ToString();
}
}
else
{
MessageBox.Show("数据解析异常:" + result);
}
}
else
{
MessageBox.Show("数据解析异常:" + result);
}
}
else
{
string str = "错误码：" + code;
if (m_HH.ContainsKey("msg"))
{
str += "," + m_HH["msg"].ToString();
}
MessageBox.Show(str);
}
}
else
{
MessageBox.Show("数据解析异常:" + result);
}
}
catch
{
}
}
else
{
MessageBox.Show("上传异常");
}
/*
Task<string> task = Task.Run<string >(() =>
{
return url;
});
*/
return url;
}
async private void button29_Click(object sender, EventArgs e)
{
string str = await UpFile();
MessageBox.Show(str);
}
防止恶意访问攻击
写一个异常访问ip清洗工具，1、同一时刻60s内请求超过60以上直接封禁该ip 100分钟，2、禁止所有国外ip和港、澳、台等地方IP。3、放进来的ip 随机进行滑块验证机器人，没通过的直接放入验证ip队列，该ip请求必须通过验证才会被从验证队列释放。4、静态媒体资源可以放到公共免费平台 用url跨站调用。5、网站只暴露一个入口，该页面加入js验证用户是否是机器人的算法，通过算法 请求后台生成jwt加密失效密钥，token通过cookie存在用户端，token有效期内可以浏览网站其他页面，负责跳转首页面。
各大地图厂商收费，免费解决方案
如果只用地图展示类，3种都满足，天地图有些丑罢了，凯立德和WAYZ和高德的风格蛮像；
如果有导航类需求那就凯立德了；
如果有设备需要接入定位，那WAYZ更方便接入
用天地图就好了，国家的，免费
所谓的支付宝/微信支付“云面签”原理
市面上最近流行所谓的“云免签”，只需要收款人在其提供云免签服务的平台上传支付宝/微信二维码，不需要收款人在PC端或手机上安装监控支付宝/微信支付结果通知的程序，然后由云端来帮助监控支付结果。
研究了一下各种“云免签”方案，大致分为几种方案：
1、微信店员模式
收款人作为店主，让收款人添加提供云免签服务的平台的微信号为店员，由于店主收款消息都会自动通知店员，因此只需要在店员端监听支付结果就行，不需要店主始终在线。
由于支付宝只允许一个店员属于一个店主，因此此种模式对支付宝成本过高。
2、监听支付宝/微信网页端模式
提供云免签服务的平台获取支付宝/微信PC网页版扫码登陆二维码，让收款人扫码登陆，提供云免签服务的平台在服务器端通过扫收款人账单来获取支付结果。
3、支付宝当面支付+当面付分账功能
这些方案还有诸多细节需要考虑，例如怎么自动化通过店员请求、怎样保持支付宝/微信登陆状态等等。
C#框架相关笔记
返回状态码
StatusCode = 200;
ctx.Response.StatusCode = StatusCode;//验证通过调用
ctx.Response.End();
大型分布式系统设计初版
大型分布式系统设计初版
人员数据落到各个节点，新数据落的规则按一致性hash决定见后面
业务数据落到各个节点，新数据落的规则按地理位置范围和表数据量和hash落到那个节点具体见后
中心业务数据地理位置库分布规则表
业务数据保存两份，按地理位置保存 按人员保存
用户共享数据节点如角色等 业务数据共享节点如栏目等
用户的归宿 如群 谁创建群，群数据落点在那个用户节点
所有的新落点按一致性hash权重获得
所有节点多个从库
关于共享数据的缓存，角色 栏目等
各区数据中心
节点ID规则 uuid+数据中心id+节点编号
数据中心所拥有的数据节点服务 调用节点服务认证
redis 数据缓存服务
redis 消息服务
Redis分布式锁
数据级缓存，页面级缓存
数据库节点服务 对应所在数据中心
根据用户id或接入IP或业务数据id 获得数据链接节点根据数据中心
栏目归宿 共享 角色 地理范围 权限 看 发 回复 购买
获得web模块的页面部分html
用户访问过程
未登录状态访问：
主gate服务根据用户IP引导用户访问最近的相对闲置web节点，根据用户的地理位置拉取相应范围内栏目及数据（根据地理位置获得数据中心，根据数据中心获得相对闲置的数据服务节点 拉取数据）
用户切换地理位置同上只是位置不再通过IP而根据用户选择引导用户访问最近相对闲置的web 节点后续相同
用户登陆访问
根据全局用户名对应的用户ID获得用户信息所在数据中心获得数据节点服务拉取用户基础数据及权限数据，根据地理位置拉取业务栏目数据流程如上
信息发布，在地理位置所在数据中心拿取相对闲置的数据库节点添加数据，同时往用户所在数据节点添加相应数据，数据回复交易等类式 回复记录节点交易等与记录关联数据位于同一数据节点
直播App架构
不用编程手段有哪些现成的软件可以将网页打包为手机和PC应用
不用编程手段将网页打包为手机应用或 PC 应用，可以使用以下一些现成的软件或平台：
在线平台：
AppGeyser：无需下载软件，在浏览器端即可操作，支持自动生成 Android（APK）和 iOS（IPA）安装包，适合完全没有编程基础的小白用户。
一门 APP 打包平台：提供模块化混合开发 APP 底层框架，有 200 + 原生功能、2000 + 映射 JS 接口。支持 Android、iOS 双系统，可在线配置多种功能，如设置顶部原生标题栏、下拉刷新、缓存管理等，还能协助上架发布。
BuildFire：也是一个在线的 APP 打包平台，具有 “一键打包、自动适配” 的功能，用户可以在浏览器中通过简单的配置，将网页转化为手机应用，无需编写代码。
桌面工具：
HBuilderX：这是一款国产免费的桌面工具，支持本地调试。利用其 “5+App” 引擎，可直接将网页封装为原生应用外壳。用户新建项目后，填写网站 URL、设置图标与启动页等信息，即可一键打包生成 Android 和 iOS 应用安装包。
FusionApp：这是一款系统工具软件，具有丰富的主题图案设计模组，用户可以自由选择，可以将任何网站制作成 APP 应用。对于需要快速测试网页转 APP 效果的用户，可以先使用 Sniff Master 抓包工具分析网页结构，再使用 FusionApp 进行封装。
Pake：是一款基于 Rust Tauri 框架的桌面应用打包工具，可在十秒钟将网页应用封装成桌面程序，打包后的程序体积小，一般在 5M 上下，支持 Windows、Mac、Linux 全平台。
Q:你是一名优秀的全栈开发工程师，同时精通物联网技术,现在有百万级的设备接入平台，设备通过MQTT与平台交互，MQTT服务采用EMQX ,消息队列用Kafka,后端使用C#通过Kafka和设备交互，给出详细实现和例子代码
A:...
纯 Go 开发、对标 Kafka 百万级物联网流消息中间件完整选型
一、首选：NATS JetStream（纯 Go，物联网最佳平替）
二、次选：NSQ（纯 Go，轻量分布式队列，适合实时设备消息）
平替EMQX的Go编写的等量级的MQTT服务有吗？
EMQX 是 Erlang 开发，优势：原生分布式集群、完整 MQTT5、内置规则引擎、Kafka/NATS 双向桥、会话持久化、百万级连接；
下面全部服务端 100% Go 编写，满足你现有架构：百万设备 + Kafka/NATS 桥接 + C# 后端物联网平台，分生产级首选、轻量备选、NATS 一体化方案（最优改造路线） 三类。
一、生产级首选：Comqtt（国内成熟、最贴合 EMQX 完整能力）
基础信息
仓库：https://github.com/wind-c/comqtt，纯 Go，国产开源，专为海量 IoT 设计，唯一完整对标 EMQX 全功能的 Go MQTT Broker
二、次选：Mochi-MQTT（https://github.com/mochi-mqtt/server，嵌入式 + 独立部署双模式）
基础信息
纯 Go，完全 MQTT5 合规，国外成熟开源，分两种部署模式：独立二进制服务 / Go 库嵌入自有服务
三、轻量备选：HMQ（https://github.com/fhmq/hmq，极简 Go MQTT Broker）
纯 Go，轻量，支持集群、Kafka 桥、WebSocket、TLS、HTTP 鉴权
四、一体化最优方案：NATS + NATS MQTT 网关（纯 Go 全家桶，替代 EMQX+Kafka 两套组件），https://github.com/nats-io/nats-server https://github.com/nats-io/nats.go
| 中间件 | 语言 | 延迟 | 持久 | 适合场景 |
|---|---|---|---|---|
| NATS | Go | 微秒级极低 | JetStream 可选 | 实时、IoT、微服务、边缘 |
| Kafka | Java | 毫秒级 | 强制持久 | 海量日志、大数据流批处理 |
| RabbitMQ | Erlang | 毫秒级 | 默认持久 | 复杂业务可靠队列、事务 |
NATS JetStream + MQTT 完整对接方案
NATS 原生内置 MQTT 网关，无需额外 MQTT Broker，一套服务同时支持 NATS TCP、WebSocket、MQTT 客户端，IoT 设备用 MQTT 上报，后端 Go 服务用 NATS JetStream 消费持久消息。
一、启动 NATS 开启 MQTT + JetStream
1. 命令行一键启动（最简）
bash
运行
# -js 开启JetStream；-mqtt 开启MQTT网关，默认端口1883
nats-server -js -mqtt
参数说明：
MQTT TCP：1883
MQTT TLS：8883（需证书）
NATS TCP：4222
WebSocket MQTT：8083
自定义 MQTT 端口、数据目录：
bash
运行
nats-server -js -mqtt -mqtt_port 1884 -sd ./nats_js_data
2. 配置文件生产部署（nats.conf）
conf
# JetStream持久存储
jetstream {
store_dir: "./nats_data"
max_file_store: 10G
}
# MQTT网关配置
mqtt {
listen: "0.0.0.0:1883"
ws_listen: "0.0.0.0:8083"
# 开启MQTT 3.1.1/5.0
allow_unknown_client_ids: true
}
启动：
bash
运行
nats-server -c nats.conf
验证是否开启 MQTT
bash
运行
nats server info
输出看到 mqtt 模块即正常。
二、核心映射规则（MQTT Topic ↔ NATS Subject）
NATS MQTT 网关自动双向映射主题，自动转换通配符：
MQTT /device/001/data → NATS device.001.data
MQTT 单层通配 + → NATS *
MQTT device/+/data = NATS device.*.data
MQTT 多层通配 # → NATS >
MQTT device/# = NATS device.>
关键：MQTT 的 / 自动替换成 NATS 的 .，两边发布订阅互通。
三、两种典型业务架构
架构 1：IoT 设备 MQTT 上报 → JetStream 持久存储 → Go 后端消费
MQTT 设备发布：device/001/sensor
NATS 自动转为 subject：device.001.sensor
JetStream 创建流绑定 device.>，持久所有设备上报数据
Go 使用 nats.go 消费 JetStream 流做业务处理
架构 2：Go 后端发指令（NATS）→ MQTT 设备接收
Go 发布 NATS subject：device.001/cmd
网关自动转为 MQTT topic：device/001/cmd
MQTT 设备订阅该主题接收指令
四、完整实操代码
1）Go 服务端：创建 JetStream 流，消费 MQTT 上报数据
go
运行
package main
import (
"fmt"
"log"
"time"
"github.com/nats-io/nats.go"
)
func main() {
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
// 绑定JetStream
js, err := nc.JetStream()
if err != nil {
log.Fatal("未开启JetStream：", err)
}
// 创建流，捕获所有device下的MQTT消息
streamName := "DEVICE_DATA"
_, err = js.StreamInfo(streamName)
if err != nil {
cfg := &nats.StreamConfig{
Name: streamName,
Subjects: []string{"device.>"}, // 对应MQTT device/#
Storage: nats.FileStorage,
MaxAge: 72 * time.Hour,
Replicas: 1,
}
_, err = js.AddStream(cfg)
if err != nil {
log.Fatal(err)
}
fmt.Println("设备数据流创建完成")
}
// 持久Push消费者，断点续消费
sub, err := js.Subscribe("device.>", func(m *nats.Msg) {
// m.Subject = device.001.sensor
fmt.Printf("收到MQTT设备数据 | nats主题:%s 内容:%s\n", m.Subject, string(m.Data))
m.Ack() // 确认消费完成
}, nats.Durable("device-consumer"))
if err != nil {
log.Fatal(err)
}
defer sub.Unsubscribe()
select {}
}
2）MQTT 客户端示例（paho.mqtt.golang，Go 设备端模拟）
模拟硬件设备通过 MQTT 上报数据到 NATS 网关：
go
运行
package main
import (
"fmt"
"log"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
func main() {
// 连接NATS内置MQTT网关 1883端口
opts := mqtt.NewClientOptions().AddBroker("tcp://127.0.0.1:1883")
opts.SetClientID("device_001")
client := mqtt.NewClient(opts)
if token := client.Connect(); token.Wait() && token.Error() != nil {
log.Fatal(token.Error())
}
// MQTT主题 device/001/sensor 映射NATS device.001.sensor
topic := "device/001/sensor"
for i := 0; i < 10; i++ {
payload := fmt.Sprintf("设备001温度：%d℃", 20+i)
token := client.Publish(topic, 1, false, payload)
token.Wait()
fmt.Println("MQTT上报：", payload)
time.Sleep(2 * time.Second)
}
client.Disconnect(1000)
}
3）Go 后端下发指令给 MQTT 设备
go
运行
// 直接用NATS发布，自动转发MQTT
nc.Publish("device.001/cmd", []byte("关闭传感器"))
// 对应MQTT设备订阅 topic: device/001/cmd
五、MQTT QoS 与 JetStream 持久关系
MQTT QoS0：最多一次，Core NATS，不进 JetStream（断开丢失）
MQTT QoS1/QoS2：消息写入 JetStream 持久，保证至少一次投递
设备离线时，MQTT 会话保留消息，重连自动补发；
JetStream 存储所有历史上报，后端可回放全量设备数据。
六、权限控制（生产必备）
1. 简单账号密码认证
启动 NATS 时开启用户验证：
bash
运行
nats-server -js -mqtt --user dev --pass 123456
MQTT 客户端连接时填入用户名密码即可。
2. JWT/NKEY 精细化权限（多租户 IoT）
限制 MQTT 设备只能发布自身主题，禁止读写其他设备：
conf
authorization {
users = [
{user: "dev001", pass: "xxx", permissions: {
publish: ["device.001.>"],
subscribe: ["device.001.cmd"]
}}
]
}
七、常用运维命令
bash
运行
# 查看MQTT在线客户端
nats mqtt clients
# 查看JetStream设备数据流
nats stream info DEVICE_DATA
# 回放所有设备历史上报消息
nats sub "device.>" --stream DEVICE_DATA --from-start
八、常见问题踩坑
MQTT 消息收不到 JetStream
确认发布 QoS≥1；QoS0 不会写入 JetStream
流 Subject 是否匹配 device.>
主题不匹配
MQTT / 转 NATS .，不要混用分隔符
设备断开后丢失消息
MQTT 客户端设置 SetCleanSession(false)，开启持久会话
MQTT 端口占用
修改 -mqtt_port 参数更换端口
九、方案优势对比（NATS MQTT vs 独立 EMQX/Mosquitto）
单进程一体化：不用维护两套中间件（MQTT Broker + NATS）
JetStream 统一持久：设备数据、业务消息一套存储
Go 原生高性能，边缘网关低内存占用
统一权限、TLS、集群、跨区域超级集群
同时支持 TCP-MQTT、WS-MQTT、原生 NATS、WebSocket-NATS 多端接入
整体架构说明（支撑百万设备）
架构链路
设备 (MQTT) → EMQX 集群 → EMQX Kafka 桥接 → Kafka 集群（多分区多副本） → C# 消费服务集群（设备上行消息）
C# 业务服务 → Kafka 生产指令消息 → EMQX 消费 Kafka 桥接 → MQTT 下发设备（下行控制）
核心组件分工
EMQX 5.x 集群
1.百万设备长连接承载，负载均衡，断线重连、心跳保活、MQTT QoS1 保障消息不丢
2.内置 Kafka 桥接插件，上行设备消息直接转发 Kafka，无需中间服务转发
3.反向桥接：订阅 Kafka 指定 Topic，将平台下发指令转 MQTT 推送给设备
Kafka 集群
1.上行 Topic：iot/device/up（设备上报：传感器、心跳、告警、日志）
2.下行 Topic：iot/device/down（平台下发：开关、参数配置、OTA 升级指令）
3.分区规划：分区数≥消费服务实例数，百万设备建议 32/64 分区，副本 3
C# 后端体系
1.消费者服务（后台 Worker Service 集群）：消费iot/device/up，设备数据解析、存储、告警判断、在线状态维护
2.生产者 SDK（WebApi / 业务服务）：生产下行指令到iot/device/down，实现远程控制设备
3.技术栈：.NET 8 Worker Service、Confluent.Kafka 2.4、Dapper/MySQL/Redis、EMQX 设备权限校验
百万并发关键优化点
1.EMQX：开启连接池、限流、共享订阅、消息持久化、会话持久化，集群节点水平扩容
2.Kafka：分区扩容、批量生产 / 消费、压缩消息、合理 retention 策略，独立磁盘存储日志
3.C# 服务：多线程消费、批量入库、Redis 缓存设备在线状态、异步 IO、服务集群水平扩展
4.消息规范：统一 JSON payload，携带设备唯一deviceId，区分消息类型msgType
一、EMQX Kafka 双向桥接配置
1. EMQX 上行桥（设备消息推 Kafka）
路径：EMQX Dashboard → 数据集成 → 桥接 → Kafka
plaintext
桥接名称：kafka_device_up
Kafka Broker：192.168.1.100:9092,192.168.1.101:9092,192.168.1.102:9092
目标Topic：iot/device/up
消息模板（把MQTT元数据+payload打包发给Kafka）：
{
"deviceId": "${clientid}",
"mqttTopic": "${topic}",
"payload": ${payload},
"qos": ${qos},
"timestamp": ${timestamp},
"node": "${node}"
}
触发规则：匹配所有设备上报主题 #
批量发送开启，批量大小100，linger 5ms
2. EMQX 下行桥（消费 Kafka 指令转 MQTT 下发设备）
桥接名称：kafka_device_down
消费 Kafka Topic：iot/device/down
消费组：emqx_down_group
消息解析模板：
plaintext
{
"targetTopic": "${value.targetTopic}",
"payload": "${value.payload}",
"qos": ${value.qos}
}
逻辑：从 Kafka 消息取出targetTopic作为 MQTT 推送主题，payload 下发设备
MQTT 主题规范约定
1.设备上报上行（EMQX 转发 Kafka）
心跳：device/${deviceId}/heartbeat
传感器数据：device/${deviceId}/data
告警：device/${deviceId}/alarm
2.平台下发下行（Kafka→EMQX→设备）
控制指令：device/${deviceId}/cmd
参数配置：device/${deviceId}/config
OTA 升级：device/${deviceId}/ota
二、C# 通用依赖
NuGet 包安装
xml
<!-- Kafka官方客户端 -->
<PackageReference Include="Confluent.Kafka" Version="2.4.0" />
<!-- JSON序列化 -->
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
<!-- 后台常驻服务 -->
<PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
<!-- Redis缓存设备在线状态（百万设备必备） -->
<PackageReference Include="StackExchange.Redis" Version="2.7.23" />
三、统一消息实体模型
1. 上行 Kafka 消息实体（EMQX 转发过来）
csharp
运行
/// <summary>
/// EMQX转发到Kafka的设备上行消息包装
/// </summary>
public class DeviceUpMessage
{
/// <summary>设备唯一编号，MQTT clientId</summary>
public string DeviceId { get; set; }
/// <summary>原始MQTT主题</summary>
public string MqttTopic { get; set; }
/// <summary>原始设备上报载荷</summary>
public object Payload { get; set; }
/// <summary>MQTT QoS</summary>
public int Qos { get; set; }
/// <summary>消息时间戳ms</summary>
public long Timestamp { get; set; }
/// <summary>EMQX集群节点</summary>
public string Node { get; set; }
}
/// <summary>设备上报业务载荷示例：传感器数据</summary>
public class DeviceDataPayload
{
/// <summary>消息类型 heartbeat/data/alarm</summary>
public string MsgType { get; set; }
/// <summary>温度</summary>
public float Temp { get; set; }
/// <summary>湿度</summary>
public float Humidity { get; set; }
/// <summary>电压</summary>
public float Voltage { get; set; }
/// <summary>设备信号强度</summary>
public int Rssi { get; set; }
}
2. 下行 Kafka 指令实体（C# 生产给 EMQX 下发设备）
csharp
运行
/// <summary>
/// 平台下发指令，发送到Kafka iot/device/down
/// EMQX消费后转MQTT推送给设备
/// </summary>
public class DeviceDownCommand
{
/// <summary>目标设备MQTT主题 device/{deviceId}/cmd</summary>
public string TargetTopic { get; set; }
/// <summary>下发指令内容</summary>
public object Payload { get; set; }
/// <summary>下发QoS 1保证送达</summary>
public int Qos { get; set; } = 1;
}
/// <summary>设备控制指令示例：继电器开关</summary>
public class RelayCmd
{
/// <summary>1开 0关</summary>
public int Status { get; set; }
/// <summary>操作来源</summary>
public string Operator { get; set; }
/// <summary>指令唯一ID，用于设备应答回执</summary>
public string CmdId { get; set; }
}
四、Kafka 生产者封装（C# 下发设备指令）
csharp
运行
using Confluent.Kafka;
using Newtonsoft.Json;
public class KafkaDeviceProducer
{
private readonly IProducer<string, string> _producer;
private const string DownTopic = "iot/device/down";
public KafkaDeviceProducer(string bootstrapServers)
{
var config = new ProducerConfig
{
BootstrapServers = bootstrapServers,
// 百万设备高可靠配置
Acks = Acks.All,
QueueBufferingMaxMessages = 100000,
QueueBufferingMaxMs = 5,
BatchSize = 16384,
CompressionType = CompressionType.Lz4,
LingerMs = 5,
RetryBackoffMs = 100,
MessageSendMaxRetries = 3
};
_producer = new ProducerBuilder<string, string>(config).Build();
}
/// <summary>
/// 下发控制指令给设备
/// </summary>
/// <param name="deviceId">目标设备编号</param>
/// <param name="cmd">指令内容</param>
public async Task SendDeviceCommandAsync(string deviceId, object cmd)
{
var downMsg = new DeviceDownCommand
{
TargetTopic = $"device/{deviceId}/cmd",
Payload = cmd,
Qos = 1
};
string json = JsonConvert.SerializeObject(downMsg);
// key使用deviceId，保证同一设备消息落到同一分区，顺序不乱
var msg = new Message<string, string>
{
Key = deviceId,
Value = json
};
try
{
var result = await _producer.ProduceAsync(DownTopic, msg);
Console.WriteLine($"下发指令成功 Device:{deviceId} Partition:{result.Partition} Offset:{result.Offset}");
}
catch (ProduceException<string, string> ex)
{
// 生产失败日志+重试/告警
Console.WriteLine($"下发指令失败 {ex.Error.Reason}");
throw;
}
}
// 程序退出释放
public void Dispose()
{
_producer.Flush(TimeSpan.FromSeconds(3));
_producer.Dispose();
}
}
五、Kafka 消费者后台服务（.NET 8 Worker Service，消费设备上行数据）
1. Worker 服务主体（百万设备批量消费优化）
csharp
运行
using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using StackExchange.Redis;
public class DeviceUpConsumeWorker : BackgroundService
{
private readonly ILogger<DeviceUpConsumeWorker> _logger;
private readonly IDatabase _redisDb;
private IConsumer<string, string> _consumer;
private const string UpTopic = "iot/device/up";
private const string ConsumerGroup = "iot_device_up_group";
// 批量消费缓冲，百万级减少DBIO
private readonly List<DeviceUpMessage> _batchBuffer = new();
private const int BatchThreshold = 200;
public DeviceUpConsumeWorker(ILogger<DeviceUpConsumeWorker> logger, IConnectionMultiplexer redis)
{
_logger = logger;
_redisDb = redis.GetDatabase();
InitKafkaConsumer();
}
private void InitKafkaConsumer()
{
var conf = new ConsumerConfig
{
BootstrapServers = "192.168.1.100:9092,192.168.1.101:9092,192.168.1.102:9092",
GroupId = ConsumerGroup,
AutoOffsetReset = AutoOffsetReset.Earliest,
// 手动提交offset，处理完数据再提交，防止丢消息
EnableAutoCommit = false,
FetchMinBytes = 1024,
FetchMaxWaitMs = 100,
MaxPollRecords = 500,
SessionTimeoutMs = 10000,
HeartbeatIntervalMs = 3000
};
_consumer = new ConsumerBuilder<string, string>(conf).Build();
_consumer.Subscribe(UpTopic);
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_logger.LogInformation("设备上行消费服务启动");
while (!stoppingToken.IsCancellationRequested)
{
try
{
var consumeResult = _consumer.Consume(stoppingToken);
string msgJson = consumeResult.Message.Value;
var upMsg = JsonConvert.DeserializeObject<DeviceUpMessage>(msgJson);
_batchBuffer.Add(upMsg);
// 达到批量阈值批量处理入库
if (_batchBuffer.Count >= BatchThreshold)
{
await BatchHandleDeviceDataAsync();
// 批量处理完成提交offset
_consumer.Commit(consumeResult);
_batchBuffer.Clear();
}
}
catch (ConsumeException ex)
{
_logger.LogError(ex, "Kafka消费异常");
}
catch (Exception ex)
{
_logger.LogError(ex, "消息处理异常");
}
}
// 退出前处理剩余缓存
if (_batchBuffer.Any())
await BatchHandleDeviceDataAsync();
_consumer.Close();
}
/// <summary>批量解析、存储、更新设备在线状态</summary>
private async Task BatchHandleDeviceDataAsync()
{
foreach (var msg in _batchBuffer)
{
// 1. 更新设备在线状态到Redis，TTL30s（心跳周期）
await _redisDb.StringSetAsync($"device:online:{msg.DeviceId}", "1", TimeSpan.FromSeconds(30));
// 2. 根据MQTT主题区分业务逻辑
if (msg.MqttTopic.Contains("heartbeat"))
{
HandleHeartbeat(msg);
}
else if (msg.MqttTopic.Contains("data"))
{
await HandleSensorDataAsync(msg);
}
else if (msg.MqttTopic.Contains("alarm"))
{
await HandleDeviceAlarmAsync(msg);
}
}
// TODO: Dapper批量插入数据库，减少IO压力
}
private void HandleHeartbeat(DeviceUpMessage msg)
{
_logger.LogDebug($"设备心跳 {msg.DeviceId}");
}
private async Task HandleSensorDataAsync(DeviceUpMessage msg)
{
var payload = JsonConvert.DeserializeObject<DeviceDataPayload>(msg.Payload.ToString());
// 阈值告警判断
if (payload.Temp > 60)
{
_logger.LogWarning($"设备{msg.DeviceId}温度超阈值 {payload.Temp}℃");
// 推送告警、短信/websocket通知业务
}
// 批量写入时序库/MySQL
}
private async Task HandleDeviceAlarmAsync(DeviceUpMessage msg)
{
// 存储告警记录、推送前台告警
}
}
2. Program.cs 注册服务
csharp
运行
using Microsoft.Extensions.DependencyInjection;
using StackExchange.Redis;
var host = Host.CreateDefaultBuilder(args)
.ConfigureServices((ctx, services) =>
{
// Redis 百万设备在线缓存
var redisConn = ConnectionMultiplexer.Connect("192.168.1.200:6379");
services.AddSingleton(redisConn);
// Kafka生产者单例
services.AddSingleton(new KafkaDeviceProducer("192.168.1.100:9092,192.168.1.101:9092"));
// 设备上行消费后台服务
services.AddHostedService<DeviceUpConsumeWorker>();
})
.Build();
await host.RunAsync();
六、业务 WebApi 调用生产者下发指令示例
csharp
运行
[ApiController]
[Route("api/device")]
public class DeviceController : ControllerBase
{
private readonly KafkaDeviceProducer _kafkaProducer;
public DeviceController(KafkaDeviceProducer kafkaProducer)
{
_kafkaProducer = kafkaProducer;
}
/// <summary>远程控制设备继电器开关</summary>
[HttpPost("relay/control")]
public async Task<IActionResult> RelayControl(string deviceId, int status)
{
var cmd = new RelayCmd
{
CmdId = Guid.NewGuid().ToString("N"),
Status = status,
Operator = "admin"
};
await _kafkaProducer.SendDeviceCommandAsync(deviceId, cmd);
return Ok(new { code = 0, msg = "指令已下发" });
}
}
七、百万级集群扩容 & 性能保障方案
1. EMQX 扩容
多节点集群，LB 四层负载均衡承载百万 MQTT 连接
开启共享订阅、会话持久化、消息磁盘持久化，防止断连丢失消息
限制单设备消息速率，防恶意设备刷屏压垮 Kafka
2. Kafka 集群扩容核心
分区数 = 消费服务最大并发数，建议 64 分区起步，不足再扩
生产开启 Lz4 压缩，大幅降低磁盘 / 网络 IO
日志保留时间根据业务设置（传感器数据保留 7 天，告警永久）
独立磁盘存放 kafka-log，禁止和系统盘共用
3. C# 服务水平扩容
Worker 消费服务多实例部署，Kafka 自动分配分区，并发处理上行消息
批量消费 + 批量入库，避免单条消息频繁操作数据库
Redis 缓存设备在线状态、设备基础信息，减少 DB 查询
高并发场景引入时序数据库（InfluxDB/TimescaleDB）存储传感器时序数据，减轻 MySQL 压力
4. 消息可靠性保障
MQTT QoS1：设备→EMQX 至少送达一次
Kafka Acks.All：消息落多副本再返回成功
C# 手动 Commit Offset：业务处理完成再提交，宕机不丢消息
EMQX 开启消息持久化，服务重启恢复会话与未下发指令
八、故障兜底方案
Kafka 不可用：EMQX 缓存消息，Kafka 恢复后批量补发
C# 消费服务宕机：重启后从上次提交 offset 继续消费，无消息丢失
设备离线下发指令：EMQX 持久化会话，设备重连自动推送缓存指令
消息堆积监控：监控 kafka 分区 lag，堆积过高自动扩容消费实例、告警通知运维
A:...
Q:用简单通俗的话讲清楚3D建模的模型如何代码控制模型
A:..
大白话讲：代码怎么控制 3D 模型
先打个比方：
3D 模型 = 舞台上的人偶（有身体、五官、手脚）
引擎（Unity/UE）= 舞台
代码 = 遥控器
一、先搞懂模型身上 2 套关键东西，代码就是操作它们
1. Transform（位置 / 旋转 / 大小，所有人偶都自带）
每个人偶有 3 组数据，代码直接改数字就能动：
1坐标 Position：x/y/z，代表人偶站在哪
2旋转 Rotation：x/y/z，代表人偶朝哪边、头歪不歪
3缩放 Scale：x/y/z，代表人变大变小
举例代码逻辑（人话）：
1每帧让 Position 的 X 加 0.1 → 人偶一直往右走
2旋转 Y 轴慢慢增加 → 人偶原地转圈
3Scale 全部 ×2 → 模型瞬间变大一倍
2. Mesh + 骨骼骨骼动画（人形 / 怪物模型专属）
1普通静态模型（箱子、石头）：只有 Mesh 网格，只用 Transform 控制挪位置；
2带骨骼模型（人物、动物）：模型内部藏一套 “骨头架子”，骨头连着皮肉。
代码不直接拽模型皮肉，而是控制骨头。
二、代码控制模型分 4 种最常见操作，由浅到深
1. 最简单：挪位置、转圈、放大缩小（静态物体通用）
逻辑：代码实时修改 Transform 的三个数值
例子：
碰到按钮，角色瞬移到门口 = 直接给 Position 赋值固定坐标
怪物巡逻左右来回走 = 循环增减 X 坐标
BOSS 暴怒体型变大 = 修改 Scale 数值
2. 控制骨骼，让人做动作（人物 / 角色核心）
模型提前在 Blender/Maya 做好动画文件（走路、跳、攻击），导入引擎后：
引擎有个动画管理器（Unity Animator / UE Animation Blueprint）
代码只干 2 件事：
1切换播放哪个动画：比如按 W 键 → 播放走路动画；按鼠标左键 → 播放攻击动画
2给动画传参数：比如速度 = 0 播放站立，速度 > 0 播放走路
底层原理：
动画文件里存好了每根骨骼每一帧的角度，代码切换动画后，引擎自动每帧修改所有骨骼的旋转，皮肉跟着骨头动。
代码不用手动去调几十根骨头，只需要切换动画片段。
3. 实时扭曲骨骼（进阶，比如转头、弯腰）
不播放预制动画，代码直接改单根骨骼旋转：
比如玩家鼠标移动，代码持续修改 “脖子骨骼” 的旋转值 → 人物头跟着鼠标转。
4. 改变模型外观（换皮肤、隐身、变色）
模型身上绑着材质球（决定颜色、贴图、透明度）
代码控制材质：
1材质透明度改成 0 → 模型隐身
2替换贴图文件 → 角色换皮肤
3修改材质颜色数值 → 受伤变红、回血变绿
三、完整流程举例（玩家操控人物）
1美术在 Blender 做人物模型 + 骨骼，导出走路、跳跃动画；
2导入 Unity，引擎自动识别骨骼、动画、Transform；
3写 C# 代码：
检测键盘 W/A/S/D 输入；
有输入就增加人物 Position 实现移动；
移动时给动画参数 speed=1，自动播放走路；松开按键 speed=0 切回站立；
按空格，触发跳跃动画，同时向上修改 Y 坐标。
四、一句话总结核心逻辑
3D 模型本身只是一堆顶点数据，代码不直接改顶点，只改两套东西：
1整体 Transform：控制整个模型在哪、多大、朝哪；
2内部骨骼 / 材质：控制模型四肢动作、外观颜色；
引擎负责根据代码给的数值，实时刷新画面上的模型样子。
A:...
下面分两套最主流：Unity C#（手游最常用）完整可运行示例，全程注释大白话，包含：模型加载、实例化、动画控制、按键切换动作。
配套说明：美术提前导出人物 FBX，带骨骼 + 动画（Idle 待机、Walk 走路、Attack 攻击），导入 Unity 后自动生成 Animator 动画控制器。
一、Unity C# 完整代码：加载模型 + 控制动作
1. 动态加载人物模型（Resources 加载方式，手游通用简易方案）
csharp
运行
using UnityEngine;
public class CharacterControllerDemo : MonoBehaviour
{
// 1. 资源路径：你的人物模型放在 Resources/Player 文件夹下，预制体名字 Hero
private string modelPath = "Player/Hero";
// 生成出来的人物对象
private GameObject playerObj;
// 动画控制器组件，用来控制动作
private Animator playerAnim;
// 动画参数名（和Animator面板参数保持一致）
private const string SpeedParam = "MoveSpeed";
private const string AttackTrigger = "Attack";
void Start()
{
// ========== 第一步：加载3D人物模型预制体 ==========
// Resources.Load 读取Resources文件夹内模型资源
GameObject heroPrefab = Resources.Load<GameObject>(modelPath);
if (heroPrefab == null)
{
Debug.LogError("模型加载失败，检查路径是否正确！");
return;
}
// ========== 第二步：在场景生成（实例化）模型 ==========
// 实例化，放在坐标(0,0,0)，无旋转
playerObj = Instantiate(heroPrefab, new Vector3(0,0,0), Quaternion.identity);
// 获取模型身上的动画组件
playerAnim = playerObj.GetComponent<Animator>();
if (playerAnim == null)
{
Debug.LogError("模型没有挂载Animator动画组件！");
}
}
void Update()
{
if (playerObj == null || playerAnim == null) return;
// ========== 第三步：代码控制模型整体移动（修改Transform坐标） ==========
// 获取摇杆/键盘输入 A D 左右，W S前后
float horizontal = Input.GetAxis("Horizontal");
float vertical = Input.GetAxis("Vertical");
// 移动方向向量
Vector3 moveDir = new Vector3(horizontal, 0, vertical).normalized;
float moveSpeed = 3f; // 移动速度
// 修改模型位置，控制人物走动
playerObj.transform.Translate(moveDir * moveSpeed * Time.deltaTime, Space.World);
// 让人物朝向移动方向
if (moveDir.magnitude > 0.1f)
{
playerObj.transform.rotation = Quaternion.LookRotation(moveDir);
}
// ========== 第四步：代码控制动作动画切换 ==========
// 把移动数值传给动画，自动切换 待机/走路
playerAnim.SetFloat(SpeedParam, moveDir.magnitude);
// 鼠标左键按下，播放攻击动画（触发器）
if (Input.GetMouseButtonDown(0))
{
playerAnim.SetTrigger(AttackTrigger);
}
}
}
2. 关键配套设置（必须做，否则代码无效）
人物 FBX 导入 Unity，勾选 Import Animation，分离出 Idle、Walk、Attack 三段动画；
创建 Animator Controller，拖入三段动画，设置参数：
Float：MoveSpeed
Trigger：Attack
设置动画切换条件：
MoveSpeed > 0.1：Idle → Walk
MoveSpeed ≤ 0.1：Walk → Idle
将 Animator Controller 赋值给人物模型身上的 Animator 组件；
人物预制体放入项目 Resources/Player/ 文件夹，路径对应代码 Player/Hero。
3. 代码对应功能解释（通俗）
Resources.Load：从硬盘读取 3D 模型文件（加载模型）；
Instantiate：把模型放到游戏场景里显示出来；
playerObj.transform：控制模型整体位置、旋转、大小；
Animator：模型骨骼动画管理器；
SetFloat("MoveSpeed", 值)：传递移动速度，引擎自动控制骨骼播放走路 / 待机；
SetTrigger("Attack")：触发一次攻击骨骼动作。
二、补充：Addressable 异步加载（商业手游正式项目标准，不卡帧）
Resources 是简易写法，上线手游用 Addressables 异步加载大模型，不会卡顿：
csharp
运行
using UnityEngine;
using UnityEngine.AddressableAssets;
using UnityEngine.ResourceManagement.AsyncOperations;
public class LoadModelAsync : MonoBehaviour
{
// 模型地址（Addressables面板设置）
public string heroAddress = "Assets/Prefabs/Player/Hero.prefab";
private GameObject playerObj;
private Animator playerAnim;
void Start()
{
// 异步加载模型，不阻塞游戏
AsyncOperationHandle<GameObject> handle = Addressables.InstantiateAsync(heroAddress);
handle.Completed += LoadModelFinish;
}
// 模型加载完成回调
void LoadModelFinish(AsyncOperationHandle<GameObject> handle)
{
if (handle.Status == AsyncOperationStatus.Succeeded)
{
playerObj = handle.Result;
playerAnim = playerObj.GetComponent<Animator>();
Debug.Log("模型异步加载完成，可以控制动作");
}
}
}
三、额外扩展代码：单独控制单根骨骼（转头瞄准）
比如射击游戏，只控制头部骨骼跟随鼠标，不影响全身动画：
csharp
运行
// 在原有脚本Update里追加
void ControlHeadBone()
{
// 获取模型里名为 "Head" 的头部骨骼
Transform headBone = playerObj.transform.Find("Bone/Head");
if (headBone == null) return;
// 根据鼠标横向移动旋转头部
float mouseX = Input.GetAxis("Mouse X") * 1.5f;
// 限制头部转动角度防止穿模
float clampAngle = Mathf.Clamp(mouseX, -30, 30);
headBone.localEulerAngles = new Vector3(0, clampAngle, 0);
}
四、UE5 简易蓝图逻辑等效 C++ 伪代码（备选）
如果用虚幻引擎，逻辑思路完全一致：
1 StaticMesh/SkeletalMesh 资源异步加载；
2 SpawnActor 创建模型实例；
3 USkeletalMeshComponent 控制骨骼；
4 AnimInstance 传参数切换走路、攻击动画；
核心逻辑和 Unity 一模一样，只是 API 名称不同。
五、整体流程总结（代码执行顺序）
1 读取硬盘 3D 模型资源（加载）
2 在场景生成模型实体
3 获取动画组件 Animator
4 Update 每帧：
修改 Transform 实现模型移动旋转
传入移动数值自动切换待机 / 走路骨骼动画
点击触发攻击骨骼动画
5 进阶：单独查找骨骼 Transform，实时局部旋转头部、手臂