Merge pull request #108 from NewLifeX/copilot/support-transaction-message

Add transaction message support to Producer (prepare + commit/rollback)

Author: nnhy

NewLife.RocketMQ/Producer.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using System.Globalization;
 using NewLife.Log;
 using NewLife.Reflection;
 using NewLife.RocketMQ.Client;
@@ -12,6 +13,8 @@ namespace NewLife.RocketMQ;
 /// <summary>生产者</summary>
 public class Producer : MqBase
 {
+    private const Int32 CommitLogOffsetHexLength = 16;
+
     #region 属性
     /// <summary>负载均衡。发布消息时，分发到各个队列的负载均衡算法，默认使用带权重的轮询</summary>
     public ILoadBalance LoadBalance { get; set; }
@@ -229,6 +232,36 @@ public virtual SendResult Publish(Object body, String tags, String keys, Int32 t
 
         return Publish(message, null, timeout);
     }
+
+    /// <summary>发布事务消息（半消息）</summary>
+    /// <param name="message">消息体</param>
+    /// <param name="queue">目标队列</param>
+    /// <param name="timeout"></param>
+    /// <returns></returns>
+    public virtual SendResult PublishTransaction(Message message, MessageQueue queue = null, Int32 timeout = -1)
+    {
+        if (message is null) throw new ArgumentNullException(nameof(message));
+
+        message.Properties["TRAN_MSG"] = "true";
+        message.Properties["PGROUP"] = Group;
+
+        return Publish(message, queue, timeout);
+    }
+
+    /// <summary>发布事务消息（半消息）</summary>
+    /// <param name="body"></param>
+    /// <param name="tags"></param>
+    /// <param name="keys"></param>
+    /// <param name="timeout"></param>
+    /// <returns></returns>
+    public virtual SendResult PublishTransaction(Object body, String tags = null, String keys = null, Int32 timeout = -1)
+    {
+        var message = CreateMessage(body);
+        message.Tags = tags;
+        message.Keys = keys;
+
+        return PublishTransaction(message, null, timeout);
+    }
     #endregion
 
     #region 异步发布消息
@@ -319,6 +352,21 @@ public virtual async Task<SendResult> PublishAsync(Message message, MessageQueue
         return null;
     }
 
+    /// <summary>异步发布事务消息（半消息）</summary>
+    /// <param name="message">消息体</param>
+    /// <param name="queue">目标队列</param>
+    /// <param name="cancellationToken">取消令牌</param>
+    /// <returns></returns>
+    public virtual Task<SendResult> PublishTransactionAsync(Message message, MessageQueue queue = null, CancellationToken cancellationToken = default)
+    {
+        if (message is null) throw new ArgumentNullException(nameof(message));
+
+        message.Properties["TRAN_MSG"] = "true";
+        message.Properties["PGROUP"] = Group;
+
+        return PublishAsync(message, queue, cancellationToken);
+    }
+
     /// <summary>发布消息</summary>
     /// <param name="body"></param>
     /// <returns></returns>
@@ -554,6 +602,60 @@ public virtual void PublishDelay(Object body, DelayTimeLevels level, String tags
     }
     #endregion
 
+    #region 结束事务消息
+    /// <summary>结束事务消息。提交或回滚事务</summary>
+    /// <param name="result">发布事务消息返回结果</param>
+    /// <param name="state">事务状态</param>
+    /// <param name="fromTransactionCheck">是否来自事务回查</param>
+    public virtual void EndTransaction(SendResult result, TransactionState state, Boolean fromTransactionCheck = false)
+    {
+        if (result is null) throw new ArgumentNullException(nameof(result));
+        if (result.Queue == null) throw new ArgumentNullException(nameof(result), "缺少队列信息");
+        if (result.Queue.BrokerName.IsNullOrEmpty()) throw new ArgumentNullException(nameof(result), "缺少BrokerName");
+
+        var header = new EndTransactionRequestHeader
+        {
+            ProducerGroup = Group,
+            TranStateTableOffset = result.QueueOffset,
+            CommitLogOffset = GetCommitLogOffset(result.OffsetMsgId),
+            CommitOrRollback = (Int32)state,
+            FromTransactionCheck = fromTransactionCheck,
+            MsgId = result.MsgId,
+            TransactionId = result.TransactionId,
+        };
+
+        var bk = GetBroker(result.Queue.BrokerName);
+        bk.Invoke(RequestCode.END_TRANSACTION, null, header.GetProperties());
+    }
+
+    /// <summary>异步结束事务消息。提交或回滚事务</summary>
+    /// <param name="result">发布事务消息返回结果</param>
+    /// <param name="state">事务状态</param>
+    /// <param name="fromTransactionCheck">是否来自事务回查</param>
+    /// <param name="cancellationToken">取消令牌</param>
+    /// <returns></returns>
+    public virtual async Task EndTransactionAsync(SendResult result, TransactionState state, Boolean fromTransactionCheck = false, CancellationToken cancellationToken = default)
+    {
+        if (result is null) throw new ArgumentNullException(nameof(result));
+        if (result.Queue == null) throw new ArgumentNullException(nameof(result), "缺少队列信息");
+        if (result.Queue.BrokerName.IsNullOrEmpty()) throw new ArgumentNullException(nameof(result), "缺少BrokerName");
+
+        var header = new EndTransactionRequestHeader
+        {
+            ProducerGroup = Group,
+            TranStateTableOffset = result.QueueOffset,
+            CommitLogOffset = GetCommitLogOffset(result.OffsetMsgId),
+            CommitOrRollback = (Int32)state,
+            FromTransactionCheck = fromTransactionCheck,
+            MsgId = result.MsgId,
+            TransactionId = result.TransactionId,
+        };
+
+        var bk = GetBroker(result.Queue.BrokerName);
+        await bk.InvokeAsync(RequestCode.END_TRANSACTION, null, header.GetProperties(), false, cancellationToken).ConfigureAwait(false);
+    }
+    #endregion
+
     #region 辅助
     /// <summary>
     /// 创建消息，设计于支持用户重载以改变消息序列化行为
@@ -594,8 +696,18 @@ private SendMessageRequestHeader CreateHeader(Message message)
             DefaultTopicQueueNums = DefaultTopicQueueNums
         };
 
