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

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

news 2025/7/7 12:45:37

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

多模态2025：技术路线“神仙打架”，视频生成冲上云霄

文｜魏琳华编｜王一粟一场大会，聚集了中国多模态大模型的“半壁江山”。智源大会2025为期两天的论坛中，汇集了学界、创业公司和大厂等三方的热门选手，关于多模态的集中讨论达到了前所未有的热度。其中，…...

编程新知 2025/7/5 10:32:33

SkyWalking 10.2.0 SWCK 配置过程

SkyWalking 10.2.0 & SWCK 配置过程 skywalking oap-server & ui 使用Docker安装在K8S集群以外，K8S集群中的微服务使用initContainer按命名空间将skywalking-java-agent注入到业务容器中。 SWCK有整套的解决方案，全安装在K8S群集中。具体可参…...

编程新知 2025/6/15 20:37:19

突破不可导策略的训练难题：零阶优化与强化学习的深度嵌合

强化学习（Reinforcement Learning, RL）是工业领域智能控制的重要方法。它的基本原理是将最优控制问题建模为马尔可夫决策过程，然后使用强化学习的Actor-Critic机制（中文译作“知行互动”机制），逐步迭代求解…...

编程新知 2025/7/5 22:47:32

Oracle查询表空间大小

1 查询数据库中所有的表空间以及表空间所占空间的大小 SELECTtablespace_name,sum( bytes ) / 1024 / 1024 FROMdba_data_files GROUP BYtablespace_name; 2 Oracle查询表空间大小及每个表所占空间的大小 SELECTtablespace_name,file_id,file_name,round( bytes / ( 1024 …...

编程新知 2025/7/5 1:06:55

UE5 学习系列（三）创建和移动物体

这篇博客是该系列的第三篇，是在之前两篇博客的基础上展开，主要介绍如何在操作界面中创建和拖动物体，这篇博客跟随的视频链接如下： B 站视频：s03-创建和移动物体如果你不打算开之前的博客并且对UE5 比较熟的话按照以…...

编程新知 2025/7/6 22:16:09

JDK 17 新特性

#JDK 17 新特性 /**************** 文本块 *****************/ python/scala中早就支持，不稀奇 String json “”" { “name”: “Java”, “version”: 17 } “”"; /**************** Switch 语句 -> 表达式 *****************/ 挺好的&#xff…...

编程新知 2025/7/7 4:55:40

用docker来安装部署freeswitch记录

今天刚才测试一个callcenter的项目，所以尝试安装freeswitch 1、使用轩辕镜像 - 中国开发者首选的专业 Docker 镜像加速服务平台编辑下面/etc/docker/daemon.json文件为 {"registry-mirrors": ["https://docker.xuanyuan.me"] }同时可以进入轩…...

编程新知 2025/7/6 22:52:33