All posts by 龙生

微服务之Service Fabric 系列 （一）：概览、环境安装

参考 微软官方文档  service fabric 百家号   大话微服务架构之微服务框架微软ServiceFabric正式开源 一、概述 1.概念 Azure Service Fabric 是一款分布式系统平台，可方便用户轻松打包、部署和管理可缩放的可靠微服务和容器。 Service Fabric 还解决了开发和管理云本机应用程序面临的重大难题。 开发人员和管理员不需解决复杂的基础结构问题，只需专注于实现苛刻的任务关键型工作负荷，即那些可缩放、可靠且易于管理的工作负荷。 Service Fabric 代表了下一代平台，用于生成和管理在容器中运行的企业级单层云规模应用程序。 　　2.微服务 　　　　微服务是指用小型服务集合来构建服务端应用的方法，每个服务在独立的进程中运行，服务间通过特定协议（http,AMQP）通信，每个服务能够独立开发、部署，拥有自己的领域数据模型和领域逻辑，可以有不同的数据存储技术甚至是不同的编程语言。 　　　　微服务的目的在：解耦。解耦是永恒的主题，依赖注入是为了解耦，应用分层是为了解耦，这些可以看做是纵向解耦，而微服务是把应用横向解耦，服务间解耦。 　　　　下图是传统一体式应用开发与微服务开发的对比： 　　　　　　     ①单一式应用包含域特定的功能，通常按照功能层来划分，例如传统三层。 ②单一式应用可通过复制到多个服务器/虚拟机/容器上进行扩展。 ③微服务应用程序将单个功能分隔成单个较小的服务。 ④微服务方法可通过独立部署每个服务而扩展，跨服务器/虚拟机/容器创建这些服务的实例。   service fabric 从三方面来帮助我们使用微服务的方式来构建应用： ①提供系统服务的平台，用于部署、升级、检测和重启失败的服务、发现服务、路由消息、管理状态和监视运行状况 ②能够部署在容器中运行或作为进程运行的应用程序。 ③有助于以微服务形式生成应用程序的生产编程 API：ASP.NET Core、Reliable Actors 和 Reliable Services。 可以选择使用任意代码来生成微服务。 但使用这些 API 不仅可让作业变得更简单，也能更深入地与平台集成。 例如，可以获取运行状况和诊断信息，或利用内置的高可用性 下图是service fabric 与spring cloud 的对比：     二、应用场景 1.高可用服务：Service Fabric 服务通过创建多个辅助服务副本提供快速的故障转移。 节点、进程或单独的服务因硬件或其他故障而不可用时，其中一个辅助副本会提升为主副本，将对服务的损失降到最低。 　　2.可缩放服务：可对单独的服务进行分区，以允许在群集范围内扩大状态。 此外，还可以动态创建并删除单独的服务。 服务可以快速简单地从几个节点上的几个实例扩大到多个节点上的数千个实例，并再次减少，具体取决于资源需求。 可以使用 Service Fabric 来生成这些服务并管理其整个生命周期。 　　3.基于回话的交互式应用：在应用程序（例如在线游戏或即时消息）需要低延迟读取和写入时，Service Fabric 非常有用。 Service Fabric 使你能够生成这些交互式的有状态应用程序，而无需创建一个无状态应用所需的单独存储或缓存。 （这会增加延迟时间并可能产生一致性问题）。 　　4.数据分析和工作流：Service Fabric 的快速读取和写入使必须可靠处理事件或数据流的应用程序成为可能。Service Fabric 还可让应用程序描述处理管道，其中的结果必须能够可靠地传递到下一个处理阶段而不会丢失。这包括交易和财务系统，其中的数据一致性和计算保证至关重要。 三、开发准备 1.Windows下开发环境准备 如果需要在windows上开发并运行service fabric 应用程序，需要安装Service Fabric 运行时、SDK 和工具 具体安装步骤可参照  Windows上安装service fabric 以下是windows 10 下，安装了vs2017的开发人员的环境准备 ①重新运行vs2017安装软件，选择修改，选择azure开发     ②安装 Microsoft Azure […]

