🧒🏽 ➿ 🚌 使用Spring的多连接IBM MQ ☂️ 📕 👩🏻‍🏭

我将给出一个示例，说明如何配置多个端点以连接到IBM MQ。

目标：

从多个名称相同但位于不同主机/队列管理器上的队列中读取
向随机定义的节点写入响应

0.假设您已经部署了MQ，或者正在使用其他人的MQ。

1.我们将依赖项加载到项目中：

专家

<dependency>
    <groupId>com.ibm.mq</groupId>
    <artifactId>mq-jms-spring-boot-starter</artifactId>
    <version>2.3.3</version>
</dependency>

摇动

compile group: 'com.ibm.mq', name: 'mq-jms-spring-boot-starter', version: '2.3.3'

2.创建一个配置，输入您的点的连接参数（您已经创建了它们吗？）。我们使用数组，因此可以有任意多个连接。

mq:
  servers:
    - queueManager: QM1
      channel: DEV.ADMIN.SVRCONN
      connName: ibmmq.ru(1414)
      user: admin
      password: passw0rd
    - queueManager: QM2
      channel: DEV.ADMIN.SVRCONN
      connName: ibmmq.ru(1415)
      user: admin
      password: passw0rd
  queue1: QUEUE1
  queue2: QUEUE2

3.创建用于读取这些属性的类：

import lombok.Data;
import lombok.EqualsAndHashCode;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConfigurationProperties(prefix = "mq")
@EqualsAndHashCode(callSuper = false)
@Data
public class MqConfig {
    private List<ConnectionConfiguration> servers;
    private String queue1;
    private String queue2;

}

import lombok.Data;
import lombok.EqualsAndHashCode;

@Data
@EqualsAndHashCode(callSuper = false)
public class ConnectionConfiguration {
    String queueManager;
    String channel;
    String connName;
    String user;
    String password;
}

4.创建一个监听器：

import javax.jms.MessageListener;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MqListener implements MessageListener {

    @SneakyThrows
    @Override
    public void onMessage(@Payload javax.jms.Message message) {
        log.info("  <" + message + ">");
        //TODO:      
    }

5.配置！我们从yml-properties为数组的每个元素定义connectionFactors。我们创建了一个用于发送消息的模板表，我们将创建的连接馈送到模板的输入中。我们创建了侦听器工厂，在它们的输入处，我们还使用了创建的connectionFactories。


import com.fasterxml.jackson.databind.ObjectMapper;
import com.ibm.mq.jms.MQConnectionFactory;
import com.ibm.msg.client.wmq.WMQConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.*;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.QosSettings;
import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.MessageType;
import org.springframework.jms.support.converter.SimpleMessageConverter;

import javax.jms.*;
import java.util.*;

import static javax.jms.DeliveryMode.NON_PERSISTENT;
import static javax.jms.Session.CLIENT_ACKNOWLEDGE;

@Configuration
@EnableJms
@Slf4j
public class MqConfiguration {

    @Autowired
    MqConfig mqConfig;

    @Autowired
    private JmsListenerEndpointRegistry registry;

//  ,       connectionFactories
    @Bean
    public List<JmsListenerContainerFactory> myFactories(
            @Qualifier("myConnFactories") 
            List<CachingConnectionFactory> connectionFactories,
            MqListener mqListener) {
        List<JmsListenerContainerFactory> factories = new ArrayList<>();
        connectionFactories.forEach(connectionFactory -> {
            DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setSessionAcknowledgeMode(CLIENT_ACKNOWLEDGE);

            QosSettings qosSettings = new QosSettings();
            qosSettings.setDeliveryMode(NON_PERSISTENT);
            factory.setReplyQosSettings(qosSettings);

            SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
            endpoint.setId("myJmsEndpoint-"+ UUID.randomUUID());
            endpoint.setDestination(mqConfig.getQueue1());
            endpoint.setMessageListener(mqListener);
            registry.registerListenerContainer(endpoint, factory);

            factories.add(factory);
        });
        return factories;
    }
//     ,      
    @Bean
    @Qualifier("myJmsTemplates")
    public List<JmsTemplate> jmsTemplates(
            @Qualifier("myConnFactories") 
            List<CachingConnectionFactory> connectionFactories) {
        return getJmsTemplates(new ArrayList<ConnectionFactory>(connectionFactories));
    }

    public List<JmsTemplate> getJmsTemplates(List<ConnectionFactory> connectionFactories) {
        List<JmsTemplate> jmsTemplates = new ArrayList<>();
        for (ConnectionFactory connectionFactory : connectionFactories) {
            JmsTemplate jmsTemplate = new JmsTemplate();
            jmsTemplate.setConnectionFactory(connectionFactory);
            jmsTemplate.setMessageConverter(new SimpleMessageConverter());
            jmsTemplate.setDefaultDestinationName(mqConfig.getQueue2());
            jmsTemplate.setDeliveryMode(NON_PERSISTENT);
            jmsTemplate.setDeliveryPersistent(false);
            jmsTemplate.setExplicitQosEnabled(true);
            jmsTemplates.add(jmsTemplate);
        }
        return jmsTemplates;
    }

//       yml-
    @Bean
    @Qualifier("myConnFactories")
    public List<CachingConnectionFactory> connectionFactories() throws JMSException {
        List<CachingConnectionFactory> factories = new ArrayList<>();

        for (ConnectionConfiguration server : mqConfig.getServers()) {
            CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
            MQConnectionFactory cf = new MQConnectionFactory();
            cachingConnectionFactory.setTargetConnectionFactory(cf);
            cf.setQueueManager(server.getQueueManager());
            cf.setChannel(server.getChannel());
            cf.setConnectionNameList(server.getConnName());
            cf.setStringProperty(WMQConstants.USERID, server.getUser());
            cf.setStringProperty(WMQConstants.PASSWORD, server.getPassword());
            cf.setStringProperty("XMSC_WMQ_CONNECTION_MODE", "1");

            factories.add(cachingConnectionFactory);
        }
        return factories;
    }

}

endpoint.setMessageListener（mqListener）;

在这里，我们指示侦听器（在步骤4中创建），以确定在接收消息时的操作。

6.让我们创建一个服务层，假设在发送响应之后会有一些逻辑。

import javax.jms.TextMessage;

public interface MqService {

    void sendToMq(TextMessage msg);

}

import javax.jms.TextMessage;
import org.springframework.jms.JmsException;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class MqServiceImpl implements MqService {

    @Autowired
    private MqConfig mqConfig;

    @Autowired
    @Qualifier("myJmsTemplates")
    List<JmsTemplate> jmsTemplates;


    @Override
    public void sendToMq(TextMessage msg ) {
        //- 
        //     /  .
        int maxIndex = jmsTemplates.size()-1; //    - ""
        int randomNumber = (int) Math.round(Math.random() * maxIndex);
        jmsTemplates.get(randomNumber).convertAndSend(mqConfig.getQueue2(), msg);
    }
}

7.添加向侦听器发送响应：

    @Autowired
    MqService mqService;

    @SneakyThrows
    @Override
    public void onMessage(@Payload javax.jms.Message message) {
        log.info("  <" + message + ">");
        mqService.sentToMq((TextMessage) message);
    }

瞧，您完成了，可以检查。

使用Spring的多连接IBM MQ

More articles: