unity 中使用zeroMq和Mqtt 进行通讯

🕗 发布于 2024-11-04 13:06 unity emqx mqtt zeromq

最近我在做一个车上的HMI项目，也就是车机应用，需要与云端和域控进行通信。HMI的功能已经外包了，但消息的统一层留给我自己来做。因为项目组其他人都没有经验，所以这个任务就落到了我头上，尽管我自己也没有太多经验，但也没办法，只能直接上手了。大概架构如下

1、整体设计

设计原则很简单，通过阻塞队列进行交互，因为要和第三方进行对接，这里通过topic和 bytes 进行交互，让他们自己转化看下消息队列定义，也没啥，就是两个阻塞队列，通过队列和第三方交互。

public class MsgQueue 
{
    // 用于存储接收到的二进制消息
    private static BlockingCollection<Msg> recMessageQueue = new BlockingCollection<Msg>();
    private static BlockingCollection<Msg> sendMessageQueue = new BlockingCollection<Msg>();

    public static BlockingCollection<Msg> RecvMessageQueue
    {
        get { return recMessageQueue; }
    }

    public static BlockingCollection<Msg> SendMessageQueue
    {
        get { return sendMessageQueue; }
    }
}

2、zeroMq功能实现

这次的需求是和域控进行通信，主要使用发布订阅模式，也就是我本地需要一个client，一个server。

2.1 zeroMq 介绍

第一次使用zeroMq ，稍微介绍下；ZeroMQ 是一个高性能的异步消息库，旨在简化分布式或多线程应用程序中的消息传递。它提供了一种灵活且高效的方式来进行数据交换，支持多种消息模式，能够在不同的进程、机器和网络之间进行通信。以下是 ZeroMQ 的一些关键特性和概念：ZeroMQ 支持多种消息模式，包括：

请求-响应（Req-Rep）：客户端发送请求，服务器处理并回复。
发布-订阅（Pub-Sub）：发布者发布消息，订阅者接收感兴趣的消息。
推送-拉取（Push-Pull）：用于分布式任务处理，推送端将任务发送到拉取端。
管道（Pipeline）：将多个组件连接起来形成数据处理管道。简单说就是一个TCP通信的框架，可以在本地作为客户端和服务器官方网站：https://zeromq.org/languages/csharp/

2.2 插件介绍

zeromq的在C#上主要是通过netMq库，这玩意好多年不更新了具体地址：https://github.com/zeromq/netmq 这玩意折腾了好久，第一次上手，主要要注意版本，通过Nuget 安装

Install-Package NetMQ

不知道为什么我安装不成功，unity里还是无法使用，我直接拷贝了dll到plugins拷贝的时候注意依赖项，总共有三个，要不然会报错

2.3 代码实现

using NetMQ;
using NetMQ.Sockets;
using System;
using System.Text;
using System.Threading.Tasks;
using UnityEngine;

public class ZeroStarter : MonoBehaviour
{
    private SubscriberSocket subscriber;
    private PublisherSocket publisherSocket;
    private bool isRunning = true;

    void Start()
    {
        AsyncIO.ForceDotNet.Force(); // 确保 NetMQ 在 Unity 中正确工作

        // 初始化发布者
        publisherSocket = new PublisherSocket();
        publisherSocket.Bind("tcp://*:5557");

        subscriber = new SubscriberSocket();
        subscriber.Connect("tcp://localhost:5556");

        // 启动发布和订阅任务
        Task.Run(() => StartPub());
        Task.Run(() => StartSubscriber());
    }

    private void StartPub()
    {
        while (isRunning)
        {
            try
            {
                // 使用 BlockingCollection 的 Take 方法获取消息
                Msg message = MsgQueue.SendMessageQueue.Take();
                publisherSocket.SendMoreFrame(message.Topic).SendFrame(message.Data);
            }
            catch (Exception e)
            {
                Debug.LogError("Error in StartPub: " + e.Message);
            }
        }
        publisherSocket?.Close(); // 确保发布套接字在退出时关闭
    }