龙生   11 Jul 2018
View Details

AMQP

AMQP，即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有 RabbitMQ等。

龙生   11 Jul 2018
View Details

C# string类型和byte[]类型相互转换

string类型转成byte[]： byte[] byteArray = System.Text.Encoding.Default.GetBytes ( str );   byte[]转成string： string str = System.Text.Encoding.Default.GetString ( byteArray );   string类型转成ASCII byte[]： （"01" 转成 byte[] = new byte[]{ 0x30,0x31}） byte[] byteArray = System.Text.Encoding.ASCII.GetBytes ( str );   ASCIIbyte[]转成string： （byte[] = new byte[]{ 0x30, 0x31} 转成"01"） string str = System.Text.Encoding.ASCII.GetString ( byteArray );   byte[]转16进制格式string： new byte[]{ 0x30, 0x31}转成"3031": publicstaticstring ToHexString ( byte[] bytes ) // 0xae00cf => "AE00CF " {string hexString = string.Empty; if ( bytes != null ) { StringBuilder strB = new StringBuilder (); for ( int i = 0; i < bytes.Length; i++ ) { strB.Append ( bytes[i].ToString ( "X2" ) ); } hexString = strB.ToString (); }return hexString; }   16进制格式string 转byte[]： publicstaticbyte[] GetBytes(string hexString, outint discarded) { discarded = 0; string newString = ""; char c;// remove all none A-F, 0-9, charactersfor (int i=0; i<hexString.Length; i++) { c = hexString[i];if (IsHexDigit(c)) newString += c; else discarded++; }// if odd number of characters, discard last characterif (newString.Length % 2 != 0){                discarded++; newString = newString.Substring(0, newString.Length-1);            } int byteLength = newString.Length / 2;byte[] bytes = newbyte[byteLength];string hex;int j = 0;for (int i=0; i<bytes.Length; i++){ hex = new String(new Char[] {newString[j], newString[j+1]}); bytes[i] = HexToByte(hex);                j = j+2; } return bytes; }   from:https://www.cnblogs.com/Maxq/p/5953682.html

龙生   06 Jul 2018
View Details

RabbitMQ使用分析和高可用集群搭建

