企业级工作流解决方案（七）--微服务Tcp消息传输模型之消息编解码

　　Tcp消息传输主要参照surging来做的，做了部分裁剪和改动，详细参见：https://github.com/dotnetcore/surging

　　Json-rpc没有定义消息如何传输，因此，Json-Rpc RpcRequest对象和RpcResponse对象需要一个传输载体，这里的传输对象主是TransportMessage，如下代码，这里的Content请求时为RcpRequest对象，答复时为RpcResponse对象，答复时Header一般情况下为空。

/// <summary>/// 传输消息模型。/// </summary>public class TransportMessage{public TransportMessage(){}[MethodImpl(MethodImplOptions.AggressiveInlining)]public TransportMessage(object content,object headers){if (content == null)throw new ArgumentNullException(nameof(content));Content = content;Headers = headers;ContentType = content.GetType().FullName;}[MethodImpl(MethodImplOptions.AggressiveInlining)]public TransportMessage(object content, object headers, string fullName){if (content == null)throw new ArgumentNullException(nameof(content));Headers = headers;Content = content;ContentType = fullName;}/// <summary>/// 消息Id。/// </summary>public string Id { get; set; }/// <summary>/// 消息内容。/// </summary>public object Content { get; set; }/// <summary>/// 消息传输Header/// </summary>public object Headers { get; set; }/// <summary>/// 内容类型。/// </summary>public string ContentType { get; set; }/// <summary>/// 是否调用消息。/// </summary>/// <returns>如果是则返回true，否则返回false。</returns>
        [MethodImpl(MethodImplOptions.AggressiveInlining)]public bool IsInvokeMessage(){return ContentType == MessagePackTransportMessageType.jsonRequestTypeName;}/// <summary>/// 是否是调用结果消息。/// </summary>/// <returns>如果是则返回true，否则返回false。</returns>
        [MethodImpl(MethodImplOptions.AggressiveInlining)]public bool IsInvokeResultMessage(){return ContentType == MessagePackTransportMessageType.jsonResponseTypeName;}/// <summary>/// 获取内容。/// </summary>/// <typeparam name="T">内容类型。</typeparam>/// <returns>内容实例。</returns>
        [MethodImpl(MethodImplOptions.AggressiveInlining)]public T GetContent<T>(){return (T)Content;}/// <summary>/// 获取Header。/// </summary>/// <typeparam name="T">Header类型。</typeparam>/// <returns>Header实例。</returns>
        [MethodImpl(MethodImplOptions.AggressiveInlining)]public T GetHeaders<T>(){return (T)Headers;}/// <summary>/// 创建一个调用传输消息。/// </summary>/// <param name="invokeMessage">调用实例。</param>/// <returns>调用传输消息。</returns>  public static TransportMessage CreateInvokeMessage(JsonRequest invokeMessage,NameValueCollection nameValueCollection){return new TransportMessage(invokeMessage, nameValueCollection, MessagePackTransportMessageType.jsonRequestTypeName){Id = Guid.NewGuid().ToString("N")};}/// <summary>/// 创建一个调用结果传输消息。/// </summary>/// <param name="id">消息Id。</param>/// <param name="invokeResultMessage">调用结果实例。</param>/// <returns>调用结果传输消息。</returns>  public static TransportMessage CreateInvokeResultMessage(string id, JsonResponse jsonResponse,NameValueCollection nameValueCollection){return new TransportMessage(jsonResponse, nameValueCollection, MessagePackTransportMessageType.jsonResponseTypeName){Id = id};}}

　　TransportMessage需要在dotnetty中传输，则需要对TransportMessage进行编码解码

消息编解码器

public interface ITransportMessageEncoder{byte[] Encode(TransportMessage message);
}
public interface ITransportMessageDecoder{TransportMessage Decode(byte[] data);}

Json编解码

　　平时编码中经常用的方式

