spring boot整合IBM WebSphere MQ，并配置多个队列管理器

IBM WebSphere MQ概述

网上关于IBM WebSphere MQ的资料挺少的，毕竟是一项老技术，整理一下从零开始对于IBM WebSphere MQ的理解

IBM WebSphere MQ是一种消息中间件技术，可用于多个系统间通信，对于消息中间件的作用，这里就不多做展开。首先要了解IBM WebSphere MQ中的几个名词定义

队列管理器：构件独立的MQ运行环境，类似于RabbitMQ中的vhost，主要作用是维护和管理消息队列
队列：存放消息的容器，可分为本地队列，远程队列等
消息：MQ中的最小对象，默认情况下，消息缺省可以达到 4MB。消息可以分成持久消息和非持久消息。所谓“持久”的意思,就是在MQ 队列管理器重启动后,消息是否仍然能保持。持久的消息写入或读出队列的同时会在 Log 中记录,所以性能上比非持久消息差不少。
通道：通道是两个队列管理器之间的一种单向的点对点的通信连接。队列管理器之间的通信是通过配置通道来实现的,通道两侧的队列管理器对这个通道的相关参数应该能对应起来。在通道上可以配置不同的通信协议,这样就使得编程接口与通信协议无关。通道两端的配置必须匹配, 且名字相同,否则无法连通。消息在通道中只能单向的流动。如果需要双向的流动，可以创建一对通道，一来一去。

其基本服务架构如图
图中显示了 IBM WebSphere MQ 编程的原理。第一步是让应用程序与队列管理器连接。它通过 MQConnect 调用来进行此连接。下一步使用 MQOpen 调用为输出打开一个队列。然后应用程序使用 MQPut 调用将其数据放到队列上。要接收数据，应用程序调用 MQOpen 调用打开输入队列。应用程序使用 MQGet 调用从队列上接收数据。
还展示了消息通道代理（MCA）、通道出口和对象权限管理器（OAM）。MCA 是 IBM WebSphere MQ 程序，它使用现有传输服务诸如 TCP/IP 与 SNA 将消息从本地传输队列移到目标队列管理器。这些传输服务即通道。通道出口是用户写入库，可以在通道运作期间，从已定义位置号之一进入这些库。OAM 是命令和对象管理的缺省授权服务（针对操作系统）。这三个组件对 IBM WebSphere MQ 的现有安全性解决方案非常重要。

IBM WebSphere MQ服务的安装

服务端的安装这里就不做描述，请查阅其他博文。
服务安装后为了使客户端能连接使用，需做如下准备：

创建队列管理器
新建本地队列
新建服务连接通道，并将其属性-MCA用户标识值设置为“MUSR_MQADMIN”

注意：此处博主并未设置客户端的登录账号与密码，因为未找到在哪里设置，请知晓的朋友告知一下。

springboot配置多个队列管理器连接

1.引入依赖

<!--ibm mq--><dependency><groupId>org.springframework</groupId><artifactId>spring-jms</artifactId></dependency><dependency><groupId>javax.jms</groupId><artifactId>javax.jms-api</artifactId><version>2.0.1</version></dependency><dependency><groupId>com.ibm.mq</groupId><artifactId>com.ibm.mq.allclient</artifactId><version>9.1.1.0</version></dependency>

yml配置文件信息

spring:ibmmq:zsam:host: 10.1.16.156port: 1415 #此处配置的是队列管理器监听端口queue-manager: qm_ntfm_zsamchannel: ch_serviceccsid: 1381 #CCSID要与连接到的队列管理器一致，Windows下默认为1381，Linux下默认为1208。1208表示UTF-8字符集，建议把队列管理器的CCSID改为1208username: MUSR_MQADMINpassword:receive-timeout: 2000pub-queue: ZSAM.TO.NTFM #推送给NTFM的对列名FCTI-queue: FCTI.TO.ZSAMzsfz:host: 10.1.16.103port: 1416queue-manager: qm_ntfm_zsfzchannel: ch_serviceccsid: 1381username: 1password: 1receive-timeout: 2000pub-queue: ZSFZ.TO.NTFMFCTI-queue: FCTI.TO.ZSFZ

