C#实现同Active MQ通讯的方法_C#教程

本文实例讲述了C#实现同Active MQ通讯的方法。分享给大家供大家参考，具体如下：

内容概要：

主要以源码的形式介绍如何用C#实现同Active MQ 的通讯。本文假设你已经正确安装JDK1.6.x，了解Active MQ并有一定的编程基础。

正文：

JMS 程序的最终目的是生产和消费的消息能被其他程序使用，JMS 的 Message 是一个既简单又不乏灵活性的基本格式，允许创建不同平台上符合非JMS 程序格式的消息。
Message 由消息头，属性和消息体三部份组成。
Active MQ支持过滤机制，即生产者可以设置消息的属性（Properties），该属性与消费者端的Selector对应，只有消费者设置的selector与消息的Properties匹配，消息才会发给该消费者。Topic和Queue都支持Selector。

示例代码：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Data;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Imaging;
using System.Windows.Navigation;
using System.Windows.Shapes;
using Apache.NMS;
using System.Diagnostics;
using Apache.NMS.Util;
using System.Windows.Threading;
/*
 * 功能描述：C#使用ActiveMQ示例
 * 修改次数：2
 * 最后更新： by Kagula,2012-07-31
 *
 * 前提条件：
 * [1]apache-activemq-5.4.2
 * [2]Apache.NMS.ActiveMQ-1.5.6-bin
 * [3]WinXP SP3
 * [4]VS2008 SP1
 * [5]WPF工程 With .NET Framework 3.5
 *
 * 启动
 *
 * 不带安全控制方式启动
 * [你的解压路径]\apache-activemq-5.4.2\bin\activemq.bat
 *
 * 安全方式启动
 * 添加环境变量：      ACTIVEMQ_ENCRYPTION_PASSWORD=activemq
 * [你的解压路径]\apache-activemq-5.4.2\bin>activemq xbean:file:../conf/activemq-security.xml
 *
 * Active MQ 管理地址
 * http://127.0.0.1:8161/admin/
 * 添加访问"http://127.0.0.1:8161/admin/"的限制
 *
 * 第一步：添加访问限制
 * 修改D:\apache\apache-activemq-5.4.2\conf\jetty.xml文件
 * 下面这行编码，原
 * <property name="authenticate" value="true" />
 * 修改为
 * <property name="authenticate" value="false" />
 *
 * 第二步：修改登录用户名密码，缺省分别为admin，admin
 * D:\apache\apache-activemq-5.4.2\conf\jetty-realm.properties
 *
 * 用户管理(前提:以安全方式启动ActiveMQ)
 *
 * 在[你的解压路径]\apache-activemq-5.4.2\conf\credentials.properties文件中修改默认的用户名密码
 * 在[你的解压路径]\apache-activemq-5.4.2\conf\activemq-security.xml文件中可以添加新的用户名
 * e.g. 添加oa用户，密码同用户名。
 * <authenticationUser username="oa" password="oa" groups="users,admins"/>
 *
 * 在[你的解压路径]\apache-activemq-5.4.2\conf\activemq-security.xml文件中你还可以设置指定的Topic或Queue
 * 只能被哪些用户组read 或 write。
 *
 *
 * 配置C# with WPF项目
 * 项目的[Application]->[TargetFramework]属性设置为[.NETFramework 3.5](这是VS2008WPF工程的默认设置)
 * 添加[你的解压路径]\Apache.NMS.ActiveMQ-1.5.6-bin\lib\Apache.NMS\net-3.5\Apache.NMS.dll的引用
 * Apache.NMS.dll相当于接口
 *
 * 如果是以Debug方式调试
 * 把[你的解压路径]\Apache.NMS.ActiveMQ-1.5.6-bin\build\net-3.5\debug\目录下的
 * Apache.NMS.ActiveMQ.dll文件复制到你项目的Debug目录下
 * Apache.NMS.ActiveMQ.dll相当于实现
 *
 * 如果是以Release方式调试
 * 参考上文，去取Apache.NMS,Release目录下相应的DLL文件，并复制到你项目的Release目录下。
 *
 *
 * 参考资料
 * [1]《C#调用ActiveMQ官方示例》 http://activemq.apache.org/nms/examples.html
 * [2]《ActiveMQ NMS下载地址》http://activemq.apache.org/nms/activemq-downloads.html
 * [3]《Active MQ在C#中的应用示例》http://www.jb51.net/article/87956.htm
 * [4]《NMS API Reference》http://activemq.apache.org/nms/nms-api.html
 */
