当前位置：首页 > news >正文

Docker Compose 构建 EMQX 集群实现mqqt 和websocket

news 文章来源：https://blog.csdn.net/qq_27229113/article/details/144818774 2025/5/18 21:26:24

EMQX 集群化管理mqqt真香

容器构建

vim docker-compose.yml

version: '3'services:emqx1:image: emqx:5.8.3container_name: emqx1environment:- "EMQX_NODE_NAME=emqx@node1.emqx.io"- "EMQX_CLUSTER__DISCOVERY_STRATEGY=static"- "EMQX_CLUSTER__STATIC__SEEDS=[emqx@node1.emqx.io,emqx@node2.emqx.io]"healthcheck:test: ["CMD", "/opt/emqx/bin/emqx", "ctl", "status"]interval: 5stimeout: 25sretries: 5networks:emqx-bridge:aliases:- node1.emqx.ioports:- 1883:1883- 8083:8083- 8084:8084- 8883:8883- 18083:18083 # volumes:#   - $PWD/emqx1_data:/opt/emqx/dataemqx2:image: emqx:5.8.3container_name: emqx2environment:- "EMQX_NODE_NAME=emqx@node2.emqx.io"- "EMQX_CLUSTER__DISCOVERY_STRATEGY=static"- "EMQX_CLUSTER__STATIC__SEEDS=[emqx@node1.emqx.io,emqx@node2.emqx.io]"healthcheck:test: ["CMD", "/opt/emqx/bin/emqx", "ctl", "status"]interval: 5stimeout: 25sretries: 5networks:emqx-bridge:aliases:- node2.emqx.io# volumes:#   - $PWD/emqx2_data:/opt/emqx/datanetworks:emqx-bridge:driver: bridge

启动

docker-compose up -d

集群状态

#查看集群状态
docker exec -it emqx1 sh -c "emqx ctl cluster status"#验证
telnet 192.168.0.15 1883
#内网
nc -zv  192.168.0.15 1883 #账户
admin
#默认密码
public

服务开放端口

1883,8083,8084,8883,18083

端口占用

EMQX 默认使用以下端口，请确保这些端口未被其他应用程序占用，并按照需求开放防火墙以保证 EMQX 正常运行。

端口	协议	描述
1883	TCP	MQTT over TCP 监听器端口，主要用于未加密的 MQTT 连接。
8883	TCP	MQTT over SSL/TLS 监听器端口，用于加密的 MQTT 连接。
8083	TCP	MQTT over WebSocket 监听器端口，使 MQTT 能通过 WebSocket 进行通信。
8084	TCP	MQTT over WSS (WebSocket over SSL) 监听器端口，提供加密的 WebSocket 连接。
18083	HTTP	EMQX Dashboard 和 REST API 端口，用于管理控制台和 API 接口。
4370	TCP	Erlang 分布式传输端口，根据节点名称不同实际端口可能是 BasePort (4370) + Offset。
5370	TCP	集群 RPC 端口（在 Docker 环境下为 5369），根据节点名称不同实际端口可能是 BasePort (5370) + Offset。

前端js示

<!DOCTYPE html>
<html>
<head><title>MQTT WebSocket Test</title><script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
</head>
<body><script>// 使用提供的客户端 ID 或生成一个唯一 IDconst clientId = 'emqx_NjI4MT2';// 配置 WebSocket MQTT broker 地址const host = 'ws://127.0.0.1:8083/mqtt';// MQTT 连接选项const options = {keepalive: 60, // 心跳时间间隔clientId: clientId,protocolId: 'MQTT', // 协议 IDprotocolVersion: 5, // 使用 MQTT 5 协议clean: true, // 是否清除会话reconnectPeriod: 1000, // 重连间隔时间 (ms)connectTimeout: 30 * 1000, // 连接超时时间 (ms)username: 'admin', // 设置用户名password: 'public', // 设置密码will: {topic: 'pushRanking/1',payload: 'Connection Closed abnormally..!',qos: 0,retain: false},};console.log('Connecting mqtt client');// 连接到 MQTT Brokerconst client = mqtt.connect(host, options);// 连接成功回调client.on('connect', () => {console.log('Connected to MQTT broker');// 订阅主题 pushRanking/#，支持通配符client.subscribe('pushRanking/1', { qos: 0 }, (err) => {if (!err) {console.log('Subscribed to topic: pushRanking/1');} else {console.error('Failed to subscribe:', err);}});});// 处理接收到的消息client.on('message', (topic, message) => {console.log(`Received message from topic "${topic}": ${message.toString()}`);});// 连接错误回调client.on('error', (err) => {console.log('Connection error:', err);client.end();});// 重新连接回调client.on('reconnect', () => {console.log('Reconnecting...');});// 连接关闭回调client.on('close', () => {console.log('Connection closed');});// 模拟消息发布以测试接收setTimeout(() => {client.publish('pushRanking/1', JSON.stringify({ msg: 'hello' }), { qos: 0 });}, 5000);</script>
</body>
</html>

