java - 将消息从 MQ 发送到 EMS(JMS) 队列

Question

我想将MQ队列中的消息发送到不同服务器上的EMS队列。我不知道如何使用 Java 来做到这一点。如何保证发送消息时不丢失消息。

我能够使用 java 来使用来自 MQ 的消息。

try {
    // Create a connection factory
    JmsFactoryFactory ff = JmsFactoryFactory.getInstance(WMQConstants.WMQ_PROVIDER);
    JmsConnectionFactory cf = ff.createConnectionFactory();

    // Set the properties
    cf.setStringProperty(WMQConstants.WMQ_HOST_NAME, HOST);
    cf.setIntProperty(WMQConstants.WMQ_PORT, PORT);
    cf.setStringProperty(WMQConstants.WMQ_CHANNEL, CHANNEL);
    cf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT);
    cf.setStringProperty(WMQConstants.WMQ_QUEUE_MANAGER, QMGR);
    cf.setStringProperty(WMQConstants.WMQ_APPLICATIONNAME, "JmsPutGet (JMS)");
    cf.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, true);
    cf.setStringProperty(WMQConstants.USERID, APP_USER);
    cf.setStringProperty(WMQConstants.PASSWORD, APP_PASSWORD);

    // Create JMS objects
    context = cf.createContext();
    destination = context.createQueue("queue:///" + QUEUE_NAME);

    long uniqueNumber = System.currentTimeMillis() % 1000;
    TextMessage message = context.createTextMessage("Your lucky number today is " + uniqueNumber);

    producer = context.createProducer();
    producer.send(destination, message);
    System.out.println("Sent message:\n" + message);

    consumer = context.createConsumer(destination); // autoclosable
    String receivedMessage = consumer.receiveBody(String.class, 15000); // in ms or 15 seconds

    System.out.println("\nReceived message:\n" + receivedMessage);

    recordSuccess();
} catch (JMSException jmsex) {
    recordFailure(jmsex);
}

System.exit(status);

Answer 1

您的代码几乎是正确的，但如果您想在程序中切换 JMS 提供程序，您还必须切换 JMS ConnectionFactory。这是用于创建消费者和消息的中央(特定于提供者)类。

因此，在创建上下文、生产者和消息时，您需要使用 TIBCO 的 com.tibco.tibjms.naming.TibjmsInitialContextFactory，例如如下所示:

try {

  // 1) Create a MQ connection factory
  JmsFactoryFactory ff = JmsFactoryFactory.getInstance(WMQConstants.WMQ_PROVIDER);
  JmsConnectionFactory cf = ff.createConnectionFactory();

  // 2) Set the properties
  cf.setStringProperty(WMQConstants.WMQ_HOST_NAME, HOST);
  cf.setIntProperty(WMQConstants.WMQ_PORT, PORT);
  cf.setStringProperty(WMQConstants.WMQ_CHANNEL, CHANNEL);
  cf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT);
  cf.setStringProperty(WMQConstants.WMQ_QUEUE_MANAGER, QMGR);
  cf.setStringProperty(WMQConstants.WMQ_APPLICATIONNAME, "JmsPutGet (JMS)");
  cf.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, true);
  cf.setStringProperty(WMQConstants.USERID, APP_USER);
  cf.setStringProperty(WMQConstants.PASSWORD, APP_PASSWORD);

  // 3) Create MQ consumer
  JMSContext mqContext = cf.createContext();
  destination = mqContext.createQueue("queue:///" + QUEUE_NAME);
  consumer = mqContext.createConsumer(destination); // autoclosable

  // 4)  wait for message from MQ (or null)
  String receivedText = mqContext.receiveBody(String.class, 15000); // in ms or 15 seconds
  System.out.println("\nReceived message:\n" + receivedText);

  // 5) Create TIBCO EMS ConnectionFactory and an EMS MessageProducer
  TibjmsConnectionFactory emsCF = new com.tibco.tibjms.TibjmsConnectionFactory( "tcp://1.2.3.4:7222");
  Connection emsConnection = emsCF.createConnection(user, password);
  Session emsSession = emsConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  MessageProducer emsProducer = emsSession.createProducer( emsSession.createQueue(QUEUE_NAME) );

  // 6) Create EMS TextMessage from MQ TextMessage
  TextMessage emsMsg = emsSession.createTextMessage( receivedText );

  // 7) publish to EMS
  emsProducer.send(emsMsg);

  // 8) cleanup
  emsConnection.close();

  recordSuccess();
} catch (JMSException jmsex) {
   recordFailure(jmsex);
}