Spring整合JMS、IBM MQ发送和接收消息_java连接ibmmq9

最近才接触到MQ，由于之前完全不知道是干嘛用的，还是很花了一点时间研究的~先来简单解释一下名词啦

一、名词解释

1. MQ

   MQ（message queue）指消息队列，是应用程序对应用程序的通信方法。可以利用消息队列暂存数据报文。
   MQ的原理其实就是生产者-消费者模式。有关生产者-消费者模式的详细解释可以看这篇博文http://blog.csdn.net/yolanda_nuonuo/article/details/62038122
   MQ是一种异步通讯。
   生产者，发送消息到MQ服务器
   消费者，到MQ服务器，获取生产者发送的消息(根据MQ的不同配置，可以是MQ推送消息给消费者，或者是消费者主动从MQ获取消息)

2. JMS

   JMS即
   JAVA消息服务（Java Message Service），是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

使用JMS 的应用程序被称为JMS 客户端，处理消息路由与传递的消息系统被称为JMS Provider，而JMS 应用则是由多个JMS 客户端和一个JMS Provider 构成的业务系统。
发送消息的JMS 客户端被称为生产者（producer），而接收消息的JMS 客户端则被称为消费者(consumer)。同一JMS 客户端既可以是生产者也可以是消费者。

所以MQ跟JMS的关系就是，我们可以通过JMS来跟MQ进行通信。

二、具体实现

1.首先是发送数据，新建application-send.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:jpa="http://www.springframework.org/schema/data/jpa"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:cache="http://www.springframework.org/schema/cache"
    xmlns:p="http://www.springframework.org/schema/p"
    xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.1.xsd
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
        http://www.springframework.org/schema/data/jpa http://www.springframework.org/schema/data/jpa/spring-jpa-1.3.xsd
        http://www.springframework.org/schema/cache http://www.springframework.org/schema/cache/spring-cache-3.1.xsd
        http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">

    <context:property-placeholder ignore-unresolvable="false" location="classpath:application.properties" />

    <context:annotation-config />
    <!-- <context:component-scan base-package="com.*" /> -->

    <bean id="ibmJmsConnectionFactory" class="com.ibm.mq.jms.MQQueueConnectionFactory">
        <property name="hostName" value="${ibm.mqHostUrl}" />
        <property name="port" value="${ibm.mqPort}" />
        <property name="CCSID" value="0000" />
        <property name="queueManager" value="${ibm.mqManager}" />
        <property name="channel" value="${ibm.mqChannel}" />
        <property name="transportType" value="1" />
    </bean>
    <bean id="ibmQueueConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <property name="targetConnectionFactory" ref="ibmJmsConnectionFactory"/>
        <property name="sessionCacheSize" value="10"/>
    </bean>

    <bean id="ibmQueue" class="com.ibm.mq.jms.MQQueue">
        <property name="baseQueueName" value="${ibm.mqName}" />
    </bean>

    <bean id="ibmJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="ibmQueueConnectionFactory" />
        <property name="defaultDestination" ref="ibmQueue" />
        <property name="pubSubDomain" value="false" />
    </bean>  

    <bean id="ibmMessageSender" class="com.yolanda.sender.MessageSenderImpl">
    </bean>

</beans>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

2.发送数据的代码
先写一个接口

public interface MessageSender {

    public void sendMessage(String message);

}
1
2
3
4
5
6

再写实现类

import java.io.UnsupportedEncodingException;
import javax.annotation.Resource;
import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;

@Component("ibmMessageSender")
public class MessageSenderImpl implements MessageSender {

    @Resource(name="ibmJmsTemplate")
    private JmsTemplate jmsTemplate;

    /**
     * 这里为什么要用final呢？是下面那个createTextMessage方法要求的...
     */
    public void sendMessage(final String message) {
        this.jmsTemplate.send(new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                    return session.createTextMessage(message);
            }
        });
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

3.接下来就是收数据辣，新建application-recv.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:jpa="http://www.springframework.org/schema/data/jpa"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:cache="http://www.springframework.org/schema/cache"
    xmlns:p="http://www.springframework.org/schema/p"
    xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.1.xsd
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
        http://www.springframework.org/schema/data/jpa http://www.springframework.org/schema/data/jpa/spring-jpa-1.3.xsd
        http://www.springframework.org/schema/cache http://www.springframework.org/schema/cache/spring-cache-3.1.xsd
        http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">

    <bean id="ibmJmsConnectionFactory" class="com.ibm.mq.jms.MQQueueConnectionFactory">
        <property name="hostName" value="${ibm.mqHostUrl}" />
        <property name="port" value="${ibm.mqPort}" />
        <property name="CCSID" value="1381" />
        <property name="queueManager" value="${ibm.mqManager}" />
        <property name="channel" value="${ibm.mqChannel}" />
        <property name="transportType" value="1" />
    </bean>
    <bean id="ibmQueueConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <property name="targetConnectionFactory" ref="ibmJmsConnectionFactory"/>
        <property name="sessionCacheSize" value="10"/>
    </bean>

    <bean id="ibmQueue" class="com.ibm.mq.jms.MQQueue">
        <property name="baseQueueName" value="MQTEST" />
    </bean>

    <bean id="queueContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="ibmQueueConnectionFactory" />
        <property name="destination" ref="ibmQueue" />
        <property name="pubSubDomain" value="false" />
        <property name="concurrentConsumers" value="20" />
        <property name="messageListener" ref="ibmTextMessageListener" />
    </bean>


    <!-- 异步接收消息处理类 -->
    <bean id="ibmextMessageListener" class="com.yolanda.receive.TextMessageListener" />
</beans>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

4.由于整合了Spring，又是用JMS，所以代码还是灰常简单的，这里我们只需要实现MessageListener的onMessage（），监听一下就可以啦

package com.yolanda.receiver;

import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.springframework.beans.factory.annotation.Autowired;

public class TextMessageListener implements MessageListener {
    @Autowired
    public void onMessage(Message message) {
        try {
            TextMessage msg = (TextMessage) message;
            String msgStr = msg.getText();
            System.out.println("Receive message:" + msgStr);
        } catch (Exception e) {
            e
.printStackTrace();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

好啦，圆满结束啦，不用spring整合的话，代码还是比较繁琐的，果然还是用比较好呢！