1ActiveMQ使用 2ActiveMQC#Client实现 3ActiveMQC++Client实现通讯 4 ActiveMQJavaClient实现

Active MQ使用

http://www.huaishao8.com/tag/activemq

Kagula

2011-9-6

介绍

Active MQ是个消息队列管理器，用于通讯的中间件。

Java + Active MQ的常见使用方式有两种：[1]点对点方式(Producer/Consumer)[2]发布/订阅者方式（Publisher/Subscriber Model）

测试环境[1]JDK1.6.x [2]Eclipse Indigo [3]Active MQ 5.4.2

建议不要使用Active MQ 5.5.0 因为activemq-all-5.5.0.jar缺少依赖项。

正文

参考资料[1] 保证Active MQ已经正确安装与启动。

点对点(Producer、Consumer)模式

图一 JMS对象模型
Producer端源码示例

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("");
}
}

Consumer端源码示例

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();
}
}

Publisher/Subscriber Model

消息传播以一对多的关系传递，即一个Publisher，多个Subscriber。

你也可以为下面的源码示例加入下面的代码

topicPublisher.setDeliveryMode(DeliveryMode.PERSISTENT);

它的作用是：ActiveMQ服务器重启后，仍保留你发上去的消息。

你也可以通过下面的代码段，设置发上去的消息在MQ服务器上的存活时间

topicPublisher.setTimeToLive(1000*60*60*24*3);//ms为单位

源码示例

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 < NUMMSGS; i++) {
message.setText(MSG_TEXT + " " + (i + 1));
System.out.println("发布者: 发布消息: " + message.getText());
topicPublisher.send(message);
}

// Send a non-text control message indicating end of messages.
topicPublisher.send(topicSession.createMessage());
} catch (JMSException e) {
System.out.println("Exception occurred: " + e.toString());
exitResult = 1;
} finally {
if (topicConnection != null) {
try {
topicConnection.close();
} catch (JMSException e) {
exitResult = 1;
}
}
}
}
}

public void run_threads() {
AsynchSubscriber   asynchSubscriber = new AsynchSubscriber();
MultiplePublisher  multiplePublisher = new MultiplePublisher();

multiplePublisher.start();
asynchSubscriber.start();
try {
asynchSubscriber.join();
multiplePublisher.join();
} catch (InterruptedException e) {}
}

public static void main(String[] args) {
AsynchTopicExample  ate = new AsynchTopicExample();

ate.topicName = "MyFirstTopic";
System.out.println("Topic name is " + ate.topicName);

ate.run_threads();
}
}

参考资料

[1]《Active MQ入门(转) 》

http://yeweiyun868.blog.163.com/blog/static/563784432010112301916556/

[2]《Eclipse中添加Src和JavaDoc的方法》

http://hi.baidu.com/rebeccacao/blog/item/d3f67ed3af8384229b5027bf.html

[3]Active MQ 官网

http://activemq.apache.org/

[4]《Java Message Service Specification》含JMS1.1标准示例代码

http://www.oracle.com/technetwork/java/docs-136352.html

[5]《JMS Example: Publish and Subscribe》

http://jmsexample.zcage.com/index2.html

=========================================================

Active MQ C#实现

Active MQ C#实现

Kagula

2011/9/24

内容概要

主要以源码的形式介绍如何用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.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);
}));
}
}
}

===================================================================

Active MQ C++实现通讯

Kagula

2011-9-13

简介

在参考资料[2]的基础上介绍如何用C++调用Active MQ的客户端API。

环境：[1]Windows XP SP3

[2]Visual Studio 2008 SP1

阅读前提：[1]熟悉Microsoft Visual Studio下的C++编程

[2]熟悉《Active MQ使用》这篇文章中的内容，即参考资料[2]

正文

CMS (stands for C++ Messaging Service)类似于JMS API用于同Message Brokers通讯（例如Active MQ）。

Active MQ-CPP是客户端库。通过参考资料[3]下载“Active MQ-CPP v3.4.0 (activemq-cpp-library-3.4.0-src.zip)”

准备开发环境

APR(Apache portable Run-time libraries，Apache可移植运行库)的目的如其名称一样，主要为上层的应用程序提供一个可以跨越多操作系统平台使用的底层支持接口库。

你必须从参考[4]中下载apr-iconv包、apr-util包、apr包等三个包，并解压缩到

C:\work\apr\

C:\work\apr-iconv\

C:\work\apr-util\

等以上三个目录（必须是在上面的路径中，否则VS2008中会找不到工程依赖文件和依赖函数实现）

第一步：apr-iconv包的配置

下载http://apache.osuosl.org/apr/apr-iconv-1.2.1-win32-src-r2.zip

解压后打开包内的apriconv.dsp文件编译出Release版本的apriconv-1.lib文件。

配置VS2008的头文件搜索路径为

C:\work\apr-iconv\include

配置VS2008的库文件搜索路径为

C:\work\apr-iconv\\LibR

第二步：apr-util包的配置

参考资料[4]下载apr-util-1.3.12-win32-src.zip 包中的源码，编译aprutil工作空间中的aprutil项目，生成aprutil-1.lib、libaprutil-1.lib。