namespace testActiveMQSubscriber
{
  /// <summary>
  /// Interaction logic for Window1.xaml
  /// </summary>
  public partial class Window1 : Window
  {
    private static IConnectionFactory connFac;
    private static IConnection connection;
    private static ISession session;
    private static IDestination destination;
    private static IMessageProducer producer;
    private static IMessageConsumer consumer;
    protected static ITextMessage message = null;
    public Window1()
    {
      InitializeComponent();
      initAMQ("MyFirstTopic");
    }
    private void initAMQ(String strTopicName)
    {
      try
      {
        connFac = new NMSConnectionFactory(new Uri("activemq:failover:(tcp://localhost:61616)"));
        //新建连接
        //connection = connFac.CreateConnection("oa","oa");//设置连接要用的用户名、密码
        //如果你要持久“订阅”，则需要设置ClientId，这样程序运行当中被停止，恢复运行时，能拿到没接收到的消息！
        connection.ClientId = "testing listener";
        connection = connFac.CreateConnection();//如果你是缺省方式启动Active MQ服务，则不需填用户名、密码
        //创建Session
        session = connection.CreateSession();
        //发布/订阅模式，适合一对多的情况
        destination = SessionUtil.GetDestination(session, "topic://" + strTopicName);
        //新建生产者对象
        producer = session.CreateProducer(destination);
        producer.DeliveryMode = MsgDeliveryMode.NonPersistent;//ActiveMQ服务器停止工作后，消息不再保留
        //新建消费者对象:普通“订阅”模式
        //consumer = session.CreateConsumer(destination);//不需要持久“订阅”
        //新建消费者对象:持久"订阅"模式：
        //  持久“订阅”后，如果你的程序被停止工作后，恢复运行，
        //从第一次持久订阅开始，没收到的消息还可以继续收
        consumer = session.CreateDurableConsumer(
          session.GetTopic(strTopicName)
          , connection.ClientId, null, false);
        //设置消息接收事件
        consumer.Listener += new MessageListener(OnMessage);
        //启动来自Active MQ的消息侦听
        connection.Start();
      }
      catch (Exception e)
      {
        //初始化ActiveMQ连接失败，往VS2008的Output窗口写入出错信息!
        Debug.WriteLine(e.Message);
      }
    }
    private void SendMsg2Topic_Click(object sender, RoutedEventArgs e)
    {
      //发送消息
      ITextMessage request = session.CreateTextMessage(DateTime.Now.ToLocalTime()+" "+tbMsg.Text);
      producer.Send(request);
    }
    protected void OnMessage(IMessage receivedMsg)
    {
      //接收消息
      message = receivedMsg as ITextMessage;
      //UI线程，显示收到的消息
      Dispatcher.Invoke(DispatcherPriority.Normal, new Action(() =>
      {
        DateTime dt = new DateTime();
        ListBoxItem lbi = new ListBoxItem();
        lbi.Content = DateTime.Now.ToLocalTime() + " " + message.Text;
        lbR.Items.Add(lbi);
      }));
    }
  }
}