IBM MQ配置类

package com.pantech.ntfmadapter.config;import com.ibm.mq.jms.MQQueueConnectionFactory;
import com.ibm.msg.client.wmq.common.CommonConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter;
import org.springframework.jms.core.JmsOperations;
import org.springframework.jms.core.JmsTemplate;/*** @author lkh* @date 2021/6/28 15:48* @description IBM MQ配置*/
@Configuration
@Slf4j
public class IbmMqConfig {/*** 配置连接工厂*/@Bean(name = "zsamMqQueueConnectionFactory")public MQQueueConnectionFactory zsamMqQueueConnectionFactory(@Value("${spring.ibmmq.zsam.host}") String host,@Value("${spring.ibmmq.zsam.ccsid}") Integer ccsid,@Value("${spring.ibmmq.zsam.channel}") String channel,@Value("${spring.ibmmq.zsam.port}") Integer port,@Value("${spring.ibmmq.zsam.queue-manager}") String queueManager) {MQQueueConnectionFactory mqQueueConnectionFactory = new MQQueueConnectionFactory();mqQueueConnectionFactory.setHostName(host);try {mqQueueConnectionFactory.setTransportType(CommonConstants.WMQ_CM_CLIENT);mqQueueConnectionFactory.setCCSID(ccsid);mqQueueConnectionFactory.setChannel(channel);mqQueueConnectionFactory.setPort(port);mqQueueConnectionFactory.setQueueManager(queueManager);} catch (Exception e) {log.error("创建ZSAM机场IBM MQ连接工厂异常,原因:", e);}return mqQueueConnectionFactory;}/*** 配置连接认证* 如不需要账户密码链接可以跳过此步，直接将MqQueueConnectionFactory注入下一步的缓存连接工厂*/@Bean(name = "zsamUserCredentialsConnectionFactoryAdapter")UserCredentialsConnectionFactoryAdapter zsamUserCredentialsConnectionFactoryAdapter(@Value("${spring.ibmmq.zsam.username}") String username,@Value("${spring.ibmmq.zsam.password}") String password,@Qualifier("zsamMqQueueConnectionFactory") MQQueueConnectionFactory mqQueueConnectionFactory) {UserCredentialsConnectionFactoryAdapter userCredentialsConnectionFactoryAdapter = new UserCredentialsConnectionFactoryAdapter();userCredentialsConnectionFactoryAdapter.setUsername(username);userCredentialsConnectionFactoryAdapter.setPassword(password);userCredentialsConnectionFactoryAdapter.setTargetConnectionFactory(mqQueueConnectionFactory);return userCredentialsConnectionFactoryAdapter;}/*** 配置缓存连接工厂:* 不配置该类则每次与MQ交互都需要重新创建连接，大幅降低速度。*/@Bean(name = "zsamCachingConnectionFactory")@Primarypublic CachingConnectionFactory zsamCachingConnectionFactory(@Qualifier("zsamUserCredentialsConnectionFactoryAdapter") UserCredentialsConnectionFactoryAdapter userCredentialsConnectionFactoryAdapter) {CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();cachingConnectionFactory.setTargetConnectionFactory(userCredentialsConnectionFactoryAdapter);cachingConnectionFactory.setSessionCacheSize(500);cachingConnectionFactory.setReconnectOnException(true);return cachingConnectionFactory;}/*** 配置DefaultJmsListenerContainerFactory, 用@JmsListener注解来监听队列消息时候，尤其存在多个监听的时候，通过实例化配置DefaultJmsListenerContainerFactory来控制消息分发* 如果只有一个连接工厂,即可跳过此步骤,因为默认会创建一个监听容器工厂*/@Bean(name = "zsamJmsQueueListenerContainerFactory")@Primarypublic DefaultJmsListenerContainerFactory zsamJmsQueueListenerContainerFactory(@Qualifier("zsamCachingConnectionFactory") CachingConnectionFactory cachingConnectionFactory) {DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();factory.setConnectionFactory(cachingConnectionFactory);//设置连接数factory.setConcurrency("3-10");//重连间隔时间factory.setRecoveryInterval(1000L);return factory;}/*** 配置JMS模板:可以在方法中通过@autowired的方式注入模板* JmsOperations为JmsTemplate的实现接口。* 重要：不设置setReceiveTimeout时，当队列为空，从队列中取出消息的方法将会一直挂起直到队列内有消息* 如果只是接收消息，可以不配置此步*/@Bean(name = "zsamJmsOperations")public JmsOperations zsamJmsOperations(@Value("${spring.ibmmq.zsam.receive-timeout}") Integer receiveTimeout,@Qualifier("zsamCachingConnectionFactory") CachingConnectionFactory cachingConnectionFactory) {JmsTemplate jmsTemplate = new JmsTemplate(cachingConnectionFactory);jmsTemplate.setReceiveTimeout(receiveTimeout);return jmsTemplate;}//************************************************以下为福州机场IBM MQ配置*******************************************************/*** 配置连接工厂*/@Bean(name = "zsfzMqQueueConnectionFactory")public MQQueueConnectionFactory zsfzMqQueueConnectionFactory(@Value("${spring.ibmmq.zsfz.host}") String host,@Value("${spring.ibmmq.zsfz.ccsid}") Integer ccsid,@Value("${spring.ibmmq.zsfz.channel}") String channel,@Value("${spring.ibmmq.zsfz.port}") Integer port,@Value("${spring.ibmmq.zsfz.queue-manager}") String queueManager) {MQQueueConnectionFactory mqQueueConnectionFactory = new MQQueueConnectionFactory();mqQueueConnectionFactory.setHostName(host);try {mqQueueConnectionFactory.setTransportType(CommonConstants.WMQ_CM_CLIENT);mqQueueConnectionFactory.setCCSID(ccsid);mqQueueConnectionFactory.setChannel(channel);mqQueueConnectionFactory.setPort(port);mqQueueConnectionFactory.setQueueManager(queueManager);} catch (Exception e) {log.error("创建ZSFZ机场IBM MQ连接工厂异常,原因:", e);}return mqQueueConnectionFactory;}/*** 配置连接认证* 如不需要账户密码链接可以跳过此步，直接将MqQueueConnectionFactory注入下一步的缓存连接工厂*/@Bean(name = "zsfzUserCredentialsConnectionFactoryAdapter")UserCredentialsConnectionFactoryAdapter zsfzUserCredentialsConnectionFactoryAdapter(@Value("${spring.ibmmq.zsfz.username}") String username,@Value("${spring.ibmmq.zsfz.password}") String password,@Qualifier("zsfzMqQueueConnectionFactory") MQQueueConnectionFactory mqQueueConnectionFactory) {UserCredentialsConnectionFactoryAdapter userCredentialsConnectionFactoryAdapter = new UserCredentialsConnectionFactoryAdapter();userCredentialsConnectionFactoryAdapter.setUsername(username);userCredentialsConnectionFactoryAdapter.setPassword(password);userCredentialsConnectionFactoryAdapter.setTargetConnectionFactory(mqQueueConnectionFactory);return userCredentialsConnectionFactoryAdapter;}/*** 配置缓存连接工厂:* 不配置该类则每次与MQ交互都需要重新创建连接，大幅降低速度。*/@Bean(name = "zsfzCachingConnectionFactory")public CachingConnectionFactory zsfzCachingConnectionFactory(@Qualifier("zsfzUserCredentialsConnectionFactoryAdapter") UserCredentialsConnectionFactoryAdapter userCredentialsConnectionFactoryAdapter) {CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();cachingConnectionFactory.setTargetConnectionFactory(userCredentialsConnectionFactoryAdapter);cachingConnectionFactory.setSessionCacheSize(500);cachingConnectionFactory.setReconnectOnException(true);return cachingConnectionFactory;}/*** 配置DefaultJmsListenerContainerFactory, 用@JmsListener注解来监听队列消息时候，尤其存在多个监听的时候，通过实例化配置DefaultJmsListenerContainerFactory来控制消息分发* 如果只有一个连接工厂,即可跳过此步骤,因为默认会创建一个监听容器工厂*/@Bean(name = "zsfzJmsQueueListenerContainerFactory")public DefaultJmsListenerContainerFactory zsfzJmsQueueListenerContainerFactory(@Qualifier("zsfzCachingConnectionFactory") CachingConnectionFactory cachingConnectionFactory) {DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();factory.setConnectionFactory(cachingConnectionFactory);//设置连接数factory.setConcurrency("3-10");//重连间隔时间factory.setRecoveryInterval(1000L);return factory;}/*** 配置JMS模板:可以在方法中通过autowired的方式注入模板* JmsOperations为JmsTemplate的实现接口。* 重要：不设置setReceiveTimeout时，当队列为空，从队列中取出消息的方法将会一直挂起直到队列内有消息* 如果只是接收消息，可以不配置此步*/@Bean(name = "zsfzJmsOperations")public JmsOperations zsfzJmsOperations(@Value("${spring.ibmmq.zsfz.receive-timeout}") Integer receiveTimeout,@Qualifier("zsfzCachingConnectionFactory") CachingConnectionFactory cachingConnectionFactory) {JmsTemplate jmsTemplate = new JmsTemplate(cachingConnectionFactory);jmsTemplate.setReceiveTimeout(receiveTimeout);return jmsTemplate;}
}

监听IBM MQ上的队列消息

package com.pantech.ntfmadapter.consumer;import com.pantech.ntfmadapter.service.HandlerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;import javax.jms.Message;
import javax.jms.TextMessage;/*** @author lkh* @date 2021/6/28 15:47* @description 厦门机场IMB消费者*/
@Component
@Slf4j
public class ZsamIbmMqConsumer extends MessageListenerAdapter {private final HandlerService handlerService;@Autowiredpublic ZsamIbmMqConsumer(HandlerService handlerService) {this.handlerService = handlerService;}@Override// 可以配置多个@JmsListener以监听多个队列@JmsListener(destination = "${spring.ibmmq.zsam.FCTI-queue}")@JmsListener(destination = "${spring.ibmmq.zsfz.FCTI-queue}", containerFactory = "zsfzJmsQueueListenerContainerFactory")public void onMessage(Message message) {try {//必须转换如果不转换直接message.tostring消息的传输有限制。TextMessage textMessage = (TextMessage) message;String messagebody = textMessage.getText();log.info("监听到NTFM系统厦门机场消息:{}", messagebody);handlerService.ntfmHandler(messagebody);} catch (Exception e) {log.error("{}消息获取发生异常,原因:", message, e);}}
}

发送消息到IBM MQ

package com.pantech.ntfmadapter.provider;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.core.JmsOperations;
import org.springframework.stereotype.Component;/*** @author lkh* @date 2021/6/29 15:45* @description RabbitMQ生产者*/
@Component
@Slf4j
public class Provider {private final RabbitTemplate rabbitTemplate;private final JmsOperations zsamJmsOperations;private final JmsOperations zsfzJmsOperations;@Value("${spring.ibmmq.zsam.pub-queue}")private String pubQueueName;// 根据类名称注入到spring容器，所以这里的zsamJmsOperations等必须得与配置类中的bean命名一样@Autowiredpublic Provider(RabbitTemplate rabbitTemplate, JmsOperations zsamJmsOperations, JmsOperations zsfzJmsOperations) {this.rabbitTemplate = rabbitTemplate;this.zsamJmsOperations = zsamJmsOperations;this.zsfzJmsOperations = zsfzJmsOperations;}public void sendToImbMq(String messageBody){zsamJmsOperations.convertAndSend(pubQueueName,messageBody);log.info("推送至IBM MQ成功,原消息为:{}",messageBody);}public void sendToRabbitMq(String messageBody){Message message = new Message(messageBody.getBytes());rabbitTemplate.convertAndSend("",message);log.info("转发至RabbitMQ成功,原消息为:{}",messageBody);}
}

springboot配置监听单个队列管理器

大体与上面配置多个队列管理器类似，这里贴一下配置类代码

@Configuration
public class IbmMqConfig {@Value("${spring.ibmmq.host}")private String host;@Value("${spring.ibmmq.port}")private Integer port;@Value("${spring.ibmmq.queue-manager}")private String queueManager;@Value("${spring.ibmmq.channel}")private String channel;@Value("${spring.ibmmq.username}")private String username;@Value("${spring.ibmmq.password}")private String password;@Value("${spring.ibmmq.receive-timeout}")private long receiveTimeout;/*** 配置连接工厂:* CCSID要与连接到的队列管理器一致，Windows下默认为1381，* Linux下默认为1208。1208表示UTF-8字符集，建议把队列管理器的CCSID改为1208*/@Beanpublic MQQueueConnectionFactory mqQueueConnectionFactory() {MQQueueConnectionFactory mqQueueConnectionFactory = new MQQueueConnectionFactory();mqQueueConnectionFactory.setHostName(host);try {mqQueueConnectionFactory.setTransportType(CommonConstants.WMQ_CM_CLIENT);mqQueueConnectionFactory.setCCSID(1381);mqQueueConnectionFactory.setChannel(channel);mqQueueConnectionFactory.setPort(port);mqQueueConnectionFactory.setQueueManager(queueManager);} catch (Exception e) {e.printStackTrace();}return mqQueueConnectionFactory;}/*** 配置连接认证:* 如不需要账户密码链接可以跳过此步，直接将mqQueueConnectionFactory注入下一步的缓存连接工厂。*/
//    @Bean
//    UserCredentialsConnectionFactoryAdapter userCredentialsConnectionFactoryAdapter(MQQueueConnectionFactory mqQueueConnectionFactory) {//        UserCredentialsConnectionFactoryAdapter userCredentialsConnectionFactoryAdapter = new UserCredentialsConnectionFactoryAdapter();
//        userCredentialsConnectionFactoryAdapter.setUsername(username);
//        userCredentialsConnectionFactoryAdapter.setPassword(password);
//        userCredentialsConnectionFactoryAdapter.setTargetConnectionFactory(mqQueueConnectionFactory);
//        return userCredentialsConnectionFactoryAdapter;
//    }/*** 配置缓存连接工厂:* 不配置该类则每次与MQ交互都需要重新创建连接，大幅降低速度。*/
//    @Bean
//    @Primary
//    public CachingConnectionFactory cachingConnectionFactory(UserCredentialsConnectionFactoryAdapter userCredentialsConnectionFactoryAdapter) {//        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
//        cachingConnectionFactory.setTargetConnectionFactory(userCredentialsConnectionFactoryAdapter);
//        cachingConnectionFactory.setSessionCacheSize(500);
//        cachingConnectionFactory.setReconnectOnException(true);
//        return cachingConnectionFactory;
//    }@Bean@Primarypublic CachingConnectionFactory cachingConnectionFactory(MQQueueConnectionFactory mqQueueConnectionFactory) {CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();cachingConnectionFactory.setTargetConnectionFactory(mqQueueConnectionFactory);cachingConnectionFactory.setSessionCacheSize(500);cachingConnectionFactory.setReconnectOnException(true);return cachingConnectionFactory;}/*** 配置事务管理器:* 不使用事务可以跳过该步骤。* 如需使用事务，可添加注解@EnableTransactionManagement到程序入口类中，事务的具体用法可参考Spring Trasaction。*/@Beanpublic PlatformTransactionManager jmsTransactionManager(CachingConnectionFactory cachingConnectionFactory) {JmsTransactionManager jmsTransactionManager = new JmsTransactionManager();jmsTransactionManager.setConnectionFactory(cachingConnectionFactory);return jmsTransactionManager;}/*** 配置JMS模板:* JmsOperations为JmsTemplate的实现接口。* 重要：不设置setReceiveTimeout时，当队列为空，从队列中取出消息的方法将会一直挂起直到队列内有消息*/@Beanpublic JmsOperations jmsOperations(CachingConnectionFactory cachingConnectionFactory) {JmsTemplate jmsTemplate = new JmsTemplate(cachingConnectionFactory);jmsTemplate.setReceiveTimeout(receiveTimeout);return jmsTemplate;}

我待你好，不要骂我。