JMS MessageConsumer Arayüzü

Giriş

Destination olarak Queue, TemporaryQueue, TemporaryTopic, Topic kullanılır. JMS 1.1 ile kullanılır.

constructor 

Örnek - Destination
Queue yoksa hem Queue yaratmak hem de okumak için şöyle yaparız.

Destination destination = ...;
Session session = ...;
MessageConsumer messageConsumer = session.createConsumer(destination);

Örnek - Queue
Şöyle yaparız

Queue replyTo = session.createTemporaryQueue();
MessageConsumer consumer = session.createConsumer(replyTo);

Örnek - Topic
Şöyle yaparız.

private void createUnsharedConsumer(ConnectionFactory connectionFactory, Topic topic)
  throws JMSException {

   Connection connection = connectionFactory.createConnection();

   Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

   MessageConsumer messageConsumer = session.createConsumer(topic);

   connection.start();

   Message message = messageConsumer.receive(10000);

   while (message != null) {
      System.out.println("Message received: " + ((TextMessage) message).getText());
      message = messageConsumer.receive(10000);
   }
   connection.close();
}

receive metodu

Örnek

Şöyle yaparız

ConnectionFactory connectionFactory = ...
String QUEUE_NAME = "user_creation_queue";

void pollUserFromQueue() throws Exception {
  Session session = ...;
  Destination destination = session.createQueue(QUEUE_NAME);
  MessageConsumer consumer = session.createConsumer(destination);
  Message message = consumer.receive();

  ObjectMessage objMessage = (ObjectMessage) message;
  User user = (User) objMessage.getObject();
  ...  
}

receive metodu - timeout

Örnek
Şöyle yaparız

Message message = consumer.receive(120000);
if (message instanceof TextMessage) {
  TextMessage textMessage = (TextMessage) message;
  text = textMessage.getText();
  System.out.println("MESSAGE RECEIVED " + System.currentTimeMillis());
}

Örnek
Elimizde şöyle bir kod olsun.

class MyConsumer extends Thread {
  private final Connection connection;
  private final Destination destination;

  MyConsumer(Connection connection, Destination destination) {
    this.connection = connection;
    this.destination = destination;
  }

  @Override
  public void run() {
    try (Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE)) {
      MessageConsumer messageConsumer = session.createConsumer(destination);
      connection.start();
      Message message = messageConsumer.receive(5000);
      if (message == null) {
        System.out.println("Did not receive message within the allotted time.");
        return;
      }
      System.out.println("Received message: " + message);
    } catch (Throwable e) {
      e.printStackTrace();
      return;
    }
  }
}

Şöyle yaparız

InitialContext initialContext = new InitialContext();
Queue queue = (Queue) initialContext.lookup("queue/exampleQueue");
ConnectionFactory cf = (ConnectionFactory) initialContext.lookup("ConnectionFactory");
Connection connection = cf.createConnection();
Thread myConsumer = myMultiThreadedApp.runConsumer(connection, queue);




JMS Connection Arayüzü

start metodu
Açıklaması şöyle

Also, keep in mind that there's no reason to call start() on the JMS connection if it's just sending a message. The start() method only impacts consumers.

Örnek

Şöyle yaparız. Burada bir MessageProducer ve MessageConsumer yaratılıyor.. MessageProducer.send(TextMessage) çağrısı ile mesaj gönderiliyor.

MessageConsumer.receive() çağrısı ile mesaj alınıyor  

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.qpid.jms.JmsConnectionFactory;

public static void main(String[] args) throws Exception {
  Connection connection = null;
  ConnectionFactory connectionFactory = new JmsConnectionFactory("amqp://localhost:5672");
  try {
    // Step 1. Create an amqp qpid 1.0 connection
    connection = connectionFactory.createConnection();
    // Step 2. Create a session
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    // Step 3. Create a sender
    Queue queue = session.createQueue("jms.queue.exampleQueue");
    MessageProducer sender = session.createProducer(queue);
    // Step 4. send a few simple message
    sender.send(session.createTextMessage("Hello world "));
    connection.start();
    // Step 5. create a moving receiver, this means the message will be removed froma
    // the queue
    MessageConsumer consumer = session.createConsumer(queue);
    // Step 7. receive the simple message
    TextMessage m = (TextMessage) consumer.receive(5000);
    System.out.println("message = " + m.getText());
  }
  finally {
    if (connection != null) {
      // Step 9. close the connection
      connection.close();
    }
  }
}

createSession metodu - boolean transacted, int acknowledgeMode
Şöyle yaparız

try {
  if (connection == null) {
    createConnection();
  }
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

  // Destination represents here our queue 'MessageProducerJMSV1' on the  JMS server
  Destination destination = session.createQueue(subject);

  MessageProducer producer = session.createProducer(destination);

  //Sending message to the queue
  TextMessage toSendMessage = session.createTextMessage(message);

  long delay = 300 * 1000;

  toSendMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);

  producer.send(toSendMessage);
} catch (Throwable th) {
  th.printStackTrace();
               
  // if there are any problems close the connection and it will be re-created next time
  if (connection != null) {
    connection.close();
    connection = null;
  }
}
  
private void createConnection() {
  // Getting JMS connection from the server
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
  connection = connectionFactory.createConnection();
}