队列通讯方式，消费者例子

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.NMS;
using System.Diagnostics;
using log4net;
using Apache.NMS.Util;
using System.Collections;
namespace Cat8637AutoCallServer
{
  public class SMTask
  {
    public String Callee { get; set; }
    public String CheckNumber { get; set; }
    public int Deadline { get; set; }
    public override String ToString()
    {
      return String.Format("Callee={0},CheckNumber={1},Deadline={2}",
        Callee,CheckNumber,Deadline);
    }
  }
  /*
   * 负责接收任务，并把任务放在任务等待队列中。
   */
  public class MQClient
  {
    private static readonly ILog logger = LogManager.GetLogger(typeof(MQClient));
    private static IConnection connection = null;
    private static ISession session = null;
    Queue _voiceSMTasks = new Queue();
    public MQClient()
    {
      try
      {
        IConnectionFactory factory = new NMSConnectionFactory(new Uri("activemq:failover:(tcp://localhost:61616)"));
        //新建连接
        //connection = connFac.CreateConnection("oa","oa");//设置连接要用的用户名、密码
        connection = factory.CreateConnection();
        session = connection.CreateSession();
        IMessageConsumer consumer = session.CreateConsumer(session.GetQueue("TaskIssue_VoiceSM"));
        consumer.Listener += new MessageListener(OnMessage);
        connection.Start();
      }
      catch (Exception ex)
      {
        Debug.WriteLine(ex.Message);
      }
    }
    protected void OnMessage(IMessage receivedMsg)
    {
      IMessage message = receivedMsg as ITextMessage;
      SMTask smTask = new SMTask();
      smTask.Callee = message.Properties["Callee"] as String;
      smTask.CheckNumber = message.Properties["Message"] as String;
      smTask.Deadline = Convert.ToInt32(message.Properties["deadline"] as String);
      logger.Info("Received: "+smTask.ToString());
      lock (_voiceSMTasks)
      {
        _voiceSMTasks.Enqueue(smTask);
      }
    }
    public SMTask GetVoiceSMTask()
    {
      SMTask result = null;
      lock (_voiceSMTasks)
      {
        if (_voiceSMTasks.Count > 0)
        {
          result = _voiceSMTasks.Dequeue() as SMTask;
        }
      }
      return result;
    }
  }
}

队列通讯方式，生产者例子

private void Send_Click(object sender, RoutedEventArgs e)
{
  try
  {
    IDestination destination = SessionUtil.GetDestination(session, "queue://TaskIssue_VoiceSM");
    //新建生产者对象
    IMessageProducer producer = session.CreateProducer(destination);
    producer.DeliveryMode = MsgDeliveryMode.NonPersistent;//ActiveMQ服务器停止工作后，消息不再保留
    ITextMessage request = session.CreateTextMessage();
    request.NMSCorrelationID = "TestVoiceSM";//这里我填了应用程序的名称。
    request.Properties["Callee"] = tbCallee.Text;
    request.Properties["Message"] = tbCheckNumber.Text;
    request.Properties["deadline"] = tbValidDuration.Text;
    producer.Send(request);
  }
  catch (Exception ex)
  {
    //初始化ActiveMQ连接失败，往VS2008的Output窗口写入出错信息!
    Debug.WriteLine(ex.Message);
  }
}
private void Window_Closed(object sender, EventArgs e)
{
  try
  {
    if (session == null)
      return;
    //if (connection == null)
    //  return;
    session.Close();
    //connection.Close();
  }
  catch (Exception ex)
  {
    Debug.WriteLine(ex.Message);
  }
}

更多关于C#相关内容感兴趣的读者可查看本站专题：《C#窗体操作技巧汇总》、《C#常见控件用法教程》、《WinForm控件用法总结》、《C#程序设计之线程使用技巧总结》、《C#操作Excel技巧总结》、《C#中XML文件操作技巧汇总》、《C#数据结构与算法教程》、《C#数组操作技巧总结》及《C#面向对象程序设计入门教程》

希望本文所述对大家C#程序设计有所帮助。

以上是小编为您精心准备的的内容，在的博客、问答、公众号、人物、课程等栏目也有的相关内容，欢迎继续使用右上角搜索按钮进行搜索c#
， mq
， 通讯
active
c站、c语言、cf、ch、c罗，以便于您获取更多的相关知识。

时间： 2024-03-05 07:52:06

C#实现同Active MQ通讯的方法_C#教程的相关文章

使用C#开发Socket通讯的方法_C#教程

