Java利用MsmqJava操作MSMQ消息队列

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

第一步：开启MSMQ

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

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

java广告位

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

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

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

在"专用队列"中，找到刚才创建的新队列，右击选择"属性"，如下图：

在属性框中，选择"安全"页，设置"Everyone"权限为完全控制，如下图：

第四步：下载MsmqJava

打开浏览器，输入"https://msmqjava.codeplex.com/"，然后下载MsmqJava包，然后解压。内容如下图：

第五步：搭建开发环境

下图为项目整体结构图：

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

java广告位

基础代码（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();
    }
}