Objectif:
- lire à partir de plusieurs files d'attente nommées de la même manière, mais situées sur des hôtes / gestionnaires de files d'attente différents
- écrire une réponse à un nœud défini aléatoirement
0. Supposons que vous ayez déjà déployé MQ ou que vous utilisiez celui de quelqu'un d'autre.
1. Nous chargeons les dépendances dans le projet:
Maven
<dependency>
<groupId>com.ibm.mq</groupId>
<artifactId>mq-jms-spring-boot-starter</artifactId>
<version>2.3.3</version>
</dependency>
gradle
compile group: 'com.ibm.mq', name: 'mq-jms-spring-boot-starter', version: '2.3.3'
2. Créez une configuration, entrez les paramètres de connexion de vos points (vous les avez déjà créés?). Nous utilisons un tableau, donc il peut y avoir autant de connexions que vous le souhaitez.
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. Créez des classes pour lire ces propriétés:
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. Créez un auditeur:
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. Configuration! Nous déterminons les connectionFactors pour chaque élément du tableau à partir de yml-properties. Nous créons une feuille de modèles pour l'envoi de messages, à l'entrée desquels nous alimentons les connexions créées. Nous créons des usines d'écoute, à l'entrée desquelles nous utilisons également les connectionFactories créées.
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);Ici, nous indiquons l'auditeur (qui a été créé à l'étape 4) pour déterminer les actions lors de la réception d'un message.
6. Créons une couche de service, où disons qu'il y aura une certaine logique et après l'envoi d'une réponse.
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. Ajoutez l'envoi de la réponse à l'écouteur:
@Autowired
MqService mqService;
@SneakyThrows
@Override
public void onMessage(@Payload javax.jms.Message message) {
log.info(" <" + message + ">");
mqService.sentToMq((TextMessage) message);
}
Voila, vous avez terminé, vous pouvez vérifier.