集成测试支持自动启动本地 RocketMQ 并硬失败

新增 RocketMQ 下载/启动/关闭流程，测试环境自包含。所有依赖 RocketMQ 的测试统一硬失败处理，提升问题发现及时性。Producer/Consumer 测试断言增强，校验消息状态与路由信息。优化 RocketMQ 配置，允许磁盘 100% 使用率，提升稳定性。测试类结构调整，便于维护。

Author: nnhy

.github/workflows/test.yml

@@ -13,6 +13,13 @@ jobs:
 
     steps:
     - uses: actions/checkout@v4
+
+    - name: Setup Java
+      uses: actions/setup-java@v4
+      with:
+        distribution: 'temurin'
+        java-version: '17'
+
     - name: Setup dotNET
       uses: actions/setup-dotnet@v4
       with:
@@ -22,6 +29,56 @@ jobs:
           8.x
           9.x
           10.x
+
+    - name: Download & Start RocketMQ
+      run: |
+        RMQ_VERSION="5.3.1"
+        wget -q "https://mirrors.tuna.tsinghua.edu.cn/apache/rocketmq/${RMQ_VERSION}/rocketmq-all-${RMQ_VERSION}-bin-release.zip" -O /tmp/rocketmq.zip
+        unzip -q /tmp/rocketmq.zip -d /tmp/
+        mv /tmp/rocketmq-all-${RMQ_VERSION}-bin-release /tmp/rocketmq
+
+        # 创建 broker.conf
+        cat > /tmp/rocketmq/conf/broker_test.conf << 'CONF'
+brokerClusterName=DefaultCluster
+brokerName=broker-a
+brokerId=0
+deleteWhen=04
+fileReservedTime=48
+brokerRole=ASYNC_MASTER
+flushDiskType=ASYNC_FLUSH
+autoCreateTopicEnable=true
+listenPort=10911
+storePathRootDir=/tmp/rocketmq/store
+brokerIP1=127.0.0.1
+diskMaxUsedSpaceRatio=100
+diskSpaceWarningLevelRatio=100
+CONF
+
+        # 启动 NameServer
+        nohup /tmp/rocketmq/bin/mqnamesrv > /tmp/namesrv.log 2>&1 &
+        echo "等待 NameServer 启动..."
+        timeout 30 sh -c 'while ! grep -q "boot success" /tmp/namesrv.log 2>/dev/null; do sleep 1; done'
+        echo "NameServer 已启动"
+
+        # 启动 Broker
+        nohup /tmp/rocketmq/bin/mqbroker -c /tmp/rocketmq/conf/broker_test.conf -n 127.0.0.1:9876 > /tmp/broker.log 2>&1 &
+        echo "等待 Broker 启动..."
+        timeout 40 sh -c 'while ! grep -q "boot success" /tmp/broker.log 2>/dev/null; do sleep 1; done'
+        echo "Broker 已启动"
+
     - name: Build
       run: |
         dotnet build -c Release
+
+    - name: Test
+      env:
+        ROCKETMQ_NAMESERVER: "127.0.0.1:9876"
+      run: |
+        dotnet test XUnitTestRocketMQ/XUnitTest.csproj -c Release --no-build -v n
+
+    - name: Stop RocketMQ
+      if: always()
+      run: |
+        # 清理 RocketMQ 进程
+        pkill -f mqbroker 2>/dev/null || true
+        pkill -f mqnamesrv 2>/dev/null || true

XUnitTestRocketMQ/Cloud/SupportApacheAclTest.cs

@@ -6,6 +6,7 @@
 using NewLife.RocketMQ;
 using NewLife.RocketMQ.Protocol;
 using Xunit;
+using XUnitTest.Integration;
 
 namespace XUnitTest.Cloud;
 
