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);