+        if (message.Properties.TryGetValue("TRAN_MSG", out var str) && str.ToBoolean()) smrh.SysFlag = (Int32)TransactionState.Prepared;
+
         return smrh;
     }
+
+    private static Int64 GetCommitLogOffset(String offsetMsgId)
+    {
+        if (offsetMsgId.IsNullOrEmpty() || offsetMsgId.Length < CommitLogOffsetHexLength) return 0;
+
+        // OffsetMsgId尾部16位是8字节（Int64）的CommitLogOffset十六进制表示
+        return Int64.TryParse(offsetMsgId.Substring(offsetMsgId.Length - CommitLogOffsetHexLength), NumberStyles.HexNumber, CultureInfo.InvariantCulture, out var rs) ? rs : 0;
+    }
     #endregion
 
     #region 选择Broker队列
@@ -752,4 +864,4 @@ internal void HandleReplyMessage(MessageExt message)
         }
     }
     #endregion
-}
+}

NewLife.RocketMQ/Protocol/EndTransactionRequestHeader.cs

@@ -0,0 +1,50 @@
+using System.Reflection;
+using NewLife.Reflection;
+
+namespace NewLife.RocketMQ.Protocol;
+
+/// <summary>结束事务请求头</summary>
+public class EndTransactionRequestHeader
+{
+    #region 属性
+    /// <summary>生产组</summary>
+    public String ProducerGroup { get; set; }
+
+    /// <summary>事务状态表偏移</summary>
+    public Int64 TranStateTableOffset { get; set; }
+
+    /// <summary>提交日志偏移</summary>
+    public Int64 CommitLogOffset { get; set; }
+
+    /// <summary>提交或回滚标记</summary>
+    public Int32 CommitOrRollback { get; set; }
+
+    /// <summary>是否来自事务回查</summary>
+    public Boolean FromTransactionCheck { get; set; }
+
+    /// <summary>消息编号</summary>
+    public String MsgId { get; set; }
+
+    /// <summary>事务编号</summary>
+    public String TransactionId { get; set; }
+    #endregion
+
+    #region 方法
+    /// <summary>获取属性字典</summary>
+    /// <returns></returns>
+    public IDictionary<String, Object> GetProperties()
+    {
+        var dic = new Dictionary<String, Object>();
+        foreach (var pi in GetType().GetProperties(BindingFlags.Public | BindingFlags.Instance))
+        {
+            if (pi.GetIndexParameters().Length > 0) continue;
+            var name = pi.Name;
+            if (!name.IsNullOrEmpty()) name = Char.ToLowerInvariant(name[0]) + name.Substring(1);
+
+            dic[name] = this.GetValue(pi);
+        }
+
+        return dic;
+    }
+    #endregion
+}