把“c:\work\apr-util\include”目录配置到VS2008头文件搜索路径里

把“c:\work\apr-util\LibR”目录配置到VS2008库文件搜索路径里

把“c:\work\apr-util\release”目录配置到VS2008库文件搜索路径里

第三步：apr基本包的配置

参考资料[4]下载、编译apr-1.4.5-win32-src.zip 包中的源码。

编译好后在LibR目录里生成apr-1.lib库、libapr-1.lib。否则activemq-cpp的acitvemq-cpp-example项目编译时会找不到库文件。

把“c:\work\apr\include”目录配置到VS2008头文件搜索路径里

把“c:\work\apr\LibR”目录配置到VS2008库文件搜索路径里

把“C:\work\apr\Release”目录配置到VS2008库文件搜索路径里

第四步：CppUnit包的配置

在下面网址下载

http://sourceforge.net/projects/cppunit/files/cppunit/1.12.1/cppunit-1.12.1.tar.gz/download

解压缩cppunit-1.12.1.tar.gz包，并以Release方式编译

把“D:\cppunit-1.12.1\include” 目录配置到VS2008头文件搜索路径里

把“D:\cppunit-1.12.1\lib”目录配置到VS2008库文件搜索路径里

其中上面的D:\是我的解压路径，你也可以是其它路径。

第五步：activemq-cpp-library-3.4.0包的配置

解包activemq-cpp-library-3.3.0-src.zip文件(从参考资料[3]中下载)，并以Release方式编译，如果前面几步正确，这一步不会产生任何错误。

这个zip包里，含activemq-cpp-example项目，里面有个C++源码示例，我这里的源码片段就是参考那里的。

把“D:\activemq-cpp-library-3.4.0\src\main” 目录配置到VS2008头文件搜索路径里

把“D:\activemq-cpp-library-3.4.0\src\main\decaf\internal” 目录配置到VS2008头文件搜索路径里

把“D:\activemq-cpp-library-3.4.0\vs2008-build\Win32\Release”目录配置到VS2008库文件搜索路径里

其中上面的D:\是我的解压路径，你也可以是其它路径。

使用“activemq-cpp”提供的API只要配置其头文件存放路径就可以了。

C++样例

要运行样例文件，请确保以releaseDLL和debugDLL方式编译成功以上各个依赖包。以Debug方式调试你的应用程序时必须使用activemq-cppd.lib库文件及其对应的activemq-cppd.dll动态链接库文件，否则创建connectionFactory对象会出错。

依赖库列表（含路径，四个动态链接库，三个静态库，不包括ws2_32.lib）

D:\activemq-cpp-library-3.4.0\vs2008-build\Win32\DebugDLL\activemq-cppd.lib

D:\activemq-cpp-library-3.4.0\vs2008-build\Win32\DebugDLL\activemq-cppd.dll

C:\work\apr\Release\libapr-1.lib

C:\work\apr\Release\libapr-1.dll

C:\work\apr-util\Release \libaprutil-1.dll

C:\work\apr-util\Release \libaprutil-1.lib

C:\work\apr-iconv\Release\ libapriconv-1.dll

新建Win32 Console项目在VS2008里，修改[Configuration Properties]->[Linker]->[Input]->[Additional Dependencies]属性为

ws2_32.lib activemq-cppd.lib libapr-1.lib libaprutil-1.lib

把需要的动态链接库文件复制到当前项目的运行目录中。

以Debug方式运行成功后，http://127.0.0.1:8161/admin/queues.jsp

你会发现多个了“FromMyCPlusPlus”队列，里面已经存放了刚才发送的消息。

下面是我的完整源码testActiveMQ.cpp文件

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

#include "stdafx.h"

#include <activemq/library/ActiveMQCPP.h>
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
#include <decaf/util/concurrent/CountDownLatch.h>
#include <decaf/lang/Integer.h>
#include <decaf/lang/Long.h>
#include <decaf/lang/System.h>
#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/util/Config.h>
#include <cms/Connection.h>
#include <cms/Session.h>
#include <cms/TextMessage.h>
#include <cms/BytesMessage.h>
#include <cms/MapMessage.h>
#include <cms/ExceptionListener.h>
#include <cms/MessageListener.h>

#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; ix<3; ++ix ){
TextMessage* message = session->createTextMessage( 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;
}

参考资料

[1]《解决Active MQ中，Java与C++交互中文乱码问题》

http://www.blogjava.net/javagrass/archive/2011/05/06/349660.html

[2]《Active MQ使用》

/article/2293714.html

[3]《Active MQ CMS》

http://activemq.apache.org/cms/

[4]Apache Portable Runtime Project

http://apr.apache.org/

编译activemq-cpp-example碰到的问题

kagula

2012-3-2

环境

[1]Win7SP1

[2]apr-1.4.6-win32-src.zip

[3]apr-util-1.4.1-win32-src.zip

[4]apr-iconv-1.2.1-win32-src-r2.zip

[5]activemq-cpp-library-3.4.1-src.zip