下面的示例显示如何使用 Socket 类向 HTTP 服务器发送数据和接收响应.  [C#]  public string DoSocketGet(string server)  {  //Sets up variables and a string to write to the server  Encoding ASCII = Encoding.ASCII;  string Get = "GET / HTTP/1.1\r\nHost: " + server +  "\r\

C#在Unity游戏开发中进行多线程编程的方法_C#教程

在这之前,有很多人在质疑Unity支不支持多线程,事实上Unity是支持多线程的.而提到多线程就要提到Unity非常常用的协程,然而协程并非真正的多线程.协程其实是等某个操作完成之后再执行后面的代码,或者说是控制代码在特定的时机执行.而多线程在Unity渲染和复杂逻辑运算时可以高效的使用多核CPU,帮助程序可以更高效的运行.本篇主要介绍在Unity中如何使用多线程. 首先引入C#中使用多线程的类库 using System.Threading; 创建线程实例的四种方式 一.线程执行无参方法 构造

C#中调用VB中Inputbox类的实现方法_C#教程

C#自己没有Inputbox这个类,但是Inputbox也蛮好用的,所以有两种方法可以使用 一:间接调用vb中的Inputbox功能       1.在项目中添加对Microsoft.VisualBasic引用       2.在项目中添加命名空间Using Microsoft.VisualBasic;       3.以后就可以直接使用VB中的好多类库(爽啊--)        例如:textBox1.Text=Microsoft.VisualBasic.Interaction.InputBo

详解C#中通过委托来实现回调函数功能的方法_C#教程

委托(delegate)是一种可以把引用存储为函数的类型,这类似于c++中的函数指针. 回调函数c++中的回调函数,就是用函数指针来实现的.类似的,c#中用委托,来实现回调函数的功能. 回调函数为什么被称为回调函数?比如你调用了一个函数,那么就叫调用,但是如果你在调用一个函数的时候,还需要把一个函数提供给该函数,让这个函数来调用你的函数,那么你提供的这个函数就被称为回调函数(callback). 对于python这样的动态语言而言,就没有c#,c++提供特殊的语法实现回调函数,因为在python

C#实现在控制台输入密码显示星号的方法_C#教程

在控制台输入的内容C#默认按照字符串进行处理,如果直接让用户一次输入完毕就很难实现 显示星号的功能.但是如果让用户一次只能输入一个字符就,在将用户输入的字符替换为星号就可以实现了! 首先,C#中能让用户按下一个按键的方法就是Console.Readkey(),用户一次只能按下一个按键,其中它的另外一个重载方法是Console.ReadKey(bool b),bool类型的参数用来控制是否在控制台上显示用户按下的按键.那么我们可以传入true参数,使用户按下的按键不显示在控制台上,这样光标始终停留

C#利用原图和水印图的重叠简单实现水印的方法_C#教程

本文实例讲述了C#利用原图和水印图的重叠简单实现水印的方法.分享给大家供大家参考,具体如下: 图片操作类 /// <summary> /// 获取一个图片按等比例缩小后的大小. /// </summary> /// <param name="maxWidth">需要缩小到的宽度</param> /// <param name="maxHeight">需要缩小到的高度</param> /// &l

C#通过html调用WinForm的方法_C#教程

本文实例讲述了C#通过html调用WinForm的方法.分享给大家供大家参考,具体如下: 完整测试代码: Form1.cs: using System; using System.Collections.Generic; using System.ComponentModel; using System.Data; using System.Drawing; using System.Text; using System.Windows.Forms; namespace test { [Syst

C#实现SQL批量插入数据到表的方法_C#教程

本文实例讲述了C#实现SQL批量插入数据到表的方法.分享给大家供大家参考,具体如下: #region 帮助实例:SQL 批量插入数据 多种方法 /// <summary> /// SqlBulkCopy往数据库中批量插入数据 /// </summary> /// <param name="sourceDataTable">数据源表</param> /// <param name="targetTableName"

C#中判断一个集合是否是另一个集合的子集的简单方法_C#教程

看到这个标题,我们首先想到的是循环遍历其中一个数组,判断数组中的每个元素是否都在另一个数组中出现了,以此来判断该数组是不是另一个数组的子集,但是这样做就太过复杂了,有没有简单一点的方法呢? 打个比方,有这样的两个集合: 复制代码 代码如下: string[] bigArr = new string[] { "a", "b", "c" }; string[] smallArr = new string[] { "a", &quo