NewLife.RocketMQ/Protocol/TransactionState.cs

@@ -0,0 +1,14 @@
+namespace NewLife.RocketMQ.Protocol;
+
+/// <summary>事务状态</summary>
+public enum TransactionState
+{
+    /// <summary>预备事务（半消息）</summary>
+    Prepared = 4,
+
+    /// <summary>提交事务</summary>
+    Commit = 8,
+
+    /// <summary>回滚事务</summary>
+    Rollback = 12,
+}

Readme.MD

@@ -62,6 +62,21 @@ consumer.OnConsume = (q, messages) =>
 consumer.Start();
 ```  
 
+## 事务消息
+
+通过 `PublishTransaction` 发送半消息，本地事务执行完成后调用 `EndTransaction` 提交或回滚：
+
+```csharp
+var producer = new Producer { Topic = "tx_topic", Group = "tx_group", NameServerAddress = "127.0.0.1:9876" };
+producer.Start();
+
+var sendResult = producer.PublishTransaction("订单创建");
+
+// 本地事务成功时提交，失败时回滚
+producer.EndTransaction(sendResult, TransactionState.Commit);
+// producer.EndTransaction(sendResult, TransactionState.Rollback);
+```
+
 ## 新生命项目矩阵
 各项目默认支持net9.0/netstandard2.1/netstandard2.0/net4.62/net4.5，旧版（2024.0801）支持net4.0/net2.0  
 

XUnitTestRocketMQ/CommandTests.cs

@@ -1,7 +1,9 @@
 using System;
 using System.IO;
+using System.Reflection;
 using NewLife;
 using NewLife.Data;
+using NewLife.RocketMQ;
 using NewLife.RocketMQ.Protocol;
 using NewLife.Serialization;
 using Xunit;
@@ -802,4 +804,45 @@ 22 76 65 72 73 69 6f 6e 22 3a 34 35 33 7d
         var pk = cmd.Payload;
         Assert.Null(pk);
     }
+
+    [Fact]
+    public void CreateHeader_TransactionMessage_SetsPreparedFlag()
+    {
+        var producer = new Producer { Topic = "nx_test", Group = "nx_group" };
+        var message = new Message();
+        message.SetBody("hello");
+        message.Properties["TRAN_MSG"] = "true";
+        message.Properties["PGROUP"] = "nx_group";
+
+        var method = typeof(Producer).GetMethod("CreateHeader", BindingFlags.Instance | BindingFlags.NonPublic);
+        Assert.NotNull(method);
+        var header = (SendMessageRequestHeader)method.Invoke(producer, new Object[] { message });
+
+        Assert.Equal((Int32)TransactionState.Prepared, header.SysFlag);
+    }
+
+    [Fact]
+    public void EndTransactionRequestHeader_ToProperties_UsesCamelCase()
+    {
+        var header = new EndTransactionRequestHeader
+        {
+            ProducerGroup = "nx_group",
+            TranStateTableOffset = 11,
+            CommitLogOffset = 22,
+            CommitOrRollback = (Int32)TransactionState.Commit,
+            FromTransactionCheck = false,
+            MsgId = "msg_1",
+            TransactionId = "tx_1",
+        };
+
+        var ext = header.GetProperties();
+
+        Assert.Equal("nx_group", ext["producerGroup"]);
+        Assert.Equal("11", ext["tranStateTableOffset"]?.ToString());
+        Assert.Equal("22", ext["commitLogOffset"]?.ToString());
+        Assert.Equal("8", ext["commitOrRollback"]?.ToString());
+        Assert.Equal("False", ext["fromTransactionCheck"]?.ToString());
+        Assert.Equal("msg_1", ext["msgId"]);
+        Assert.Equal("tx_1", ext["transactionId"]);
+    }
 }