[6]apache-activemq-5.4.3-bin.zip

[7]VS2010SP1

打开apache-activemq包里自带的activemq-cpp-example的项目虽然编译成功

但是一运行提示找不到libparutil-1.dll。

进入c:\work\aprutil编译aprutil，静态库能生成，但是动态库无法生成。

察看APR官方文档，我们知道如果在LINK的依赖项里设置LIBXXX.lib，则程序运

行的时候，会到当前运行目录里寻找相应的XXX.dll文件。

动态不成，我们用静态，把LINK依赖项的libapr-1.lib、libaprutil-1.lib

分别改为apr-1.lib和aprutil-1.lib，但是编译会有很多符号（函数）找不到实现。

我们替这些符号找到实现所在的LIB文件，然后，补上去。

完成后的，"[Linker]->[Input]->[Additional Dependencies]"项内容如下

“ws2_32.lib apr-1.lib aprutil-1.lib apriconv-1.lib Mswsock.lib

Rpcrt4.lib”共六个库文件。

现在activemq-cpp-example程序编译成功。一运行停留在命令行下，这时启动

ActiveMQ，程序会正常往下运行并结束。问题解决。

但是如果你在C++端发送中文信息给Java端，Java端会报UTF-8解析出错，这时

你只需要传送中文前把GBK转成UTF-8再发给Java端就可以了。

Active MQ JavaClient实现

package com.test.mq.activemq;

import javax.jms.Connection;

import javax.jms.DeliveryMode;

import javax.jms.Destination;

import javax.jms.ExceptionListener;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**

* Hello world!

*/

public class App {

public static void main(String[] args) throws Exception {

thread(new HelloWorldProducer(), false);

thread(new HelloWorldProducer(), false);

thread(new HelloWorldConsumer(), false);

Thread.sleep(1000);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldProducer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldProducer(), false);

Thread.sleep(1000);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldProducer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldProducer(), false);

thread(new HelloWorldProducer(), false);

Thread.sleep(1000);

thread(new HelloWorldProducer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldProducer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldProducer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldProducer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldConsumer(), false);

thread(new HelloWorldProducer(), false);

}

public static void thread(Runnable runnable, boolean daemon) {

Thread brokerThread = new Thread(runnable);

brokerThread.setDaemon(daemon);

brokerThread.start();

}

public static class HelloWorldProducer implements Runnable {

public void run() {

try {

// Create a ConnectionFactory

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");

// Create a Connection

Connection connection = connectionFactory.createConnection();

connection.start();

// Create a Session

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

// Create the destination (Topic or Queue)

Destination destination = session.createQueue("TEST.FOO");

// Create a MessageProducer from the Session to the Topic or Queue

MessageProducer producer = session.createProducer(destination);

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

// Create a messages

String text = "Hello world! From: " + Thread.currentThread().getName() + " : " + this.hashCode();

TextMessage message = session.createTextMessage(text);

// Tell the producer to send the message

System.out.println("Sent message: "+ message.hashCode() + " : " + Thread.currentThread().getName());

producer.send(message);

// Clean up

//producer.close();

session.close();

connection.close();

}

catch (Exception e) {

System.out.println("Caught: " + e);

e.printStackTrace();

}

}

}

public static class HelloWorldConsumer implements Runnable, ExceptionListener {

public void run() {

try {

// Create a ConnectionFactory

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");

// Create a Connection

Connection connection = connectionFactory.createConnection();

connection.start();

connection.setExceptionListener(this);

// Create a Session

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

// Create the destination (Topic or Queue)

Destination destination = session.createQueue("TEST.FOO");

// Create a MessageConsumer from the Session to the Topic or Queue

MessageConsumer consumer = session.createConsumer(destination);

// Wait for a message

Message message = consumer.receive(1000);

if (message instanceof TextMessage) {

TextMessage textMessage = (TextMessage) message;

String text = textMessage.getText();

System.out.println("Received: " + text);

} else {

System.out.println("Received: " + message);

}

consumer.close();

session.close();

connection.close();

} catch (Exception e) {

System.out.println("Caught: " + e);

e.printStackTrace();

}

}

public synchronized void onException(JMSException ex) {

System.out.println("JMS Exception occured. Shutting down client.");

}

}

}

FAQ常见问题：

1 activemq重新连接机制：

生产者或消费者配置实例：

private String url = "failover:(tcp://192.168.4.170:61616?wireFormat.maxInactivityDuration=1000)";

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url)

failover机制原理：

ActiveMQ uses a keepalive protocol on top of its transports, to keep firewalls open and also detect whether the broker is no longer reachable.

The keepalive protocol periodically sends a lightweight command message to the broker, and expects a response. If it dosen't receive one within a given time period,

AcitveMQ will assume that the transport is no longer walid. The failover transport listens for failed transports and will select another transport to use on such a failure.

参考:

[1]《ActiveMQ-In-Action-Manning》 12.5 Surviving network or broker failure with the failover protocol

[2] activemq重新连接机制 参考：http://jinguo.iteye.com/blog/243514