    private void StartSubscriber()
    {

        subscriber.Subscribe("");

        while (isRunning)
        {
            try
            {
                // 通过超时和 TryReceiveFrameString 检查订阅消息
                if (subscriber.TryReceiveFrameString(out string topic))
                {
                    byte[] bytes = subscriber.ReceiveFrameBytes();
                    MsgQueue.RecvMessageQueue.Add(new Msg(topic, bytes));
                    string str = Encoding.Default.GetString(bytes);
                    Debug.Log($"{topic} {str}");
                }
                Task.Delay(10).Wait(); // 使用短暂的延迟来避免过度循环
            }
            catch (Exception e)
            {
                Debug.LogError("Error in StartSubscriber: " + e.Message);
            }
        }

        subscriber.Close(); // 手动关闭订阅者套接字
    }

    private void OnDestroy()
    {
        isRunning = false; // 设置标志位，通知任务退出

        // 确保发布套接字关闭
        publisherSocket?.Close();
        NetMQConfig.Cleanup(); // 清理 NetMQ
    }
}

这里有一个注意点就是zeroMq 不支持topic，但是支持 SendMoreFrame(message.Topic).SendFrame(message.Data) 这个有点风险，也很奇怪，但是貌似是官方推荐的做法，不纠结，就这样吧。

3、Mqtt功能实现

3.1 emqx介绍

mqtt也是一个消息队列，主要是用在IOT，简单来说就是一个TCP服务器，消息头会小一些。适合不稳定的网络。我们用的是Emqx ，还行，挺好用。工具客户端使用的是mqttx，下载地址：https://mqttx.app/zh 官方网站：https://www.emqx.com/zh

3.2 unity插件

这里使用的是官方推荐的插件库 sdk 介绍地址：https://docs.emqx.com/zh/emqx/latest/connect-emqx/introduction.html

官方示例：https://github.com/emqx/MQTT-Client-Examples

我这里使用的是：https://github.com/eclipse/paho.mqtt.m2mqtt 我也使用nuget安装了库，也不起作用，不知道为毛，盲猜是版本的问题，打开上面库，下载之后自己编译因为用unity，所以打开了mono的库

编译之后

将生成的dll拷贝到plugins下

3.3 代码实现

using System.Text;
using UnityEngine;
using uPLibrary.Networking.M2Mqtt;
using uPLibrary.Networking.M2Mqtt.Messages;

public class MqttStarter : MonoBehaviour
{
    // Start is called before the first frame update
    void Start()
    {
        string ipAddr = "192.168.3.8";
        int emqxPort = 1883;
        string clientId = "csharpclientid";
        //服务器默认密码是这个
        string username = "username";
        string password = "pwd";
        MqttClient client = new MqttClient(ipAddr, emqxPort,  false, null, null, MqttSslProtocols.None);

        // register to message received
        client.MqttMsgPublishReceived += client_MqttMsgPublishReceived;

        client.Connect(clientId,username,password);

        // 
        client.Subscribe(
            new string[] { "idse/cloud2veh/#" }, 
            new byte[] { MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE });

    }

    private void client_MqttMsgPublishReceived(object sender, MqttMsgPublishEventArgs e)
    {
        MsgQueue.RecvMessageQueue.Add(new Msg(e.Topic, e.Message));
        Debug.Log("mqtt--> " +e.Topic +"  ==== >" + Encoding.Default.GetString(e.Message));
    }
}

这里需要注意的是m2mqtt 会自动开启线程，不需要单独实现线程。其他的都是常规操作。

4、验证

先看下zeromq的收发

验证下Mqtt的收发

5、Java代码zeromq的使用

pom中加入

   <dependency>
      <groupId>org.zeromq</groupId>
      <artifactId>jeromq</artifactId>
      <version>0.5.2</version>
    </dependency>

zeromq 发布者

package org.example;

/**
 * Hello world!
 *
 */
import org.zeromq.ZMQ;
import org.zeromq.ZContext;

