import org.apache.activemq.ActiveMQConnection;   
import org.apache.activemq.ActiveMQConnectionFactory;   
  
import javax.jms.*;   
  
/**  
* 消息的生产者（发送者）  
*  
*/  
public class JmsSender {   
        public static void main(String[] args) throws JMSException {   
                // ConnectionFactory ：连接工厂，JMS 用它创建连接   
                ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(   
                                ActiveMQConnection.DEFAULT_USER,   
                                ActiveMQConnection.DEFAULT_PASSWORD,   
                                "failover://(tcp://127.0.0.1:61616)");   
                
                //JMS 客户端到JMS Provider 的连接   
                Connection connection = connectionFactory.createConnection();   
                connection.start();   
                
                // Session： 一个发送或接收消息的线程   
                // createSession的第一个参数Boolean.FALSE指的是当前session不是一个事务
                //                即消息是自动提交，你不能调用session.commit函数提交事务。
                // createSession的第一个参数Boolean.TRUE指的是新创建的session是一个事务
                //                你必须调用session.commit函数来提交事务，才能真正把消息放到队列服务器中
                Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                
                // Destination ：消息的目的地;消息发送给谁.   
                // 获取session注意参数值my-queue是Query的名字   
                Destination destination = session.createQueue("my-queue");
                
                // MessageProducer：消息生产者   
                MessageProducer producer = session.createProducer(destination);
                
                //设置不持久化   
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                
                //发送一条消息   
                sendMsg(session, producer);  
                connection.close();   
        }   
  
        /**  
         * 在指定的会话上，通过指定的消息生产者发出一条消息  
         *  
         * @param session    消息会话  
         * @param producer 消息生产者  
         */  
        public static void sendMsg(Session session, MessageProducer producer) throws JMSException {   
                //创建一条文本消息   
                TextMessage message = session.createTextMessage("Hello ActiveMQ！");   
                
                //通过消息生产者发出消息   
                producer.send(message);   
                System.out.println("");   
        }   
}  

import org.apache.activemq.ActiveMQConnection;   
import org.apache.activemq.ActiveMQConnectionFactory;   
  
import javax.jms.*;   

public class JmsReceiver {   
        public static void main(String[] args) throws JMSException {   
                // ConnectionFactory ：连接工厂，JMS 用它创建连接   
                ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(   
                                ActiveMQConnection.DEFAULT_USER,   
                                ActiveMQConnection.DEFAULT_PASSWORD,   
                                "tcp://127.0.0.1:61616");   
                //JMS 客户端到JMS Provider 的连接   
                Connection connection = connectionFactory.createConnection();   
                connection.start();   
                
                // Session： 一个发送或接收消息的线程   
                // createSession的第一个参数Boolean.FALSE指的是当前session不是一个事务
                //                即消息是自动提交或接收，你不能调用session.commit函数提交事务。
                // createSession的第一个参数Boolean.TRUE指的是新创建的session是一个事务
                //                你必须调用session.commit函数来提交事务，否则接收过的消息还在
                //                队列服务器中，你再接收一次，发现他们还待着。
                Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                
                // Destination ：消息的目的地;消息从哪个队列接收.    
                Destination destination = session.createQueue("my-queue");   
                
                // 消费者，消息接收者   
                MessageConsumer consumer = session.createConsumer(destination);   
                while (true) {   
                   	    //1秒内没有收到消息，即返回
                        TextMessage message = (TextMessage) consumer.receive(1000);   
                        if (null != message)  
                        	System.out.println("收到消息：" + message.getText());
                        else  
                                break;   
                }   
                session.commit();
                session.close();   
                connection.close();   
        }   
}  

public class SampleUtilities {
    static public class DoneLatch {
        boolean  done = false;
        /**
         * Waits until done is set to true.
         */
        public void waitTillDone() {
            synchronized (this) {
                while (! done) {
                    try {
                        this.wait();
                    } catch (InterruptedException ie) {}
                }
            }
        }
        
        /**
         * Sets done to true.
         */
        public void allDone() {
            synchronized (this) {
                done = true;
                this.notify();
            }
        }
    }
}

import java.util.Date;

import javax.jms.*;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * @author Kim Haase
 * @comment&modified by Kagula
 * @version 1.6, 08/18/00
 * @lastUpdateDate  09/06/11
 */
public class AsynchTopicExample {
    final String  CONTROL_QUEUE = "controlQueue";
    String        topicName = null;
    int           exitResult = 0;

