It looks like you've hit flow control - you might want to disable it -
see here: http://activemq.apache.org/producer-flow-control.html
cheers,
Rob
On Jan 18, 2008, at 6:59 AM, Olivier Bigard wrote:
Hello,
We had the same problem with ActiveMQ 5.0.0.
Yesterday we downloaded the last 5.1 SNAPSHOT and the message
disappeared,
but ActiveMQ is still frozen after some messages...
I put a message in the forum with our Java code.
We are running ActiveMQ on Solaris platform with JDK 1.5.0_07.
Can anyone help us?
Thank you
Juergen.Schumacher wrote:
Hello,
ok, attaching the source code file does apparently not work here.
So here are the relevant code parts from the producer and processor
classes,
which create the connections to ActiveMQ and the messages.
Sorry for the trouble...
Regards,
Jürgen Schumacher
----------------
Procucer.java
private void initializeBrokerConnection() throws JMSException {
LOG.info("Connecting to " + brokerUrl);
ActiveMQConnectionFactory connectionFactory =
new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, brokerUrl);
connection = connectionFactory.createQueueConnection();
connection.start();
session = connection.createQueueSession(true,
Session.CLIENT_ACKNOWLEDGE);
destination = session.createQueue("queue1");
producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
}
private void startIteration() throws Exception {
for (int i = 0; i < numberOfMessages; i++) {
LOG.info("Creating message " + i);
TextMessage message = session.createTextMessage("Message #" +
i);
LOG.info("Sending message " + i);
producer.send(message);
session.commit();
producedCount++;
LOG.info("Produced " + producedCount + " messages.");
}
}
---------------
Processor.java
private void initializeBrokerConnection() throws JMSException {
LOG.info("Connecting to " + brokerUrl);
ActiveMQConnectionFactory connectionFactory =
new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, brokerUrl);
sourceConnection = connectionFactory.createQueueConnection();
destinationConnection = connectionFactory.createQueueConnection();
sourceConnection.start();
destinationConnection.start();
sourceSession = sourceConnection.createQueueSession(true,
Session.CLIENT_ACKNOWLEDGE);
destinationSession =
destinationConnection.createQueueSession(true,
Session.CLIENT_ACKNOWLEDGE);
source = sourceSession.createQueue("queue1");
consumer = sourceSession.createConsumer(source);
destination = destinationSession.createQueue("queue2");
producer = destinationSession.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
}
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
String id = ((TextMessage) message).getText();
LOG.info("Processing id = " + id);
MapMessage map = destinationSession.createMapMessage();
map.setString("id", id);
map.setString("text", messageText);
LOG.info("Send message");
producer.send(map);
LOG.info("Acknowledge message");
message.acknowledge();
LOG.info("Commit source");
sourceSession.commit();
LOG.info("Commit destination");
destinationSession.commit();
producedCount++;
LOG.info("Produced " + producedCount + " messages.");
} catch (Exception ex) {
LOG.error("Error during processing", ex);
}
}
}
--
View this message in context:
http://www.nabble.com/High-message-frequency-causes-ActiveMQ-to-freeze-tp14919292s2354p14947393.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