public sealed class JsonTransportMessageEncoder : ITransportMessageEncoder{#region Implementation of ITransportMessageEncoderpublic byte[] Encode(TransportMessage message){var content = JsonConvert.SerializeObject(message);return Encoding.UTF8.GetBytes(content);}#endregion Implementation of ITransportMessageEncoder
}public sealed class JsonTransportMessageDecoder : ITransportMessageDecoder{#region Implementation of ITransportMessageDecoderpublic TransportMessage Decode(byte[] data){var content = Encoding.UTF8.GetString(data);var message = JsonConvert.DeserializeObject<TransportMessage>(content);if (message.IsInvokeMessage()){message.Content = JsonConvert.DeserializeObject<JsonRequest>(message.Content.ToString());}if (message.IsInvokeResultMessage()){message.Content = JsonConvert.DeserializeObject<JsonResponse>(message.Content.ToString());}return message;}#endregion Implementation of ITransportMessageDecoder}

MessagePack

　　官网地址：https://msgpack.org/

　　贴出代码，不过多的解释

[MessagePackObject]public class MessagePackTransportMessage{public MessagePackTransportMessage(TransportMessage transportMessage){Id = transportMessage.Id;ContentType = transportMessage.ContentType;object contentObject;if (transportMessage.IsInvokeMessage()){contentObject = new MessagePackJsonRequest(transportMessage.GetContent<JsonRequest>());}else if (transportMessage.IsInvokeResultMessage()){contentObject = new MessagePackJsonResponse(transportMessage.GetContent<JsonResponse>());}else{throw new NotSupportedException($"无法支持的消息类型：{ContentType}！");}Content = SerializerUtilitys.Serialize(contentObject);var headersObject = transportMessage.GetHeaders<NameValueCollection>();Headers = SerializerUtilitys.Serialize(JsonConvert.SerializeObject(MessagePackTransportMessageType.NvcToDictionary(headersObject)));}public MessagePackTransportMessage(){}[Key(0)]public string Id { get; set; }[Key(1)]public byte[] Content { get; set; }[Key(2)]public byte[] Headers { get; set; }[Key(3)]public string ContentType { get; set; }[MethodImpl(MethodImplOptions.AggressiveInlining)]public bool IsInvokeMessage(){return ContentType == MessagePackTransportMessageType.jsonRequestTypeName;}[MethodImpl(MethodImplOptions.AggressiveInlining)]public bool IsInvokeResultMessage(){return ContentType == MessagePackTransportMessageType.jsonResponseTypeName;}public TransportMessage GetTransportMessage(){var message = new TransportMessage{ContentType = ContentType,Id = Id,Content = null,Headers = null,};object contentObject;if (IsInvokeMessage()){contentObject =SerializerUtilitys.Deserialize<MessagePackJsonRequest>(Content).GetJsonRequest();}else if (IsInvokeResultMessage()){contentObject =SerializerUtilitys.Deserialize<MessagePackJsonResponse>(Content).GetJsonResponse();}else{throw new NotSupportedException($"无法支持的消息类型：{ContentType}！");}message.Content = contentObject;var headers = SerializerUtilitys.Deserialize<string>(Headers);message.Headers = JsonConvert.DeserializeObject(headers);return message;}
}public sealed class MessagePackTransportMessageEncoder:ITransportMessageEncoder{#region Implementation of ITransportMessageEncoder[MethodImpl(MethodImplOptions.AggressiveInlining)]public byte[] Encode(TransportMessage message){var transportMessage = new MessagePackTransportMessage(message){Id = message.Id,ContentType = message.ContentType,};return SerializerUtilitys.Serialize(transportMessage);}#endregion Implementation of ITransportMessageEncoder}public sealed class MessagePackTransportMessageDecoder : ITransportMessageDecoder{#region Implementation of ITransportMessageDecoder[MethodImpl(MethodImplOptions.AggressiveInlining)]public TransportMessage Decode(byte[] data){var message = SerializerUtilitys.Deserialize<MessagePackTransportMessage>(data);return message.GetTransportMessage();}#endregion Implementation of ITransportMessageDecoder}

ProtoBuffer

　　这个应该听得比较多

[ProtoContract]public class ProtoBufferTransportMessage{public ProtoBufferTransportMessage(TransportMessage transportMessage){Id = transportMessage.Id;ContentType = transportMessage.ContentType;object contentObject;if (transportMessage.IsInvokeMessage()){contentObject = new ProtoBufferJsonRequest(transportMessage.GetContent<JsonRequest>());}else if (transportMessage.IsInvokeResultMessage()){contentObject = new ProtoBufferJsonResponse(transportMessage.GetContent<JsonResponse>());}else{throw new NotSupportedException($"无法支持的消息类型：{ContentType}！");}Content = SerializerUtilitys.Serialize(contentObject);Headers = SerializerUtilitys.Serialize(transportMessage.GetHeaders<NameValueCollection>());}public ProtoBufferTransportMessage(){}[ProtoMember(1)]public string Id { get; set; }[ProtoMember(2)]public byte[] Content { get; set; }[ProtoMember(3)]public byte[] Headers { get; set; }[ProtoMember(4)]public string ContentType { get; set; }public bool IsInvokeMessage(){return ContentType == MessagePackTransportMessageType.jsonRequestTypeName;}public bool IsInvokeResultMessage(){return ContentType == MessagePackTransportMessageType.jsonResponseTypeName;}public TransportMessage GetTransportMessage(){var message = new TransportMessage{ContentType = ContentType,Id = Id,Content = null,Headers = null,};object contentObject;if (IsInvokeMessage()){contentObject =SerializerUtilitys.Deserialize<ProtoBufferJsonRequest>(Content).GetJsonRequest();}else if (IsInvokeResultMessage()){contentObject =SerializerUtilitys.Deserialize<ProtoBufferJsonResponse>(Content).GetJsonResponse();}else{throw new NotSupportedException($"无法支持的消息类型：{ContentType}！");}message.Content = contentObject;message.Headers = SerializerUtilitys.Deserialize<NameValueCollection>(Headers);return message;}
}public sealed class ProtoBufferTransportMessageEncoder : ITransportMessageEncoder{#region Implementation of ITransportMessageEncoderpublic byte[] Encode(TransportMessage message){var transportMessage = new ProtoBufferTransportMessage(message){Id = message.Id,ContentType = message.ContentType,};return SerializerUtilitys.Serialize(transportMessage);}#endregion Implementation of ITransportMessageEncoder}public sealed class ProtoBufferTransportMessageDecoder : ITransportMessageDecoder{#region Implementation of ITransportMessageDecoderpublic TransportMessage Decode(byte[] data){var message = SerializerUtilitys.Deserialize<ProtoBufferTransportMessage>(data);return message.GetTransportMessage();}#endregion Implementation of ITransportMessageDecoder}

转载于:https://www.cnblogs.com/spritekuang/p/10805754.html

企业级工作流解决方案（七）--微服务Tcp消息传输模型之消息编解码相关推荐

详解OSI七层网络模型 TCP/IP四层模型
@TOC这里对OSI七层网络模型和TCP/IP四层模型及其下的各层和TCP协议的设计核心做出总结供大家参考,如有错误欢迎指出讨论! OSI七层网络模型 & TCP/IP四层模型 OSI七层模型 ...
tcp/ip协议中消息传输对帧消息的操作
2019独角兽企业重金招聘Python工程师标准>>> 接口:Framer.java: package com.tcpip;import java.io.IOException; i ...
企业级工作流解决方案（八）--微服务Tcp消息传输模型之服务端处理
服务端启动服务端启动主要做几件事情,1. 从配置文件读取服务配置(主要是服务监听端口和编解码配置),2. 注册编解码器工厂,3. 启动dotnetty监听端口,4. 读取配置文件,解析全局消息处理模 ...
springcloud微服务架构开发实战：分布式消息总线
消息总线的定义前面在1.4.2节中强调过,在微服务架构中,经常会使用REST 服务或基于消息的通信机制. 在3.6节中也详细介绍了消息通信的实现方式.消息总线就是一种基于消息的通信机制. 消息总线是 ...
企业级工作流解决方案（十六）--工作流--工作流实体模型
DDD思想强调先有领域实体模型定义,再有数据库设计,数据库只是做为领域模型的一种持久化介质,但是在工作中,还是习惯性的先做数据库设计,再翻译成领域实体模型.所有我还是以数据库设计为起点,讲解工作流整体 ...
.NET Core微服务之基于EasyNetQ使用RabbitMQ消息队列
Tip: 此篇已加入.NET Core微服务基础系列文章索引一.消息队列与RabbitMQ 1.1 消息队列 "消息"是在两台计算机间传送的数据单位.消息可以非常简单,例如只包含 ...
微服务解决方案_微服务为您提供正确的解决方案
微服务解决方案 I have been writing about Microservices for quite a few years, both its benefits and its dow ...
微服务架构开发实战：分布式消息总线，实现配置信息的自动更新
实现配置信息的自动更新在上一篇文章中节演示了集成Spring Cloud Bus 的过程.在示例中,当微服务实例启动的时候,可以去加载最新的配置信息.当时这种做法有一定的局限性,即只有在应用启动的过 ...
OSI七层与TCP/IP四层模型
2.层次关系一.物理层: 目的:保证原始数据比特流的无误传输. 任务:物理层定义电压.接口.线缆标准.传输距离等.物理层协议定义接口的四个基本功能特性: 机械特性:说明了接口所用 ...

企业级工作流解决方案（七）--微服务Tcp消息传输模型之消息编解码

企业级工作流解决方案（七）--微服务Tcp消息传输模型之消息编解码相关推荐

最新文章

热门文章