    public static ConnectionFactory getJmsConnectionFactory()
            throws JMSException {
        String user = ActiveMQConnection.DEFAULT_USER;
        String password = ActiveMQConnection.DEFAULT_PASSWORD;
        String url = ActiveMQConnection.DEFAULT_BROKER_URL;

        return new ActiveMQConnectionFactory(user, password, url);
    }
    
    public class AsynchSubscriber extends Thread {

    	 private class TextListener implements MessageListener {
             final SampleUtilities.DoneLatch  monitor = new SampleUtilities.DoneLatch();

             //若收到消息onMessage被调用(实现了MessageListener代理)
             public void onMessage(Message message) {
                 if (message instanceof TextMessage) {
                     TextMessage  msg = (TextMessage) message;

                     try {
                         System.out.println("订阅者线程:读取消息: " + msg.getText());
                     } catch (JMSException e) {
                         System.out.println("Exception in onMessage(): " + e.toString());
                     }
                 } else {
                 	//收到非TextMessage类型的消息，Publisher指示你可以结束订阅
                     monitor.allDone();
                 }
             }
         }//结束TextListener Class的定义

        /**
         * Runs the thread.
         */
        public void run() {
            ConnectionFactory       topicConnectionFactory = null;
            Connection              topicConnection = null;
            Session                 topicSession = null;
            Topic                   topic = null;
            MessageConsumer         topicSubscriber = null;
            TextListener            topicListener = null;

            try {
                topicConnectionFactory = AsynchTopicExample.getJmsConnectionFactory();
                topicConnection = topicConnectionFactory.createConnection();
                topicSession = topicConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                topic = topicSession.createTopic(topicName);
            } catch (Exception e) {
                System.out.println("Connection problem: " + e.toString());
                if (topicConnection != null) {
                    try {
                        topicConnection.close();
                    } catch (JMSException ee) {}
                }
    	        System.exit(1);
            } 

            try {
            	topicSubscriber = topicSession.createConsumer(topic);
                topicListener = new TextListener();
                topicSubscriber.setMessageListener(topicListener);
                topicConnection.start();                
             
                /*
                 * Asynchronously process messages.
                 * Block until publisher issues a control message indicating
                 * end of publish stream.
                 */
                topicListener.monitor.waitTillDone();
            } catch (JMSException e) {
                System.out.println("Exception occurred: " + e.toString());
                exitResult = 1;
            } finally {
                if (topicConnection != null) {
                    try {
                        topicConnection.close();
                    } catch (JMSException e) {
                        exitResult = 1;
                    }
                }
            }
        }	    
    }

    /**
     * The MultiplePublisher class publishes several message to a topic. 
     *
     * @author Kim Haase
     * @version 1.6, 08/18/00
     */
    public class MultiplePublisher extends Thread {

        /**
         * Runs the thread.
         */
        public void run() {
            ConnectionFactory  topicConnectionFactory = null;
            Connection         topicConnection = null;
            Session            topicSession = null;
            Topic              topic = null;
            MessageProducer    topicPublisher = null;
            TextMessage        message = null;
            final int          NUMMSGS = 20;
            final String       MSG_TEXT = new String("Here is a message");

            try {
                topicConnectionFactory = AsynchTopicExample.getJmsConnectionFactory();
                topicConnection = topicConnectionFactory.createConnection();
                topicSession = topicConnection.createSession(false,Session.AUTO_ACKNOWLEDGE);
                topic = topicSession.createTopic(topicName);
            } catch (Exception e) {
                System.out.println("Connection problem: " + e.toString());
                if (topicConnection != null) {
                    try {
                        topicConnection.close();
                    } catch (JMSException ee) {}
                }
    	        System.exit(1);
            } 

            try {                
                topicPublisher = topicSession.createProducer(topic);
                message = topicSession.createTextMessage();
                for (int i = 0; i 
<p><br>
 </p>
<h2 id="参考资料">参考资料</h2>
<p>[1]《Active MQ入门(转)  》</p>
<p> http://yeweiyun868.blog.163.com/blog/static/563784432010112301916556/</p>
<p>[2]《Eclipse中添加Src和JavaDoc的方法》</p>
<p>http://hi.baidu.com/rebeccacao/blog/item/d3f67ed3af8384229b5027bf.html</p>
<p>[3]Active MQ 官网</p>
<p>http://activemq.apache.org/</p>
<p>[4]《Java Message Service Specification》含JMS1.1标准示例代码</p>
<p>http://www.oracle.com/technetwork/java/docs-136352.html</p>
<p>[5]《JMS Example: Publish and Subscribe》</p>
<p>http://jmsexample.zcage.com/index2.html</p>
<p><br>
</p>
<p><br>
</p>
<p><br>
</p>
<p><br>
</p>
<p><br>
</p>
<p><br>
</p>
<p>=========================================================</p>

<h3 id="span-Active-MQ-C-实现-span"><span>Active MQ C#实现</span></h3>
<h1 id="Active-MQ-C-实现">Active MQ C#实现</h1>

<p>Kagula</p>
<p>2011/9/24</p>
<h2 id="内容概要">内容概要</h2>
<p>主要以源码的形式介绍如何用C#实现同Active MQ 的通讯。本文假设你已经正确安装JDK1.6.x，了解Active MQ并有一定的编程基础。</p>
<h2 id="正文">正文</h2>
<p>    JMS 程序的最终目的是生产和消费的消息能被其他程序使用，JMS 的 Message 是一个既简单又不乏灵活性的基本格式，允许创建不同平台上符合非JMS 程序格式的消息。</p>
<p>Message 由消息头，属性和消息体三部份组成。</p>
<p>    Active MQ支持过滤机制，即生产者可以设置消息的属性（Properties），该属性与消费者端的Selector对应，只有消费者设置的selector与消息的Properties匹配，消息才会发给该消费者。Topic和Queue都支持Selector。</p>
<p><strong>示例代码</strong><br>
</p>
<pre class="brush:php;toolbar:false">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>
 * 修改为
 * <property name="authenticate" value="false"></property>
 * 
 * 第二步：修改登录用户名密码，缺省分别为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"></authenticationuser>
 * 
 * 在[你的解压路径]\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.cnblogs.com/guthing/archive/2010/06/17/1759333.html
 * [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);
            }));
        }
    }
}

