BizTalk自定义管道组件：如何为拆分后的消息添加序列号

你已经把消息拆分的核心逻辑跑通了，就差给每条拆分出的消息加个序列号对吧？其实很容易实现，我们只需要在循环里维护一个计数器，然后把序列号加到消息的上下文或者消息体里就行。下面给你两种常用的实现方式：

方式一：将序列号存入消息上下文（Context） #

这种方式适合后续流程需要读取序列号做业务判断的场景，不会改动消息体本身：

public new void Disassemble(IPipelineContext pContext, IBaseMessage pInMsg) { 
    IBaseMessage Temp = pInMsg; 
    IBaseMessage outMsg; 
    string msg = MessageToString(pInMsg); 
    string[] msgs = msg.Split(new string[] { FlatFileTagSplitterSplitTag }, StringSplitOptions.None); 
    // 初始化序列号计数器，从1开始计数
    int sequenceNumber = 1;
    foreach (string tmpstr in msgs) { 
        if (tmpstr.Trim() != "") { 
            outMsg = pContext.GetMessageFactory().CreateMessage(); 
            outMsg.AddPart("Body", pContext.GetMessageFactory().CreateMessagePart(), true); 
            // 克隆原上下文，避免修改原消息的上下文数据
            outMsg.Context = pInMsg.Context.Clone();
            // 给当前消息添加序列号属性，后续可通过上下文读取
            outMsg.Context.Write("MessageSequenceNumber", sequenceNumber.ToString());
            
            MemoryStream msOut = new MemoryStream(); 
            byte[] bts = System.Text.Encoding.GetEncoding(FlatFileTagSplitterDataEncoding).GetBytes(FlatFileTagSplitterSplitTag + tmpstr); 
            msOut.Seek(0, SeekOrigin.Begin); 
            msOut.Write(bts, 0, bts.Length); 
            msOut.Position = 0; 
            outMsg.BodyPart.Data = msOut; 
            long i = outMsg.BodyPart.Data.Length; //Crazy hack 
            base.Disassemble(pContext, outMsg); 
            IBaseMessage outboundMessage = null; 
            while ((outboundMessage = base.GetNext(pContext)) != null) 
                qOutputMsgs.Enqueue(outboundMessage);
            
            // 每条消息处理完成后，序列号自增
            sequenceNumber++;
        } 
    } 
}

方式二：将序列号直接嵌入消息体内容 #

如果需要把序列号作为消息内容的一部分传递（比如在消息开头标注序号），可以这样修改：

public new void Disassemble(IPipelineContext pContext, IBaseMessage pInMsg) { 
    IBaseMessage Temp = pInMsg; 
    IBaseMessage outMsg; 
    string msg = MessageToString(pInMsg); 
    string[] msgs = msg.Split(new string[] { FlatFileTagSplitterSplitTag }, StringSplitOptions.None); 
    int sequenceNumber = 1;
    foreach (string tmpstr in msgs) { 
        if (tmpstr.Trim() != "") { 
            outMsg = pContext.GetMessageFactory().CreateMessage(); 
            outMsg.AddPart("Body", pContext.GetMessageFactory().CreateMessagePart(), true); 
            outMsg.Context = pInMsg.Context; 
            MemoryStream msOut = new MemoryStream(); 
            // 在消息内容前拼接序列号，格式可根据业务需求调整
            string numberedMsg = $"[{sequenceNumber}] {FlatFileTagSplitterSplitTag}{tmpstr}";
            byte[] bts = System.Text.Encoding.GetEncoding(FlatFileTagSplitterDataEncoding).GetBytes(numberedMsg); 
            msOut.Seek(0, SeekOrigin.Begin); 
            msOut.Write(bts, 0, bts.Length); 
            msOut.Position = 0; 
            outMsg.BodyPart.Data = msOut; 
            long i = outMsg.BodyPart.Data.Length; //Crazy hack 
            base.Disassemble(pContext, outMsg); 
            IBaseMessage outboundMessage = null; 
            while ((outboundMessage = base.GetNext(pContext)) != null) 
                qOutputMsgs.Enqueue(outboundMessage);
            
            sequenceNumber++;
        } 
    } 
}

关键注意点 #

• 计数器sequenceNumber要放在foreach循环外面初始化，确保每条拆分的消息都能拿到连续递增的序号。
• 方式一中使用Context.Clone()是为了保证每个输出消息的上下文独立，不会互相干扰。
• 如果需要全局唯一的序列号（比如跨多个消息拆分任务），可以考虑用静态变量或外部存储维护，但要注意处理线程安全问题。

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