后端

package emqximport ("fmt"mqtt "github.com/eclipse/paho.mqtt.golang""testing""time"
)func TestMQTT(t *testing.T) {// 创建 EMQX 客户端实例client := NewEMQXClient("tcp://127.0.0.1:1883", "test-client", "admin", "QfTzLy3cop9NOGWj")// 连接到 EMQXif err := client.Connect(); err != nil {fmt.Printf("Failed to connect: %v\n", err)return}defer client.Disconnect()// 订阅主题client.Subscribe("testtopic/#", 1, func(client mqtt.Client, msg mqtt.Message) {fmt.Printf("Message received: %s\n", msg.Payload())})// 发布消息client.Publish("testtopic/1", 1, false, "Hello from Golang!")// 保持连接一段时间以接收消息time.Sleep(10 * time.Second)
}/*长连接的场景 DEMOfunc main() {client := emqxclient.NewEMQXClient("tcp://broker.emqx.io:1883", "test-client", "", "")if err := client.Connect(); err != nil {fmt.Printf("Failed to connect: %v\n", err)return}// 使用 defer 确保程序退出时断开连接defer client.Disconnect()// 订阅主题client.Subscribe("test/topic", 1, func(client mqtt.Client, msg mqtt.Message) {fmt.Printf("Received message: %s\n", msg.Payload())})// 发布消息client.Publish("test/topic", 1, false, "Hello from Golang!")// 捕获退出信号signalChan := make(chan os.Signal, 1)signal.Notify(signalChan, os.Interrupt, syscall.SIGTERM)fmt.Println("Running... Press Ctrl+C to exit.")<-signalChanfmt.Println("Exiting...")
}*//*JS 调用如下
import mqtt from 'mqtt';const brokerURL = 'ws://broker.emqx.io:8083/mqtt'; // WebSocket 连接地址
const clientID = `mqttjs_${Math.random().toString(16).substr(2, 8)}`;// 创建客户端
const client = mqtt.connect(brokerURL, {clientId: clientID,username: '', // 如需要认证，填入用户名password: '', // 如需要认证，填入密码
});// 连接事件
client.on('connect', () => {console.log('Connected to EMQX');// 订阅主题client.subscribe('test/topic', (err) => {if (!err) {console.log('Subscribed to topic: test/topic');} else {console.error('Failed to subscribe:', err);}});// 发布消息client.publish('test/topic', 'Hello from JavaScript!');
});// 接收消息事件
client.on('message', (topic, message) => {console.log(`Received message on topic "${topic}": ${message.toString()}`);
});// 错误事件
client.on('error', (err) => {console.error('Connection error:', err);
});*/

common封装调用

package emqximport ("fmt""time"mqtt "github.com/eclipse/paho.mqtt.golang"
)type EMQXClient struct {client mqtt.Client
}// NewEMQXClient 初始化 EMQX 客户端
func NewEMQXClient(broker string, clientID string, username string, password string) *EMQXClient {opts := mqtt.NewClientOptions().AddBroker(broker).SetClientID(clientID).SetUsername(username).SetPassword(password).SetKeepAlive(60 * time.Second).SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {fmt.Printf("Received message on topic: %s, message: %s\n", msg.Topic(), msg.Payload())}).SetPingTimeout(1 * time.Second)client := mqtt.NewClient(opts)return &EMQXClient{client: client}
}// Connect 连接到 EMQX
func (c *EMQXClient) Connect() error {token := c.client.Connect()if token.Wait() && token.Error() != nil {return token.Error()}fmt.Println("Connected to EMQX broker")return nil
}// Publish 发布消息
func (c *EMQXClient) Publish(topic string, qos byte, retained bool, payload interface{}) error {token := c.client.Publish(topic, qos, retained, payload)token.Wait()return token.Error()
}// Subscribe 订阅主题
func (c *EMQXClient) Subscribe(topic string, qos byte, callback mqtt.MessageHandler) error {token := c.client.Subscribe(topic, qos, callback)token.Wait()return token.Error()
}// Unsubscribe 取消订阅
func (c *EMQXClient) Unsubscribe(topics ...string) error {token := c.client.Unsubscribe(topics...)token.Wait()return token.Error()
}// Disconnect 断开连接
func (c *EMQXClient) Disconnect() {c.client.Disconnect(250)fmt.Println("Disconnected from EMQX broker")
}