一、RabbitMQ 基础理解     RabbitMQ，是一个使用 erlang 编写的 AMQP（高级消息队列协议）的服务实现，简单来说，就是一个功能强大的消息队列服务。     概念理解： Producer: 消息发送者 RabbitMQ: Vhost: 相当于分组，每个vhost下数据是隔离的 Exchange: 路由器，接收消息，本根据RoutingKey分发消息 headers：消息头类型 路由器，内部应用 direct：精准匹配类型 路由器 topic：主题匹配类型 路由器，支持正则 模糊匹配 fanout：广播类型 路由器，RoutingKey无效 RoutingKey: 路由规则 Queue: 队列，用于存储消息（消息的目的地） Consumer: 消息消费者 持久化： 一个好的消息队列当然需要消息持久化功能，服务宕机，未消费消息不丢失，RabbitMQ持久化分为Exchange、Queue、Message Exchange 和 Queue 持久化 指持久化Exchange、Queue 元数据，持久化的是自身，服务宕机，Exchange 和 Queue 自身就没有了 Message 持久化 顾名思义 把每一条消息体持久化，服务宕机，消息不丢失 Durable 持久、Transient 临时，Queue新建类似 分析理解: 便于更直观的理解，把 RabbitMQ 的消息流对比与Http Rest接口更家熟悉形象 www.xxx.com/webappPath/trade/getOrder -> getOrder(message) GET RabbitMQ Server：同比 域名 www.xxx.com，只有通过域名才能到达 Server Vhost：同比 /webappPath，一个域名可能指向多个app Exchange：同比 /trade，trade/* 下有多个method，但是需要先到达这个Class RoutingKey：同比 /getOrder，只有完成的 URL 才是有效的，才能确定到具体的方法                  Queue：同比 getOrder(message) 消息的最终目的地 Exchange Type:  同比 GET，但是Rest MethodType是整个URL的Type，而不是 Queue 以上只是为了更好理解，千万不要混淆 Producer / Consumer 就很好理解了，基于AMQP协议链接RabbitMQ Server，发送消息 / 接收消息 二、RabbitMQ 消息确认策略分析 Confrim / Transaction 概念应用 RabbitMQ 提供了两种可靠性的确认策略 Confrim / Transaction，Producer […]

龙生   05 Jul 2018
View Details

Currents 云计算报告：容器正流行，Serverless 未成熟

云计算公司 DigitalOcean 近日发布了2018年6月 Currents 云计算季度报告，该报告对全球 5000 名行业专业人士进行调查。报告显示，随着容器技术和相关管理工具在过去一年的快速发展，49％ 的受访开发者表示目前正在使用容器。Serverless 则依然处于初期阶段，有一大半的受访者表示对 Serverless 不了解。 具体来看，随着容器技术的普及，近一半（49％）的受访者表示他们目前正在使用容器。其中， 40％ 的人只使用容器进行测试和开发，另外 60％ 也会将其用于生产。之所以选择容器，主要是因为其高扩展性，以及能够简化和加速测试。 在具体方案的选择上，容器集群管理系统 Kubernetes 总体上最受欢迎。不过在小公司中，更多地是使用 Docker Swarm（41％ 使用 Swarm ，31％ 使用 Kubernetes ）。 与容器的流行不同，Serverless 似乎仍处于早期阶段，有一大半的受访者表示对 Serverless 不太了解。Serverless 从出现至今已有两三年的时间，但一直没有公认的权威定义。 在使用容器的开发者中，使用 JavaScript 的占大多数，其次是 Python 和 PHP 。 更多详情可查阅完整报告   from: https://www.oschina.net/news/97689/developers-in-cloud

龙生   04 Jul 2018
View Details

SignalR官方底层连接示例（中文）

由于公司项目用到了微软的SignalR，项目后期出现了写Cookie和跨域的问题。所以汉化了官方的示例，给小伙伴们普及一下入门知识。   进来后有另一个用户加入的效果：   发个广播：   设置名字，名字是记录到cookie里的，设置的时候别忘了加path，不然SignalR获取不到。   ……其实SignalR对Cookie和跨域已经提供了支付和示例。   直接上代码：SignalRDemo.zip

龙生   04 Jul 2018
View Details

ConcurrentDictionary 类

表示可由多个线程同时访问的键/值对的线程安全集合。 命名空间:   System.Collections.Concurrent 程序集:  mscorlib（位于 mscorlib.dll） 继承层次结构 System.Object System.Collections.Concurrent.ConcurrentDictionary<TKey, TValue> 语法 C#

1
2
3
4
5
6
7
8
[SerializableAttribute]
[ComVisibleAttribute(false)]
[HostProtectionAttribute(SecurityAction.LinkDemand, Synchronization = true,
ExternalThreading = true)]
public class ConcurrentDictionary<TKey, TValue> : IDictionary<TKey, TValue>,
ICollection<KeyValuePair<TKey, TValue>>, IEnumerable<KeyValuePair<TKey, TValue>>,
IEnumerable, IDictionary, ICollection, IReadOnlyDictionary<TKey, TValue>,
IReadOnlyCollection<KeyValuePair<TKey, TValue>>

类型参数 TKey 字典中的键的类型。 TValue 字典中的值的类型。 构造函数 名称 说明 ConcurrentDictionary<TKey, TValue>() 初始化 ConcurrentDictionary<TKey, TValue> 类的新实例，该实例为空，具有默认的并发级别和默认的初始容量，并为键类型使用默认比较器。 ConcurrentDictionary<TKey, TValue>(IEnumerable<KeyValuePair<TKey, TValue>>) 初始化 ConcurrentDictionary<TKey, TValue> 类的新实例，该实例包含从指定的 IEnumerable<T> 中复制的元素，具有默认的并发级别和默认的初始容量，并为键类型使用默认比较器。 ConcurrentDictionary<TKey, TValue>(IEnumerable<KeyValuePair<TKey, TValue>>, IEqualityComparer<TKey>) 初始化 ConcurrentDictionary<TKey, TValue> 类的新实例，该实例包含从指定的 IEnumerable 中复制的元素，具有默认的并发级别和默认的初始容量，并使用指定的 IEqualityComparer<T>。 ConcurrentDictionary<TKey, TValue>(IEqualityComparer<TKey>) 初始化 ConcurrentDictionary<TKey, TValue> 类的新实例，该实例为空，具有默认的并发级别和容量，并使用指定的System.Collections.Generic.IEqualityComparer<T>。 ConcurrentDictionary<TKey, TValue>(Int32, IEnumerable<KeyValuePair<TKey, TValue>>, IEqualityComparer<TKey>) 初始化 ConcurrentDictionary<TKey, TValue> 类的新实例，该实例包含从指定的 IEnumerable 中复制的元素并使用指定的System.Collections.Generic.IEqualityComparer<T>。 ConcurrentDictionary<TKey, TValue>(Int32, Int32) 初始化 ConcurrentDictionary<TKey, TValue> 类的新实例，该实例为空，具有指定的并发级别和容量，并为键类型使用默认比较器。 ConcurrentDictionary<TKey, TValue>(Int32, Int32, IEqualityComparer<TKey>) 初始化 System.Collections.Concurrent.ConcurrentDictionary<TKey, TValue> 类的新实例，该实例为空，具有指定的并发级别和指定的初始容量，并使用指定的 System.Collections.Generic.IEqualityComparer<T>。 属性 名称 说明 Count 获取包含在 System.Collections.Concurrent.ConcurrentDictionary<TKey, TValue> 中的键/值对的数目。 IsEmpty 获取一个值，该值指示 System.Collections.Concurrent.ConcurrentDictionary<TKey, TValue> 是否为空。 Item[TKey] 获取或设置与指定的键关联的值。 Keys 获得一个包含 System.Collections.Generic.Dictionary<TKey, TValue> 中的键的集合。 Values 获取包含 System.Collections.Generic.Dictionary<TKey, TValue> 中的值的集合。 方法 名称 说明 AddOrUpdate(TKey, TValue, Func<TKey, TValue, TValue>) 如果该键不存在，则将键/值对添加到 ConcurrentDictionary<TKey, TValue> 中；如果该键已经存在，则通过使用指定的函数更新 ConcurrentDictionary<TKey, TValue> 中的键/值对。 AddOrUpdate(TKey, Func<TKey, TValue>, Func<TKey, TValue, TValue>) 如果该键不存在，则使用指定函数将键/值对添加到 ConcurrentDictionary<TKey, TValue>；如果该键已存在，则使用该函数更新 ConcurrentDictionary<TKey, TValue>中的键/值对。 Clear() 将所有键和值从 ConcurrentDictionary<TKey, TValue> 中移除。 ContainsKey(TKey) 确定是否 ConcurrentDictionary<TKey, TValue> 包含指定键。 Equals(Object) 确定指定的对象是否等于当前对象。（继承自 Object。） Finalize() 在垃圾回收将某一对象回收前允许该对象尝试释放资源并执行其他清理操作。（继承自 Object。） GetEnumerator() 返回循环访问 ConcurrentDictionary<TKey, TValue> 的枚举数。 GetHashCode() 作为默认哈希函数。（继承自 Object。） GetOrAdd(TKey, TValue) 如果该键不存在，则将键/值对添加到 ConcurrentDictionary<TKey, TValue> 中。 GetOrAdd(TKey, Func<TKey, TValue>) 如果该键不存在，则通过使用指定的函数将键/值对添加到 ConcurrentDictionary<TKey, TValue> 中。 GetType() 获取当前实例的 Type。（继承自 Object。） MemberwiseClone() 创建当前 Object 的浅表副本。（继承自 Object。） ToArray() 将 System.Collections.Concurrent.ConcurrentDictionary<TKey, TValue> 中存储的键和值对复制到新数组中。 ToString() 返回表示当前对象的字符串。（继承自 Object。） TryAdd(TKey, TValue) 尝试将指定的键和值添加到 System.Collections.Concurrent.ConcurrentDictionary<TKey, TValue> 中。 TryGetValue(TKey, TValue) 尝试从 System.Collections.Concurrent.ConcurrentDictionary<TKey, TValue> 获取与指定的键关联的值。 TryRemove(TKey, TValue) 尝试从 System.Collections.Concurrent.ConcurrentDictionary<TKey, TValue> 中移除并返回具有指定键的值。 TryUpdate(TKey, TValue, TValue) 将指定键的现有值与指定值进行比较，如果相等，则用第三个值更新该键。   示例 C#

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class CD_Ctor
{
        // Demonstrates:
        //      ConcurrentDictionary<TKey, TValue> ctor(concurrencyLevel, initialCapacity)
        //      ConcurrentDictionary<TKey, TValue>[TKey]
        static void Main()
        {
            // We know how many items we want to insert into the ConcurrentDictionary.
            // So set the initial capacity to some prime number above that, to ensure that
            // the ConcurrentDictionary does not need to be resized while initializing it.
            int NUMITEMS = 64;
            int initialCapacity = 101;
 
            // The higher the concurrencyLevel, the higher the theoretical number of operations
            // that could be performed concurrently on the ConcurrentDictionary.  However, global
            // operations like resizing the dictionary take longer as the concurrencyLevel rises.
            // For the purposes of this example, we'll compromise at numCores * 2.
            int numProcs = Environment.ProcessorCount;
            int concurrencyLevel = numProcs * 2;
 
            // Construct the dictionary with the desired concurrencyLevel and initialCapacity
            ConcurrentDictionary<int, int> cd = new ConcurrentDictionary<int, int>(concurrencyLevel, initialCapacity);
 
            // Initialize the dictionary
            for (int i = 0; i < NUMITEMS; i++) cd[i] = i * i;
 
            Console.WriteLine("The square of 23 is {0} (should be {1})", cd[23], 23 * 23);
        }
}

[…]

龙生   03 Jul 2018
View Details

SignalR 跨域解决方案全面

SignalR 分：PersistentConnection和Hub 2种模式。 跨域又分：UseCors和JsonP  2种方法 所以例子写了4种。 核心代码： UseCors

1
2
3
4
5
6
7
8
9
10
11
12
//PersistentConnection Mode
 
            app.UseCors(CorsOptions.AllowAll);
 
            //PersistentConnection Mode
            app.MapSignalR&lt;MyConnection&gt;("/server", new ConnectionConfiguration()
            {
            });
            // 有关如何配置应用程序的详细信息，请访问 http://go.microsoft.com/fwlink/?LinkID=316888
 
            //Hub Mode
            app.MapSignalR("/lcc", new HubConfiguration());

Use JsonP

1
2
3
4
5
6
7
8
9
//PersistentConnection Mode
            app.MapSignalR&lt;MyConnection&gt;("/server", new ConnectionConfiguration()
            {
                EnableJSONP = true
            });
            // 有关如何配置应用程序的详细信息，请访问 http://go.microsoft.com/fwlink/?LinkID=316888
 
            //Hub Mode
            app.MapSignalR("/lcc", new HubConfiguration() { EnableJSONP = true });

Cross  HUb 前端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 建立对应server端Hub class的对象，请注意geffChat的第一个字母要改成小写
        var chat = $.connection.myHub;
 
        $.connection.hub.url = 'http://localhost:2001/lcc';
 
        //把connection打开
        $.connection.hub.start().done(function () {
            chat.server.hello();
        });
 
        // 定义client端的javascript function，供server端hub，通过dynamic的方式，调用所有Clients的javascript function
        chat.client.sendMessage = function (message) {
            //当server端调用sendMessage时，将server push的message数据，呈现在wholeMessage中
            $('#messages').append('&lt;li&gt;' + message + '&lt;/li&gt;');
        };
 
        $("#broadcast").click(function () {
            chat.server.sendMessage($('#msg').val());
            $('#msg').val("");
        });
        $("#btnStop").click(function () {
            $.connection.hub.stop();
        });

Cross Connection 前端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
  var connection = $.connection("http://localhost:2001/server");
 
        connection.received(function (data) {
            $('#messages').append('&lt;li&gt;' + data + '&lt;/li&gt;');
        });
 
        connection.start();
 
        $("#broadcast").click(function () {
            connection.send($('#msg').val());
        });
        $("#btnStop").click(function () {
            connection.stop();
        });

JsonP  HUb

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 建立对应server端Hub class的对象，请注意geffChat的第一个字母要改成小写
        var chat = $.connection.myHub;
 
        $.connection.hub.url = 'http://localhost:16727/lcc';
 
        //把connection打开
        $.connection.hub.start({ jsonp: true }).done(function () {
            chat.server.hello();
        });
 
        // 定义client端的javascript function，供server端hub，通过dynamic的方式，调用所有Clients的javascript function
        chat.client.sendMessage = function (message) {
            //当server端调用sendMessage时，将server push的message数据，呈现在wholeMessage中
            $('#messages').append('&lt;li&gt;' + message + '&lt;/li&gt;');
        };
 
        $("#broadcast").click(function () {
            chat.server.sendMessage($('#msg').val());
            $('#msg').val("");
        });
        $("#btnStop").click(function () {
            $.connection.hub.stop();
        });

JsonP  connection

1
2
3
4
5
6
7
8
9
10
11
12
13
14
var connection = $.connection("http://localhost:16727/server");
 
        connection.received(function (data) {
            $('#messages').append('&lt;li&gt;' + data + '&lt;/li&gt;');
        });
 
        connection.start({ jsonp: true });
 
        $("#broadcast").click(function () {
            connection.send($('#msg').val());
        });
        $("#btnStop").click(function () {
            connection.stop();
        });

代码下载：http://download.csdn.net/detail/shikyoh/9734460 from: https://www.cnblogs.com/shikyoh/p/6272679.html

龙生   03 Jul 2018
View Details

Activemq 平台搭建与C#示列

ActiveMQ ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。 特性  多种语言和协议编写客户端。语言： Java、C、C++、C#、Ruby、Perl、Python、PHP。应用协议：OpenWire、Stomp REST、WS Notification、XMPP、AMQP 完全支持JMS1.1和J2EE 1.4规范 （持久化，XA消息，事务) 对Spring的支持，ActiveMQ可以很容易内嵌到使用Spring的系统里面去，而且也支持Spring2.0的特性 通过了常见J2EE服务器（如 Geronimo、JBoss 4、GlassFish、WebLogic)的测试，其中通过JCA 1.5 resource adaptors的配置，可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上 支持多种传送协议：in-VM、TCP、SSL、NIO、UDP、JGroups、JXTA 支持通过JDBC和journal提供高速的消息持久化 从设计上保证了高性能的集群，客户端-服务器，点对点 支持Ajax 支持与Axis的整合 可以很容易得调用内嵌JMS provider，进行测试 环境准备 我以windows7为列子 windows 7 jdk1.6+ 并在windows下配置好环境变量等，具体可以看 http://jingyan.baidu.com/article/ff41162596a77912e4823716.html 下载最新版activemq  http://activemq.apache.org/activemq-5121-release.html 安装 解压下载好的安装包，至本地任何磁盘 启动activemq服务 启动成功后的界面是 如果出现major.minor version51.0 之类的错误，都是java版本安装错误的问题，安装1.6+以上的版本就能解决 启动成功后，http://localhost:8161/admin，默认用户名和密码admin/admin。如果你想修改用户名和密码的话，在conf/jetty-realm.properties中修改即可。 如果需要修改端口可以在jetty文件中修改 管理员界面如下 ActiviteMQ接收和发送消息基本流程 摘自 http://www.cnblogs.com/hoojo/p/active_mq_jms_apache_activeMQ.html 发送消息的基本步骤： (1)、创建连接使用的工厂类JMS ConnectionFactory (2)、使用管理对象JMS ConnectionFactory建立连接Connection，并启动 (3)、使用连接Connection 建立会话Session (4)、使用会话Session和管理对象Destination创建消息生产者MessageSender (5)、使用消息生产者MessageSender发送消息 消息接收者从JMS接受消息的步骤 (1)、创建连接使用的工厂类JMS ConnectionFactory (2)、使用管理对象JMS ConnectionFactory建立连接Connection，并启动 (3)、使用连接Connection 建立会话Session (4)、使用会话Session和管理对象Destination创建消息接收者MessageReceiver (5)、使用消息接收者MessageReceiver接受消息，需要用setMessageListener将MessageListener接口绑定到MessageReceiver消息接收者必须实现了MessageListener接口，需要定义onMessage事件方法。   C#消息队列写入 ActiveMQ官方网站下载最新版的Apache.NMS，网址：http://activemq.apache.org/nms/download.html，需要下载Apache.NMS和Apache.NMS.ActiveMQ两个bin包 http://www.apache.org/dyn/closer.lua/activemq/apache-nms/1.7.0/Apache.NMS.ActiveMQ-1.7.1-bin.zip 创建C#项目，将这两个DLL添加到项目中。 写入地址如下 写入账号，密码如下 写入代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
namespace ConsoleApplication21
{
    class Program
    {
        static void Main(string[] args)
        {
            String QueuesNameESF = "queue://test.log";
            Uri _uri = new Uri(String.Concat("activemq:tcp://10.58.8.239:61616"));
            IConnectionFactory factory = new ConnectionFactory(_uri);
            using (IConnection conn = factory.CreateConnection("admin", "manager"))
            {
                using (ISession session = conn.CreateSession())
                {
                    IDestination destination = SessionUtil.GetDestination(session, QueuesNameESF);
                    using (IMessageProducer producer = session.CreateProducer(destination))
                    {
                        conn.Start();
                        //可以写入字符串，也可以是一个xml字符串等
                        ITextMessage request = session.CreateTextMessage("messsage");
                        producer.Send(request);
 
                    }
                }
            }
        }
    }

写入后，可以查看   读取代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
using System;
using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.Util;
 
namespace ConsoleApplication21
{
    class Program
    {
        static void Main(string[] args)
        {
            String QueuesNameESF = "queue://test.log";
            Uri _uri = new Uri(String.Concat("activemq:tcp://10.58.8.239:61616?wireFormat.maxInactivityDuration=0"));
            IConnectionFactory factory = new ConnectionFactory(_uri);
            using (IConnection conn = factory.CreateConnection("admin", "manager"))
            {
                using (ISession session = conn.CreateSession())
                {
                    conn.Start();
                    IDestination destination = SessionUtil.GetDestination(session, QueuesNameESF);
                    using (IMessageConsumer consumer = session.CreateConsumer(destination))
                    {
                        ITextMessage message = consumer.Receive() as ITextMessage;
                        Console.Write(message.Text);
                    }
                }
            }
            Console.ReadLine();
        }
    }
}

    from:https://www.cnblogs.com/woxpp/p/5001373.html

龙生   02 Jul 2018
View Details
1 … 182 183 184 … 434