Java利用MsmqJava操作MSMQ消息队列

http://www.hxstrive.com/article/512.htm

 

该篇文章将向读者展示，怎样使用java去操作MSMQ（Windows自带的一个消息队列）。分为发送和接收消息，在操作MSMQ时，我们需要下载MsmqJava文件，该文件包含一个jar包和一个dll文件。下面将逐步教您怎样编写一个MSMQ程序。

 

第一步：开启MSMQ

Windows默认是没有开启MSMQ功能的，需要我们自己手动去开启，开启方法如下：

注意：点击“启动或关闭Windows功能”，然后在弹出的选择框中找到MSMQ即可（这就不详细说了，相信您能办到）。

 

第二步：创建一个测试队列

右击“我的电脑”点击管理。在管理中找到“服务和应用程序”项，点击“消息队列”。然后右击“专用队列”，选择“新建”/“专用队列”。

 

第三步：设置新建专用队列权限

在“专用队列”中，找到刚才创建的新队列，右击选择“属性”。在属性框中，选择“安全”页，设置“Everyone”权限为完全控制。

 

第四步：下载MsmqJava

打开浏览器，输入“http://msmqjava.codeplex.com/”，然后下载MsmqJava包，然后解压。

 

Java利用MsmqJava操作MSMQ消息队列

该项目使用Eclipse进行开发，jdk1.6，同时引入了日志框架（Slf4j、log4j）、MsmqJava.dll、MsmqJava.jar。

 

基础代码（BaseMessage.java）

package com.huangx;

 

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

 

/**

 * 基础类

 *

 * @author Administrator

 * @date 2017-06-23 22:01:27

 */

public abstract class BaseMessage {

    private static final Logger LOG = LoggerFactory.getLogger(BaseMessage.class);

 

    /**

     * 队列名称。其中：192.32.12.76为本机的IP地址（经测试不能使用127.0.0.1，不然会报错）

     * private$\\javaTest是队列名字

     */

    protected static final String QUEUE_NAME = "direct=tcp:192.168.1.115\\private$\\javaTest";

 

    static {

        try {

            LOG.debug("开始加载DLL文件...");

            System.loadLibrary("MsmqJava");

            LOG.debug("加载DLL文件成功");

        } catch (Exception e) {

            LOG.error("加载DLL失败，原因：" + e.getMessage(), e);

        }

    }

 

    /**

     * 执行业务逻辑，由具体业务实现

     */

    public abstract void execute();

 

}

 

 

log4j.properties

log4j.rootLogger=debug,Console

log4j.logger.org.apache.zookeeper=warn,Console

 

log4j.appender.Console=org.apache.log4j.ConsoleAppender

log4j.appender.Console.layout=org.apache.log4j.PatternLayout

log4j.appender.Console.layout.ConversionPattern=%d %-5p [%c] - %m%n

 

 

Java发送代码（SendMessage.java）

package com.huangx;

 

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

 

import ionic.Msmq.Message;

import ionic.Msmq.Queue;

 

/**

 * 发送消息到MSMQ队列

 *

 * @author Administrator

 * @date 2017-06-23 21:38:11

 */

public class SendMessage extends BaseMessage {

    private static final Logger LOG = LoggerFactory.getLogger(SendMessage.class);

 

    @Override

    public void execute() {

        try {

            LOG.debug("开始发送消息...");

            // 创建队列

            Queue queue = new Queue(QUEUE_NAME);

 

            // 创建消息

            String label = "testMessage";

            String body = "Hello, World!";

            byte[] correlationId = { 0, 2, 4, 6, 8, 9 };

            Message msg = new Message(body, label, correlationId);

 

            // 发送消息

            LOG.debug("发送消息[label={}, body={}]", msg.getLabel(), msg.getBodyAsString());

            queue.send(msg);

 

            LOG.debug("发送消息成功^_^");

        } catch (Exception e) {

            LOG.error("发送消息失败！原因：" + e.getMessage(), e);

        }

    }

 

    public static void main(String[] args) {

        new SendMessage().execute();

    }

 

}

 

 

编写Java接收代码（ReceiveMessage.java）

package com.huangx;

 

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

 

import ionic.Msmq.Message;

import ionic.Msmq.Queue;

 

/**

 * 接收消息

 *

 * @author Administrator

 * @date 2017-06-23 22:19:49

 */

public class ReceiveMessage extends BaseMessage {

    private static final Logger LOG = LoggerFactory.getLogger(ReceiveMessage.class);

 

    @Override

    public void execute() {

        try {

            // 创建队列

            Queue queue = new Queue(QUEUE_NAME);

 

            // 接收消息

            Message message = queue.receive();

            LOG.debug("接收到消息[label={}, body={}]",

                    message.getLabel(), message.getBodyAsString());

 

        } catch (Exception e) {

            LOG.debug(e.getMessage(), e);

        }

    }

 

    public static void main(String[] args) {

        new ReceiveMessage().execute();

    }

 

}