public class App {
    public static void main(String[] args) {
        int i = 0;
        // 创建一个ZeroMQ的上下文
        try (ZContext context = new ZContext()) {
            // 创建一个发布者socket
            ZMQ.Socket publisher = context.createSocket(ZMQ.PUB);

            // 绑定到指定的端口
            String address = "tcp://*:5556";
            publisher.bind(address);
            System.out.println("Publisher started at " + address);

            // 持续发送消息
            int messageNumber = 0;
            while (!Thread.currentThread().isInterrupted()) {
                String message = "Message " + messageNumber;
                publisher.sendMore("t:"+ (messageNumber%2));
                publisher.send("abc".getBytes());
                System.out.println("Sent: " + message);

                // 增加消息计数并休眠1秒
                messageNumber++;
                Thread.sleep(1000);
            }

            // 关闭发布者socket
            publisher.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

zeromq 订阅者

package org.example;

import org.zeromq.ZMQ;
import org.zeromq.ZContext;

import java.nio.ByteBuffer;

public class ZmqSubscriber {
    public static void main(String[] args) {
        // 创建一个ZeroMQ的上下文
        try (ZContext context = new ZContext()) {
            // 创建一个订阅者socket
            ZMQ.Socket subscriber = context.createSocket(ZMQ.SUB);

            // 连接到发布者的地址
            String address = "tcp://localhost:5557"; // 确保使用正确的地址
            subscriber.connect(address);
            System.out.println("Subscriber connected to " + address);

            // 订阅所有消息
            subscriber.subscribe("".getBytes()); // 订阅所有消息，可以根据需要指定主题

            // 持续接收消息
            while (!Thread.currentThread().isInterrupted()) {
                // 接收消息
                String topic = subscriber.recvStr(0);
                byte[] message = subscriber.recv(0);
                ByteBuffer wrap = ByteBuffer.wrap(message);

                if (message != null) {
                    System.out.println("Received: " + topic +"  " + wrap.order(java.nio.ByteOrder.LITTLE_ENDIAN).getInt());
                }
            }

            // 关闭订阅者socket
            subscriber.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

6、总结

第一次搞zeromq的消息队列，和平常用的kafka和rocketmq 差的很多，甚至完全不在同一个讨论方向。

还有一些需要研究

在unity中nuget的学习
c# 不同平台的学习
启动后台线程之后无法关闭，导致unity死掉。

byte[] bytes = BitConverter.GetBytes(66666);

C#侧生成的bytes 是小端序列

原文地址：https://blog.csdn.net/perfect2011/article/details/143366173

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Linux练习作业
下一篇：Web应用性能测试工具 - httpstat

关于做完 C# 项目的问题总结 2
c#
阅读更多2024-11-21
Hello-Go
GOPATH 和 GOROOT ：不同于其他语言，go中没有项目的说法，只有包，其中有两个重要的路径，GOROOT 和 GOPATHGo 语言依赖几个关键的环境变量来管理开发环境、编译器、工具链和包管
阅读更多2024-11-21
深入探索Golang的GMP调度机制：源码解析与实现原理
Go语言的GMP调度模型是实现高效并发编程的关键所在。通过深入了解GMP调度机制的实现原理和源码细节，我们可以更好地理解和利用Go语言的并发特性。无论是对于Go语言的开发者还是对于系统性能优化感兴趣的
阅读更多2024-11-21
kafka是如何做到高效读写
消息持久化：Kafka 将消息存储在磁盘上，并且通过顺序写入的方式提高写入性能。消息被追加到日志文件的尾部，避免了随机写操作，从而提高了写入速度。零拷贝技术：利用操作系统的零拷贝特性，数据可以从
阅读更多2024-11-21
Kafka 2.8 源码导读
Kafka 是一个分布式流处理平台，广泛用于实时数据流的处理和传输。Kafka 2.8 版本引入了一些新特性和改进。以下是 Kafka 2.8 源码的导读，帮助你理解其核心组件和实现机制。
阅读更多2024-11-21
【蓝桥杯C/C++】C语言和C++的常量概念与区别分析
2024.11.16-✍53min
阅读更多2024-11-21
40分钟学 Go 语言高并发：开发环境搭建与工程化实践
Windows系统：Linux/Mac系统：2. Go Module依赖管理2.1 主要命令表格命令说明示例go mod init初始化模块go mod tidy整理依赖go mod download
阅读更多2024-11-21
QT如何共享文件+拷贝文件
在这里插入图片描述!
阅读更多2024-11-21
渗透测试导学
渗透测试是干什么？渗透测试的定义和目的：渗透测试是一种通过模拟恶意黑客的攻击方法，来评估计算机网络系统安全性能的评估方法。它的目的是通过识别安全问题，帮助了解当前的安全状况，从而帮助管理人员建立可靠的
阅读更多2024-11-21
45.坑王驾到第九期：Mac安装typescript后tsc命令无效的问题
Mac 安装 typescript 后 tsc 命令无效的问题
阅读更多2024-11-21