// testActiveMQ.cpp : Defines the entry point for the console application.
//

#include "stdafx.h"

#include <activemq>
#include <decaf>
#include <decaf>
#include <decaf>
#include <decaf>
#include <decaf>
#include <decaf>
#include <activemq>
#include <activemq>
#include <cms>
#include <cms>
#include <cms>
#include <cms>
#include <cms>
#include <cms>
#include <cms>

#include <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <memory>

using namespace activemq::core;
using namespace decaf::util::concurrent;
using namespace decaf::util;
using namespace decaf::lang;
using namespace cms;
using namespace std;


int _tmain(int argc, _TCHAR* argv[])
{
	activemq::library::ActiveMQCPP::initializeLibrary();

	Connection*      connection;
    Session*         session;
    Destination*     destination;
    MessageProducer* producer;


    std::string brokerURI("failover:(tcp://localhost:61616)");

	try {
		// Create a ConnectionFactory
		auto_ptr<connectionfactory> connectionFactory( ConnectionFactory::createCMSConnectionFactory( brokerURI ) );

        // Create a Connection
        connection = connectionFactory->createConnection();
        connection->start();

        // Create a Session
		session = connection->createSession( Session::AUTO_ACKNOWLEDGE );

        // Create the destination (Topic or Queue)
		destination = session->createQueue( "FromMyCPlusPlus" );

        // Create a MessageProducer from the Session to the Topic or Queue
        producer = session->createProducer( destination );
        producer->setDeliveryMode( DeliveryMode::NON_PERSISTENT );

        // Create a messages
        string text("Hello world from c++");

        for( int ix=0; ixcreateTextMessage( text );                
			message->setIntProperty( "Integer", ix );
            producer->send( message );
            delete message;
        }
        }catch ( CMSException& e ) {
            e.printStackTrace();
        }

	//释放资源
    try{
        if( destination != NULL ) delete destination;
    }catch ( CMSException& e ) { e.printStackTrace(); }
    destination = NULL;

    try{
        if( producer != NULL ) delete producer;
    }catch ( CMSException& e ) { e.printStackTrace(); }
    producer = NULL;

    // Close open resources.
    try{
        if( session != NULL ) session->close();
        if( connection != NULL ) connection->close();
    }catch ( CMSException& e ) { e.printStackTrace(); }

    try{
        if( session != NULL ) delete session;
    }catch ( CMSException& e ) { e.printStackTrace(); }
    session = NULL;

    try{
        if( connection != NULL ) delete connection;
    }catch ( CMSException& e ) { e.printStackTrace(); }
    connection = NULL;
	
	activemq::library::ActiveMQCPP::shutdownLibrary();

	return 0;
}</connectionfactory></memory></iostream></stdio.h></stdlib.h></cms></cms></cms></cms></cms></cms></cms></activemq></activemq></decaf></decaf></decaf></decaf></decaf></decaf></activemq>