@@ -24,6 +25,7 @@ public class SupportApacheAclTest
     [Fact]
     public void CreateTopicTest()
     {
+        BasicTest.EnsureAvailable();
         using var producer = CreateProducerInstance(DefaultSysTopic);
         producer.Start();
         producer.CreateTopic(TestTopic, 2);
@@ -33,6 +35,7 @@ public void CreateTopicTest()
     [Fact]
     public void PublishMessageTest()
     {
+        BasicTest.EnsureAvailable();
         using var producer = CreateProducerInstance(TestTopic);
         producer.Start();
 

XUnitTestRocketMQ/Consumers/ConsumerTests.cs

@@ -12,12 +12,14 @@ namespace XUnitTest.Consumers;
 
 public class ConsumerTests
 {
-    private static Consumer _consumer;
+    private Consumer _consumer;
 
     [Fact]
-    [System.ComponentModel.DisplayName("Consumer集成测试_拉取消息不抛异常")]
-    public static void ConsumeTest()
+    [System.ComponentModel.DisplayName("Consumer集成测试_启动消费不抛异常_并获取路由信息")]
+    public void ConsumeTest()
     {
+        BasicTest.EnsureAvailable();
+
         var set = BasicTest.GetConfig();
         var consumer = new Consumer
         {
@@ -36,15 +38,21 @@ public static void ConsumeTest()
 
         _consumer = consumer;
 
+        // 等待 Rebalance 后验证路由信息
         Thread.Sleep(3000);
+
+        // 验证消费者已获取到 Broker 路由信息
+        Assert.NotNull(consumer.Brokers);
+        Assert.NotEmpty(consumer.Brokers);
+        XTrace.WriteLine("Consumer 获取到 {0} 个 Broker", consumer.Brokers.Count);
         //foreach (var item in consumer.Clients)
         //{
         //    var rs = item.GetRuntimeInfo();
         //    Console.WriteLine("{0}\t{1}", item.Name, rs["brokerVersionDesc"]);
         //}
     }
 
-    private static Boolean OnConsume(MessageQueue q, MessageExt[] ms)
+    private Boolean OnConsume(MessageQueue q, MessageExt[] ms)
     {
         Console.WriteLine("[{0}@{1}]收到消息[{2}]", q.BrokerName, q.QueueId, ms.Length);
 

XUnitTestRocketMQ/Integration/BasicTest.cs

@@ -14,7 +14,7 @@ public class BasicTest
 {
     private static MqSetting _config;
 
-    /// <summary>获取 RocketMQ 配置</summary>
+    /// <summary>获取 RocketMQ 配置并验证 NameServer 可达。不可达时硬失败。</summary>
     public static MqSetting GetConfig()
     {
         if (_config != null) return _config;
@@ -32,20 +32,26 @@ public static MqSetting GetConfig()
 
             XTrace.WriteLine("RocketMQ配置：{0}", set.NameServer);
 
-            return _config = set;
+            _config = set;
         }
+
+        // 首次获取配置时检查连通性，不可达直接硬失败
+        EnsureAvailable();
+
+        return _config;
     }
 
-    /// <summary>若 NameServer 不可达则跳过当前测试。适合不使用 RocketMqFixture 的集成测试方法。</summary>
-    public static void SkipIfUnavailable()
+    /// <summary>确保 NameServer 可达，不可达时抛出异常（硬失败）。</summary>
+    public static void EnsureAvailable()
     {
         var addr = Environment.GetEnvironmentVariable("ROCKETMQ_NAMESERVER");
         if (String.IsNullOrEmpty(addr)) addr = GetConfig().NameServer;
 
-        Skip.If(!IsReachable(addr),
-            $"无法连接 RocketMQ NameServer [{addr}]，跳过集成测试。\n" +
-            "请检查 Config/RocketMQ.xml 配置或通过 ROCKETMQ_NAMESERVER 环境变量指定地址。\n" +
-            "启动本机 RocketMQ：dotnet run --file scripts/RocketMqSetup.cs");
+        if (!IsReachable(addr))
+            throw new InvalidOperationException(
+                $"无法连接 RocketMQ NameServer [{addr}]。\n" +
+                "请确认 RocketMQ 服务已启动，或通过 ROCKETMQ_NAMESERVER 环境变量指定地址。\n" +
+                "启动本机 RocketMQ：dotnet run --file scripts/RocketMqSetup.cs");
     }
 
     /// <summary>TCP 连通性检测</summary>

XUnitTestRocketMQ/Integration/RocketMqFixture.cs

@@ -26,14 +26,11 @@ public sealed class RocketMqFixture : IAsyncLifetime
 
     private Exception? _initException;
 
-    /// <summary>若 NameServer 不可达则跳过当前测试</summary>
-    public void SkipIfUnavailable()
+    /// <summary>确保 NameServer 可达，不可达时抛出异常（硬失败）</summary>
+    public void EnsureAvailable()
     {
         if (_initException != null)
-            Skip.If(true,
-                $"无法连接 RocketMQ：{_initException.Message}\n" +
-                "请检查 Config/RocketMQ.xml 中的 NameServer 配置，并确认 RocketMQ 服务已启动。\n" +
-                "启动本机 RocketMQ：dotnet run --file scripts/RocketMqSetup.cs");
+            throw _initException;
     }
 
     /// <inheritdoc/>
@@ -45,6 +42,8 @@ public Task InitializeAsync()
 
         NameServerAddress = addr ?? String.Empty;
         _initException    = VerifyConnection(NameServerAddress);
+        // 不可达直接硬失败，不做静默跳过
+        if (_initException != null) throw _initException;
         return Task.CompletedTask;
     }
 

XUnitTestRocketMQ/Integration/RocketMqIntegrationTests.cs

@@ -17,7 +17,7 @@ namespace XUnitTest.Integration;
 /// <remarks>初始化</remarks>
 /// <param name="fixture">RocketMQ Fixture</param>
 [Collection("RocketMQ")]
-public class ProducerIntegrationTests(RocketMqFixture fixture) : IClassFixture<RocketMqFixture>
+public class ProducerIntegrationTests(RocketMqFixture fixture)
 {
     [Fact]
     [DisplayName("发送普通消息_返回SendOK")]
@@ -87,7 +87,7 @@ public async Task PublishMessagesParallel_AllSucceed()
 /// <remarks>初始化</remarks>
 /// <param name="fixture">RocketMQ Fixture</param>
 [Collection("RocketMQ")]
-public class ConsumerIntegrationTests(RocketMqFixture fixture) : IClassFixture<RocketMqFixture>
+public class ConsumerIntegrationTests(RocketMqFixture fixture)
 {
     [Fact]
     [DisplayName("先发再消费_能收到消息")]

XUnitTestRocketMQ/Producers/ProducerTests.cs

@@ -1,5 +1,6 @@
 using NewLife.Log;
 using NewLife.RocketMQ;
+using NewLife.RocketMQ.Protocol;
 using Xunit;
 using XUnitTest.Integration;
 
@@ -11,6 +12,7 @@ public class ProducerTests
     [System.ComponentModel.DisplayName("Producer_创建主题_返回队列数")]
     public void CreateTopic()
     {
+        BasicTest.EnsureAvailable();
         var set = BasicTest.GetConfig();
         var mq = new Producer
         {
@@ -30,9 +32,11 @@ public void CreateTopic()
     }
 
     [Fact]
-    [System.ComponentModel.DisplayName("Producer_发送消息_不抛异常")]
-    public static void ProduceTest()
+    [System.ComponentModel.DisplayName("Producer_发送消息_全部返回SendOK")]
+    public void ProduceTest()
     {
+        BasicTest.EnsureAvailable();
+
         var set = BasicTest.GetConfig();
         using var mq = new Producer
         {
@@ -50,6 +54,7 @@ public static void ProduceTest()
             //var str = Rand.NextString(1337);
 
             var sr = mq.Publish(str, "TagA", null);
+            Assert.Equal(SendStatus.SendOK, sr.Status);
         }
     }
 }

XUnitTestRocketMQ/Trace/MessageTraceTests.cs

@@ -5,6 +5,7 @@
 using NewLife.RocketMQ;
 using NewLife.RocketMQ.Protocol;
 using Xunit;
+using XUnitTest.Integration;
 
 namespace XUnitTest.Trace;
 
@@ -17,6 +18,7 @@ public class MessageTraceTests
     [Fact]
     public void Producer_And_Consumer_With_Trace_Enabled_Should_Work()
     {
+        BasicTest.EnsureAvailable();
         // 使用 ManualResetEvent 来同步测试的完成
         var mre = new ManualResetEvent(false);
 

scripts/RocketMqSetup.cs

@@ -221,6 +221,10 @@ static void WriteBrokerConf()
             listenPort=10911
             storePathRootDir={StoreDir.Replace('\\', '/')}
             brokerIP1=127.0.0.1
+            # 允许磁盘使用率到 100%，避免因磁盘满拒绝写入（测试环境）
+            diskMaxUsedSpaceRatio=100
+            diskSpaceWarningLevelRatio=100
+            logicalDiskSpaceCleanForciblyThreshold=1.0
             """;
 
         File.WriteAllText(BrokerConf, conf, new UTF8Encoding(false));