Docker Compose 构建 EMQX 集群实现mqqt 和websocket

EMQX 集群化管理mqqt真香目录 #目录 /usr/emqx 容器构建 vim docker-compose.yml version: 3services:emqx1:image: emqx:5.8.3container_name: emqx1environment:- "EMQX_NODE_NAMEemqxnode1.emqx.io"- "EMQX_CLUSTER__DISCOVERY_STRATEGYstatic"- …...

编程日记 2025/1/1 22:28:25

Spring 过滤器：OncePerRequestFilter 应用详解

在Web应用中，过滤器（Filter）是一个强大的工具，它可以在请求到达目标资源之前或响应返回客户端之前对请求或响应进行拦截和处理。然而，在某些情况下，我们可能希望确保过滤器逻辑在一次完整的HTTP请求中仅执行…...

编程日记 2025/1/1 22:25:18

3.CSS字体属性

3.1字体系列 CSS使用font-family属性定义文本的字体系列。 p{font-family:"微软雅黑"} div{font-family:Arial,"Microsoft Yahei",微软雅黑} 3.2字体大小 css使用font-size属性定义字体大小 p{ font-size:20px; } px(像素)大小是我们网页的最常用的单…...

编程日记 2025/1/1 22:22:12

微信小程序单选多选radio/checkbox 纯代码分享

单选按钮 <radio-group class"radiogroup" bindchange"radioChange"> <label class"radio" wx:for"{{items}}"> <radio value"{{item.name}}" checked"{{item.checked}}" /> {{item.value}} &…...

编程日记 2025/1/1 22:20:09

k8s 部署meilisearch UI

https://github.com/riccox/meilisearch-ui 拉取镜像 sudo docker pull riccoxie/meilisearch-ui:latestk8s 部署 apiVersion: v1 kind: Service metadata:name: meilisearch-uinamespace: meilisearch spec:type: NodePortselector:app: meilisearch-uiports:- port: 24900…...

编程日记 2025/1/1 22:17:59

gitlab 还原合并请求

事情是这样的： 菜鸡从 test 分支切了个名为 pref-art 的分支出来，发布后一机灵，发现错了，于是在本地用 git branch -d pref-art 将该分支删掉了。之后切到了 prod 分支，再切出了一个相同名称的 pref-art 分支出来&…...

编程日记 2025/1/1 22:11:48

ChatGPT最新版本“o3”的概要

o3简介 o3于2024年12月20日发布——也就是OpenAI 12天直播的最后一天。目前处于安全性测试阶段。它是o1的继任者，旨在处理更复杂的推理任务。o3特别针对数学、科学和编程等领域进行了优化。 o3在多项基准测试中表现出色。例如，在ARC-AGI基准测试中&…...

编程日记 2025/1/1 22:07:44

uniapp——App下载文件，保存、打开文件（二）

uniapp如何下载文件、保存、打开文件时光荏苒，2024即将过去！ 迈向2025，祝大家新的一年工作顺利、万事如意，少一点BUG，涨一点工资…↖(ω)↗ 文章目录 uniapp如何下载文件、保存、打开文件下载文件保存并打开文件处理 …...

编程日记 2025/1/1 22:05:42

Postman接口测试05|实战项目笔记

目录一、项目接口概况二、单接口测试-登录接口：POST 1、正例 2、反例 ①姓名未注册 ②密码错误 ③姓名为空 ④多参 ⑤少参 ⑥无参三、批量运行测试用例四、生成测试报告 1、Postman界面生成 2、Newman命令行生成五、token鉴权（“…...

编程日记 2025/1/1 22:04:41

