如何让SQL Server 2008 Service Broker对接Mosquitto MQTT Broker？

整体流程说明 #

SQL Server本身无法直接和MQTT Broker通信，需借助SQL Service Broker作为中间消息队列，配合外部桥接程序完成转发：触发器触发后将消息发送至Service Broker队列，外部程序监听该队列并将消息转发到Mosquitto。

一、SQL Service Broker 基础配置 #

1. 启用目标数据库的Service Broker #

-- 启用Service Broker，若有活动连接需先切换单用户模式
ALTER DATABASE YourDatabaseName SET SINGLE_USER WITH ROLLBACK IMMEDIATE;
ALTER DATABASE YourDatabaseName SET ENABLE_BROKER;
ALTER DATABASE YourDatabaseName SET MULTI_USER;

2. 创建消息类型 #

定义消息的格式验证规则，根据实际需求选择：

-- 若消息为合法XML
CREATE MESSAGE TYPE [MQTTMessage] VALIDATION = WELL_FORMED_XML;

-- 若消息为纯文本/任意格式
CREATE MESSAGE TYPE [MQTTMessage] VALIDATION = NONE;

3. 创建契约 #

指定消息的交互模式（仅发起方发送，无需响应）：

CREATE CONTRACT [MQTTContract]
([MQTTMessage] SENT BY INITIATOR);

4. 创建目标队列与服务 #

触发器发送的消息将投递到该队列，外部程序从这里读取消息：

-- 创建目标队列（无需自动激活，由外部程序监听）
CREATE QUEUE [MQTTQueue];

-- 创建绑定队列与契约的服务（对应触发器中的'MosquitoService'）
CREATE SERVICE [MosquitoService]
ON QUEUE [MQTTQueue] ([MQTTContract]);

5. 修正触发器代码 #

原代码缺少契约指定与必要的对话结束操作，修正后示例：

CREATE TRIGGER TriggerName ON YourTableName
AFTER INSERT
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @dialog_handle UNIQUEIDENTIFIER;
    -- 从inserted表中提取要发送的数据，示例为拼接字符串
    DECLARE @Message NVARCHAR(MAX) = (SELECT CONCAT('新增数据ID：', ID) FROM inserted);

    -- 初始化对话
    BEGIN DIALOG @dialog_handle
    TO SERVICE 'MosquitoService'
    ON CONTRACT [MQTTContract]
    WITH ENCRYPTION = OFF; -- 本地部署无需加密

    -- 发送消息
    SEND ON CONVERSATION @dialog_handle
    MESSAGE TYPE [MQTTMessage] (@Message);

    -- 结束对话（无响应需求，及时清理资源）
    END CONVERSATION @dialog_handle;
END

二、编写外部MQTT桥接程序 #

需编写一个后台程序（如Windows服务），监听Service Broker队列并转发消息到Mosquitto。以下是C#示例（基于.NET Framework 4.0+，兼容SQL Server 2008环境）：

1. 依赖准备 #

通过NuGet安装MQTTnet库（MQTT客户端工具包）。

2. 示例代码 #

using System;
using System.Data.SqlClient;
using MQTTnet;
using MQTTnet.Client;

namespace SQLToMQTTBridge
{
    class Program
    {
        static void Main(string[] args)
        {
            // 配置参数
            string sqlConnStr = "Server=你的SQL服务器地址;Database=目标数据库名;Integrated Security=True;";
            string mqttBrokerAddr = "localhost"; // Mosquitto地址，跨机器需改为实际IP
            int mqttBrokerPort = 1883; // Mosquitto默认端口
            string mqttPublishTopic = "sql/server/notification"; // MQTT主题

            // 初始化MQTT客户端
            var mqttFactory = new MqttFactory();
            var mqttClient = mqttFactory.CreateMqttClient();
            var mqttOptions = new MqttClientOptionsBuilder()
                .WithTcpServer(mqttBrokerAddr, mqttBrokerPort)
                .Build();

            // 连接MQTT Broker
            mqttClient.ConnectAsync(mqttOptions).Wait();
            Console.WriteLine("已连接到MQTT Broker");

            // 持续监听Service Broker队列
            using (SqlConnection sqlConn = new SqlConnection(sqlConnStr))
            {
                sqlConn.Open();
                Console.WriteLine("已连接到SQL Server");

                while (true)
                {
                    // 等待队列消息，超时5秒轮询
                    string receiveSql = @"
                        WAITFOR(
                            RECEIVE TOP(1) message_body, conversation_handle
                            FROM MQTTQueue
                        ), TIMEOUT 5000;";

                    using (SqlCommand cmd = new SqlCommand(receiveSql, sqlConn))
                    {
                        using (SqlDataReader reader = cmd.ExecuteReader())
                        {
                            if (reader.Read())
                            {
                                // 读取消息内容
                                string messageContent = reader["message_body"].ToString();
                                Guid convHandle = (Guid)reader["conversation_handle"];

                                // 发布到MQTT
                                var mqttMsg = new MqttApplicationMessageBuilder()
                                    .WithTopic(mqttPublishTopic)
                                    .WithPayload(messageContent)
                                    .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
                                    .Build();

                                mqttClient.PublishAsync(mqttMsg).Wait();
                                Console.WriteLine($"已转发消息：{messageContent}");

                                // 结束对话，清理SQL资源
                                using (SqlCommand endConvCmd = new SqlCommand(
                                    "END CONVERSATION @convHandle;", sqlConn))
                                {
                                    endConvCmd.Parameters.AddWithValue("@convHandle", convHandle);
                                    endConvCmd.ExecuteNonQuery();
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}

3. 部署程序 #

将程序编译后，通过Windows服务工具（如sc create）注册为后台服务，确保其随系统自动启动。

三、Mosquitto Broker 配置（可选） #

若桥接程序与Mosquitto不在同一机器，需修改mosquitto.conf：

• 将bind_address设为0.0.0.0（允许外部连接）
• 若无需认证，确保allow_anonymous true（默认开启）

内容的提问来源于stack exchange，提问作者Ken