【paddle】初次尝试

张量张量是 paddlepaddle， torch， tensorflow 等 python 主流机器学习包中唯一通货变量，因此应当了解其基本的功能。张量 paddle.Tensor 与 numpy.array 的转化 import paddle as paddle import matplotlib.pyplot as plt apaddle.to_t…...

编程日记 2025/1/1 22:00:37

01-2023年上半年软件设计师考试java真题解析

1.真题内容在某系统中，类 Interval(间隔) 代表由下界(lower bound(边界))上界(upper bound )定义的区间。要求采用不同的格式显示区间范围。如［lower bound , upper bound ]、[ lower bound … upper bound ]、[ lower bou nd - upper bound &#x…...

编程日记 2025/1/1 21:59:36

一文讲清楚CSS3新特性

文章目录一文讲清楚CSS3新特性1. 新增选择器特性2. 新增的样式3. 新增布局方式一文讲清楚CSS3新特性 1. 新增选择器特性层次选择器(div~p)选择前面有div的p元素伪类选择器 :first-of-type 表示⼀组同级元素中其类型的第⼀个元素:last-of-type 表示⼀组同级元素中其类型的最…...

编程日记 2025/1/1 21:57:32

系统设计案例：设计 Spotify

https://levelup.gitconnected.com/system-design-interview-question-design-spotify-4a8a79697dda 这是一道系统设计面试题，即设计 Spotify。在真正的面试中，你通常会关注应用程序的一两个主要功能，但在本文中，我想从高层次概述…...

编程日记 2025/1/1 21:54:29

太速科技-633-4通道2Gsps 14bit AD采集PCie卡

4通道2Gsps 14bit AD采集PCie卡一、板卡概述二、性能指标板卡功能参数内容 ADC 芯片型号 AD9689 路数 4路ADC， 采样率 2Gsps 数据位 14bit 数字接口 JESD204B 模拟接口交流耦合模拟输入 1V 连接器 6路 SMA 输入阻抗 50Ω 模拟指…...

编程日记 2025/1/1 21:53:26

图片叠加拖拽对比展示效果实现——Vue版

图片叠加拖拽对比展示效果实现——Vue版项目中遇见一个需求：2张图片按竖线分割，左右两侧分别展示对应图片，通过滚动条拖动对应展示图片区域；; 网上搜索了下，没有找到直接可用的组件，这里自己封装了一个次功…...

编程日记 2025/1/1 21:52:23

结合长短期记忆网络（LSTM）和无迹卡尔曼滤波器（UKF）的技术在机器人导航和状态估计中的应用前景

结合长短期记忆网络（LSTM）和无迹卡尔曼滤波器（UKF）的技术在机器人导航和状态估计中具有广泛的应用前景。如有滤波、导航方面的代码定制需求，可通过文末卡片联系作者获得帮助文章目录结合LSTM和UKF的背景结合LSTM和UKF的优势应用实例研究现状MATLAB代码示例结论结合LSTM和…...

编程日记 2025/1/1 21:49:18

【MATLAB APP Designer】小波阈值去噪（第一期）

代码原理及流程小波阈值去噪是一种信号处理方法，用于从信号中去除噪声。这种方法基于小波变换，它通过将信号分解到不同的尺度和频率上来实现。其基本原理可以分为以下几个步骤： （1）小波变换：首先对含噪信…...

编程日记 2025/1/1 21:46:08

ClickHouse副本搭建

一. 副本概述副本的目的主要是保障数据的高可用性，ClickHouse中的副本没有主从之分。所有的副本都是平等的。副本写入流程： 二. 副本搭建 1. 实验环境 hadoop1(192.168.47.128) hadoop2(192.168.47.129)2. 修改配置文件修改两台主机/etc/click…...

编程日记 2025/1/1 21:45:07

K3知识点

提示：文章文章目录前言一、顺序队列和链式队列题目顺序队列和链式队列的定义和特性实际应用场景顺序表题目链式队列二、AVL树三、红黑树四、二叉排序树五、树的概念题目1左子树右子树前序遍历、中序遍历，后序遍历先根遍历、中根遍历左孩子右孩子题目…...

编程日记 2025/1/1 21:43:05

cocos creator 3.x版本如何添加打开游戏时首屏加载进度条

前言项目有一个打开游戏时添加载入进度条的需求。这个功能2.X版本是自带的，不知为何在3.X版本中移除了。实现先说一下解决思路，就是在引擎源码加载场景的位置插入一个方法，然后在游戏入口HTML处监听即可。 1.找到对应源码脚本在coco…...

编程日记 2025/1/1 21:42:04

Fama MacBeth两步法与多因子模型的回归检验

Fama MacBeth两步法与多因子模型的回归检验 – 潘登同学的因子投资笔记本文观点来自最近学习的石川老师《因子投资：方法与实践》一书文章目录 Fama MacBeth两步法与多因子模型的回归检验 -- 潘登同学的因子投资笔记多因子回归检验时序回归检验截面回归检验Fama–…...

编程日记 2025/1/1 21:38:00

IDEA 搭建 SpringBoot 项目之配置 Maven

目录 1?配置 Maven 1.1?打开 settings.xml 文件1.2?配置本地仓库路径1.3?配置中央仓库路径1.4?配置 JDK 版本1.5?重新下载项目依赖 2?配置 idea 2.1?在启动页打开设置2.2?配置 Java Compiler2.3?配置 File Encodings2.4?配置 Maven2.5?配置 Auto Import2.6?配置 C…...

编程日记 2025/1/1 21:35:58

node.js之---事件循环机制

事件循环机制 Node.js 事件循环机制（Event Loop）是其核心特性之一，它使得 Node.js 能够高效地处理大量并发的 I/O 操作。Node.js 基于非阻塞 I/O，使用事件驱动的模型来实现异步编程。事件循环是 Node.js 实现异步编程的基础&…...

编程日记 2025/1/1 21:33:51

Python OpenAI 库开发指南：从入门到实战精通

在人工智能（AI）领域，OpenAI无疑是全球最受瞩目的机构之一。它推出的GPT系列模型、DALLE等创新技术，正在深刻改变各行各业。作为Python开发者，我们该如何快速上手并高效利用OpenAI的API，成为了提升个人竞争力…...

编程日记 2025/1/1 21:32:49

flash-attention保姆级安装教程

FlashAttention安装教程 FlashAttention 是一种高效且内存优化的注意力机制实现，旨在提升大规模深度学习模型的训练和推理效率。高效计算：通过优化 IO 操作，减少内存访问开销，提升计算效率。内存优化：降低内存占用…...

编程日记 2025/1/1 21:30:45

送给一年编程道路的自己

回望过去一年在编程道路上的成长与收获，是一个很有意义的过程。总结自己这一年的编程经历，不仅可以帮助你更清晰地了解自己的进步和不足，还能为未来的发展指引方向。以下是一些可能的收获，供你参考： 1. 技能提升语言…...

编程日记 2025/1/1 21:29:41

LeRobot（1）

Train python lerobot/scripts/train.py \ policyact \ envaloha \ env.taskAlohaInsertion-v0 \ dataset_repo_idlerobot/aloha_sim_insertion_human \ load_data一直报错，忘记截图了，反正是ssh报错，下不下来，网…...

编程日记 2025/1/1 21:26:35

C++ 设计模式：组合模式（Composite Pattern）

链接：C 设计模式链接：C 设计模式 - 迭代器模式链接：C 设计模式 - 职责链模式组合模式（Composite Pattern）是一种结构型设计模式，它允许你将对象组合成树形结构来表示“部分-整体”的层次结构。组合模式…...

编程日记 2025/1/1 21:22:31

OpenHarmony源码编译后烧录镜像教程，RK3566鸿蒙开发板演示

本文介绍瑞芯微主板/开发板编译OpenHarmony源码后烧录镜像的教程，触觉智能Purple Pi OH鸿蒙开发板演示。搭载了瑞芯微RK3566四核处理器，树莓派卡片电脑设计，支持开源鸿蒙OpenHarmony3.2-5.0系统，适合鸿蒙开发入门学习。编译源码…...

编程日记 2025/1/1 21:21:27

强化学习（1）

Reinforcement Learning Goal-directed learing from ineraction with the environment. 1. Basic Element 基本元素 1.1 Agent 玩家 1.2 Environment 1.3 Goal 2. Main Element 主要元素 2.1 State 2.2 Action 状态与行为往复 2.3 Reward 目标：最大化总…...

编程日记 2025/1/1 21:17:22

目录